Update to upstream

This commit is contained in:
Constantin Gierczak--Galle 2023-12-15 10:45:10 +01:00
parent ac928d20d9
commit 1089c3941d
3 changed files with 117 additions and 110 deletions

View file

@ -2,7 +2,7 @@
let
attrs = {
name = "pyjecteur";
version = "2.0";
version = "3.0";
doCheck = false;
src = ./. ;
passthru = {

View file

@ -1,135 +1,107 @@
"""
Module providing class for handling fixtures and generating the appropriate DMX.
"""
from copy import deepcopy
from typing import Any, Callable, Optional, Union
""""""
from colour import Color
from .reactive import BaseReactiveValue, ReactiveMixin
from .widget import Widget
from .lights import AbstractLight
from .reactive import RBool, RColor, RInt, RList
class Universe:
"""Represents a DMX universe.
class Strob(AbstractLight):
address_size = 2
freq = RInt(0, 1)
dim = RInt(0, 0)
Manages the adress space and responsibles for sending DMX to widget when
`Universe.update_dmx()` is called.
class UVBar(AbstractLight):
address_size = 3
strob = RInt(0, 1)
dim = RInt(0, 0)
class StrobInv(AbstractLight):
address_size = 2
freq = RInt(0, 0)
dim = RInt(0, 1)
class Wash(AbstractLight):
"""
Wash
"""
lights = {}
address_size = 14
def __init__(self, widget):
"""
Initializes the Universe
widget must be a class providing `widget.set_dmx(data, address)`
"""
self.widget: Widget = widget
def register(self, light: "AbstractLight", address: int) -> None:
"""
Register a light at the specified address
"""
# TODO: add checks for address overlapping
self.lights[light] = address
light.register_universe(self)
light.update_dmx()
def update_dmx(self, light: "AbstractLight", data: Union[bytearray, bytes]) -> None:
"""
Update the dmx data of the specified light
"""
# TODO: add checks for length
self.widget.set_dmx(self.lights[light], data)
pan = RInt(0, 0)
tilt = RInt(0, 1)
speed = RInt(0, 2)
color = RColor(Color("black"), 3)
white = RInt(0, 6)
dimmer = RInt(255, 9)
shutter = RBool(True, 10, true_val=b"\x15")
zoom = RInt(0, 11)
class AbstractLight:
class Tradi(AbstractLight):
"""
Abstract class for lights
Tradi RGB
"""
address_size: int = 0
address_size = 3
def __init__(self):
self._universe: Optional[Universe] = None
# The dmx values
self._dmx: bytes = bytearray(self.address_size)
# dmx memory_view to change in O(1) the values
self._dmx_mv = memoryview(self._dmx)
color = RColor(Color("black"), 3)
# Dict holdin conversion functions for attr values to dmx:
# { attr_name => (address, length, converter function) }
self._attrs_to_dmx: dict[str, tuple[int, int, Callable[[Any], bytes]]] = {}
# List holding conversion functions from dmx bytes to attrs.
# [ ( "attr_name", dmx_addr, length, converter function ) ]
self._dmx_to_attrs: list[
Optional[tuple[str, int, int, Callable[[bytes], Any]]]
] = [None for _ in range(self.address_size)]
class ParMKII(AbstractLight):
"""
Par 56 led
"""
self._enable_auto_update: bool = False
address_size = 8
for key, rValueObject in self.__class__.__dict__.items():
if isinstance(rValueObject, BaseReactiveValue):
# On copie la valeur
val = deepcopy(rValueObject.value)
if isinstance(val, ReactiveMixin):
val.light = self
val.key = key
self._attrs_to_dmx[key] = rValueObject.attr_to_dmx()
color = RColor(Color("black"), 0)
amber = RInt(0, 3)
dimmer = RInt(255, 7)
for i, length, callback in rValueObject.dmx_to_attr():
for k in range(i, i + length):
self._dmx_to_attrs[k] = (key, i, length, callback)
# Finally set the attributes to their value
setattr(self, key, val)
self._enable_auto_update: bool = True
def register_universe(self, universe: "Universe") -> None:
"""Assign a universe to this light"""
if self._universe is not None:
raise ValueError("Can't assign light to more than one universe")
self._universe = universe
class ParLed(AbstractLight):
"""
Par Led Theatre
"""
def update_dmx(self) -> None:
"""
Method to be called when the DMX values may have changed.
address_size = 7
color = RColor(Color("black"), 0)
This method sends DMX velues to the Universe. It is automatically
triggered by property assignments.
"""
if self._universe is not None and self._enable_auto_update:
self._universe.update_dmx(self, self._dmx)
dimmer = RInt(255, 6)
def __setattr__(self, name: str, value: Any) -> None:
"""
Automatically update dmx when a fixture param is set
"""
self.__dict__[name] = value
if not name.startswith("_"):
self.attr_set_hook(name, value)
def attr_set_hook(self, name, value):
"""
Hook to be called when an attribute is set in order to update DMX
values
"""
if name in self._attrs_to_dmx:
# if the attr is linked to dmx, update self._dmx
position, length, converter = self._attrs_to_dmx[name]
self._dmx_mv[position : position + length] = converter(value)
self.update_dmx()
class Blinder(AbstractLight):
"""
Blinder
"""
def __getitem__(self, key) -> bytes:
return self._dmx[key]
address_size = 51
def __setitem__(self, key: int, value: bytes) -> None:
self._dmx_mv[key : key + 1] = value
dimmer = RInt(255, 1)
flash = RInt(0, 2)
colors = RList(
[Color(rgb=(0, 0, 0)) for i in range(16)],
3,
3,
from_byte=RColor.from_bytes,
to_byte=RColor.to_bytes,
)
if self._dmx_to_attrs[key] is not None:
attr, position, length, converter = self._dmx_to_attrs[
key
] # pyright: ignore
self._enable_auto_update = False
setattr(self, attr, converter(self._dmx[position : position + length]))
self._enable_auto_update = True
self.update_dmx()
class LedBar48Ch(AbstractLight):
"""
Led Bar addressed on 48 channels
"""
address_size = 48
colors = RList(
[Color(rgb=(0, 0, 0)) for i in range(16)],
0,
3,
from_byte=lambda x: Color(f"#{x.hex()}"),
to_byte=lambda x: bytes.fromhex(x.hex_l[1:]),
)

View file

@ -4,6 +4,8 @@ Module holding classes for easy specification of fixtures attributes
from typing import Any, Callable, Iterable, Optional
from colour import Color
from .widget import from_bytes, to_bytes
@ -92,6 +94,39 @@ class RBool(BaseReactiveValue):
)
class RColor(BaseReactiveValue):
"""
Boolean light attribute
"""
def __init__( # pylint: disable=too-many-arguments
self,
value,
address,
):
self.value = value
self.address = address
self.length = 3
@staticmethod
def from_bytes(x):
return Color(f"#{x.hex()}")
@staticmethod
def to_bytes(x):
return bytes.fromhex(x.hex_l[1:])
def dmx_to_attr(self):
return [(self.address, self.length, lambda _, x: self.from_bytes(x))]
def attr_to_dmx(self):
return (
self.address,
self.length,
self.to_bytes,
)
class L(ReactiveMixin): # ruff: disable=invalid-name
"""
Thin wrapper around lists to handle reactivity inside lists