0.8.0: Added support for channels with 16, 24 and 32bits

pyartnet/__init__.py

@@ -2,8 +2,8 @@
 from . import errors
 from . import fades
 
-from .dmx_channel   import DmxChannel
+from .dmx_channel   import DmxChannel, DmxChannel16Bit, DmxChannel24Bit, DmxChannel32Bit
 from .dmx_universe  import DmxUniverse
 from .artnet_node   import ArtNetNode
 
-from . import output_correction
+from . import output_correction

pyartnet/__version__.py

@@ -1 +1 @@
-__version__ = '0.7.1'
+__version__ = '0.8.0'

pyartnet/dmx_channel.py

@@ -5,21 +5,36 @@
 
 import pyartnet
 
+
 log = logging.getLogger('pyartnet.DmxChannel')
 
 
 class DmxChannel:
-    def __init__(self, universe, start: int, width: int):
-        assert 1 <= start <= 512
-        assert width > 0
-        assert start + width - 1 <= 512
+    _CHANNEL_SIZE: int = 1   # Channel size in byte
+
+    def __init__(self, universe: 'pyartnet.DmxUniverse', start: int, width: int):
+        self.width: int = width
+        byte_width: int = width * self._CHANNEL_SIZE
+
+        self.start: int = start
+        self.stop: int = start + byte_width - 1
 
-        self.start = start
-        self.width = width
+        if self.start < 1 or self.start > 512:
+            raise pyartnet.errors.ChannelOutOfUniverseError(
+                f'Start position of channel out of universe (1..512): {self.start}')
 
-        self.__val_act_i = [0 for k in range(width)]
+        if width <= 0 or not isinstance(width, int):
+            raise pyartnet.errors.ChannelWidthInvalid(f'Channel width must be int > 0: {width} ({type(width)})')
 
-        self.__fades: typing.List[typing.Optional[pyartnet.fades.FadeBase]] = [None for k in range(width)]
+        if self.stop > 512:
+            raise pyartnet.errors.ChannelOutOfUniverseError(
+                f'End position of channel out of universe (1..512): '
+                f'start: {self.start} width: {self.width} * {byte_width}bytes -> {self.stop}'
+            )
+
+        self.__val_act_i = [0 for _ in range(self.width)]
+
+        self.__fades: typing.List[typing.Optional[pyartnet.fades.FadeBase]] = [None for k in range(self.width)]
         self.__fade_running = False
 
         self.__step_max = 0
@@ -37,10 +52,10 @@ def __init__(self, universe, start: int, width: int):
 
     def __apply_output_correction(self, channel_val):
         if self.output_correction is not None:
-            return self.output_correction(channel_val)
+            return self.output_correction(channel_val, 255 ** self._CHANNEL_SIZE)
 
         if self.__universe.output_correction is not None:
-            return self.__universe.output_correction(channel_val)
+            return self.__universe.output_correction(channel_val, 255 ** self._CHANNEL_SIZE)
 
         return channel_val
 
@@ -51,6 +66,15 @@ def fade_running(self) -> bool:
     def get_channel_values(self) -> typing.List[int]:
         return self.__val_act_i.copy()
 
+    def get_bytes(self) -> typing.Iterable:
+        for obj in self.__val_act_i:
+            if self._CHANNEL_SIZE == 1:
+                yield obj
+            else:
+                for i in range(self._CHANNEL_SIZE, 0, -1):
+                    val = (obj >> 8 * (i - 1)) & 0xFF
+                    yield val
+
     def add_fade(self, fade_list: typing.List[int], duration_ms: int, fade_class=pyartnet.fades.LinearFade):
         fade_list = fade_list[:]
         assert isinstance(fade_list, list)
@@ -64,7 +88,7 @@ def add_fade(self, fade_list: typing.List[int], duration_ms: int, fade_class=pya
 
             assert isinstance(k, pyartnet.fades.FadeBase), type(k)
             assert isinstance(k.val_target, int)
-            assert 0 <= k.val_target <= 255
+            assert 0 <= k.val_target <= 255 ** self._CHANNEL_SIZE
 
         # calculate how much steps we will be having
         step_time_ms = self.__universe._artnet_node.sleep_time * 1000
@@ -129,3 +153,15 @@ def process(self):
         self.__step_is += 1
 
         return self.__fade_running
+
+
+class DmxChannel16Bit(DmxChannel):
+    _CHANNEL_SIZE: int = 2   # Channel size in byte
+
+
+class DmxChannel24Bit(DmxChannel):
+    _CHANNEL_SIZE: int = 3   # Channel size in byte
+
+
+class DmxChannel32Bit(DmxChannel):
+    _CHANNEL_SIZE: int = 4   # Channel size in byte

pyartnet/dmx_universe.py

@@ -34,10 +34,12 @@ def get_channel(self, channel_name: str) -> pyartnet.DmxChannel:
         except KeyError:
             raise ChannelNotFoundError(f'Channel "{channel_name}" not found in the universe!') from None
 
-    def add_channel(self, start: int, width: int, channel_name: str = '') -> pyartnet.DmxChannel:
+    def add_channel(self, start: int, width: int, channel_name: str = '',
+                    channel_type: typing.Type[pyartnet.DmxChannel] = pyartnet.DmxChannel) -> pyartnet.DmxChannel:
         assert isinstance(channel_name, str), type(channel_name)
-        chan = pyartnet.DmxChannel(self, start, width)
-        stop = start + width - 1
+        assert issubclass(channel_type, pyartnet.DmxChannel)
+
+        chan = channel_type(self, start, width)
 
         # build name if not supplied
         if not channel_name:
@@ -50,17 +52,15 @@ def add_channel(self, start: int, width: int, channel_name: str = '') -> pyartne
         # Make sure channels are not overlapping because they will overwrite each other
         # and this leads to unintended behavior
         for _n, _c in self.__chans.items():
-            _c_start = _c.start
-            _c_end = _c.start + _c.width
-            if _c_start > stop or _c_end < start:
+            if _c.start > chan.stop or _c.stop < chan.start:
                 continue
-            for i in range(_c_start, _c_end):
-                if start <= i <= stop:
+            for i in range(_c.start, _c.stop + 1):
+                if start <= i <= chan.stop:
                     raise OverlappingChannelError(f'New channel {channel_name} is overlapping with channel {_n:s}!')
 
         # Keep track of highest channel so we can pad
         highest_was = self.highest_channel
-        self.highest_channel = max(self.highest_channel, start + width - 1)
+        self.highest_channel = max(self.highest_channel, chan.stop)
 
         # round channels
         if self.highest_channel % 2:
@@ -89,8 +89,7 @@ def process(self) -> bool:
             running = True
 
             # get new universe Data
-            new_val = c.get_channel_values()
-            for k, val in enumerate(new_val):
+            for k, val in enumerate(c.get_bytes()):
                 self.data[c.start + k - 1] = val
 
         self.__fade_running = running

pyartnet/errors.py

@@ -8,3 +8,11 @@ class ChannelNotFoundError(Exception):
 
 class OverlappingChannelError(Exception):
     pass
+
+
+class ChannelOutOfUniverseError(Exception):
+    pass
+
+
+class ChannelWidthInvalid(Exception):
+    pass

pyartnet/output_correction.py

@@ -1,12 +1,12 @@
 
 
-def quadratic( val):
-    return (val ** 2) / 255
+def quadratic(val: float, max_val: int = 255):
+    return (val ** 2) / max_val
 
 
-def cubic( val):
-    return (val ** 3) / 65_025  # 255^2
+def cubic(val: float, max_val: int = 255):
+    return (val ** 3) / (max_val * max_val)
 
 
-def quadruple( val):
-    return (val ** 4) / 16_581_375  # 255^3
+def quadruple(val: float, max_val: int = 255):
+    return (val ** 4) / (max_val * max_val * max_val)

readme.md

@@ -1,7 +1,9 @@
 # pyartnet
+![Tests](https://github.com/spacemanspiff2007/PyArtNet/workflows/Tests/badge.svg)
 ![PyPI - Python Version](https://img.shields.io/pypi/pyversions/pyartnet)
 [![Downloads](https://pepy.tech/badge/pyartnet/month)](https://pepy.tech/project/pyartnet/month)
 
+
 pyartnet is a python implementation of the ArtNet protocol using [asyncio](https://docs.python.org/3/library/asyncio.html).
 
 # Usage
@@ -13,7 +15,12 @@ from pyartnet import ArtNetNode
 node = ArtNetNode('IP')
 await node.start()
 
+# Create universe 0
 universe = node.add_universe(0)
+
+# Add a channel to the universe which consists of 3 values
+# Default size of a value is 8Bit (0..255) so this would fill
+# the DMX values 1..3 of the universe
 channel  = universe.add_channel(start=1, width=3)
 
 # Fade channel to 255,0,0 in 5s
@@ -78,8 +85,34 @@ linear (default when nothing is set), quadratic, cubic then quadruple
 
 Quadratic or cubic results in much smoother and more pleasant fades when using LED Strips.
 
+## Wider DMX Channels
+The library supports wider dmx channels for advanced control. 
+
+````python
+from pyartnet import ArtNetNode, DmxChannel16Bit
+
+node = ArtNetNode('IP')
+await node.start()
+
+# Create universe 0
+universe = node.add_universe(1)
+
+# Add a channel to the universe which consists of 3 values where each value is 16Bits
+# This would fill the DMX values 1..6 of the universe
+channel  = universe.add_channel(start=1, width=3, channel_type=DmxChannel16Bit)
+
+# Notice the higher maximum value for the fade
+channel.add_fade([0xFFFF,0,0], 5000)   
+````
+
+
 # Changelog
 
+#### 0.8.0 (11.02.2021)
+- Added support for channels with 16, 24 and 32bits 
+
+
+
 #### 0.7.0 (28.10.2020)
 - renamed logger to ``pyartnet`` to make it consistent with the module name
 - callbacks on the channel now get the channel passed in as an argument

tests/test_channel.py

@@ -1,8 +1,10 @@
+import time
+from itertools import zip_longest
 from unittest.mock import Mock
 
 import pytest
-import time
 
+import pyartnet
 from .conftest import PatchedArtNetNode
 
 
@@ -93,3 +95,60 @@ async def test_channel_wait_till_complete(running_artnet_node: PatchedArtNetNode
 
     assert channel.get_channel_values() == [255]
     assert 0.4 <= duration <= 0.6
+
+
+@pytest.mark.asyncio
+async def test_byte_iterator(running_artnet_node: PatchedArtNetNode):
+
+    universe = running_artnet_node.add_universe(0)
+
+    for i, obj in enumerate(([10], [20, 30], [255, 254, 253, 252, 251])):
+        channel = universe.add_channel(5 * i + 1, len(obj))
+        channel.add_fade(obj, 0)
+        await channel.wait_till_fade_complete()
+        assert channel.get_channel_values() == obj
+        for soll, ist in zip_longest(obj, channel.get_bytes()):
+            assert soll == ist
+
+    target_vals = [[254 * 255], [253 * 255, 252 * 255], [251 * 255, 250 * 255, 249 * 250]]
+    target_bytes = []
+    for val in target_vals:
+        b = []
+        for obj in val:
+            b.append(obj >> 8)
+            b.append(obj & 255)
+        target_bytes.append(b)
+
+    for i, obj in enumerate(target_vals):
+        channel = universe.add_channel(10 * i + 50, len(obj), channel_type=pyartnet.DmxChannel16Bit)
+        channel.add_fade(obj, 0)
+        await channel.wait_till_fade_complete()
+        assert channel.get_channel_values() == obj
+        for soll, ist in zip_longest(target_bytes[i], list(channel.get_bytes())):
+            assert soll == ist
+
+
+def test_channel_boundaries():
+    node = pyartnet.ArtNetNode(host='')
+    univ = pyartnet.DmxUniverse(node)
+
+    with pytest.raises(pyartnet.errors.ChannelOutOfUniverseError) as r:
+        pyartnet.DmxChannel(univ, 0, 1)
+    assert str(r.value) == 'Start position of channel out of universe (1..512): 0'
+    pyartnet.DmxChannel(univ, 1, 1)
+
+    with pytest.raises(pyartnet.errors.ChannelOutOfUniverseError) as r:
+        pyartnet.DmxChannel(univ, 513, 1)
+    assert str(r.value) == 'Start position of channel out of universe (1..512): 513'
+    pyartnet.DmxChannel(univ, 512, 1)
+
+    with pytest.raises(pyartnet.errors.ChannelOutOfUniverseError) as r:
+        pyartnet.DmxChannel(univ, 512, 2)
+    assert str(r.value) == 'End position of channel out of universe (1..512): start: 512 width: 2 * 2bytes -> 513'
+    pyartnet.DmxChannel(univ, 511, 2)
+
+    # 16 Bit Channels
+    with pytest.raises(pyartnet.errors.ChannelOutOfUniverseError) as r:
+        pyartnet.DmxChannel16Bit(univ, 512, 1)
+    assert str(r.value) == 'End position of channel out of universe (1..512): start: 512 width: 1 * 2bytes -> 513'
+    pyartnet.DmxChannel16Bit(univ, 511, 1)

tests/test_output_correction.py

@@ -12,3 +12,9 @@ def test_endpoints(corr):
     while i <= 255:
         assert 0 <= corr(i) <= 255
         i += 0.001
+
+
+@pytest.mark.parametrize('corr', [quadratic, quadruple, cubic])
+def test_endpoints_16bit(corr):
+    assert corr(0) == 0
+    assert corr(0xFFFF, max_val=0xFFFF) == 0xFFFF

tests/test_universe.py

@@ -53,7 +53,6 @@ def test_exceptions(artnet_node: PatchedArtNetNode):
         universe.add_channel(8, 20)
 
 
-
 def test_container(artnet_node: PatchedArtNetNode):
     universe = DmxUniverse(artnet_node)