thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
|
|
|
|
def clamp_channel(value: int) -> int:
|
|
return max(0, min(255, int(value)))
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FrameLayer:
|
|
name: str
|
|
priority: int
|
|
values_by_universe: dict[int, dict[int, int]]
|
|
precedence_map_by_universe: dict[int, dict[int, str]] = field(default_factory=dict)
|
|
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
|
|
@property
|
|
def values(self) -> dict[int, int]:
|
|
return self.values_by_universe.get(1, {})
|
|
|
|
@property
|
|
def precedence_map(self) -> dict[int, str]:
|
|
return self.precedence_map_by_universe.get(1, {})
|
|
|
|
@classmethod
|
|
def from_channel_values(
|
|
cls,
|
|
name: str,
|
|
priority: int,
|
|
values: dict[int, int],
|
|
precedence_map: dict[int, str] | None = None,
|
|
*,
|
|
universe: int = 1,
|
|
) -> "FrameLayer":
|
|
return cls(
|
|
name=name,
|
|
priority=priority,
|
|
values_by_universe={universe: values},
|
|
precedence_map_by_universe={universe: precedence_map or {}},
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DmxFrame:
|
|
universe: int = 1
|
|
values: list[int] = field(default_factory=lambda: [0] * 512)
|
|
source_map: list[str] = field(default_factory=lambda: ["idle"] * 512)
|
|
|
|
def set_channel(self, channel: int, value: int, source: str = "manual") -> None:
|
|
if not 1 <= channel <= 512:
|
|
raise ValueError("Channel must be in range 1..512")
|
|
self.values[channel - 1] = clamp_channel(value)
|
|
self.source_map[channel - 1] = source
|
|
|
|
def copy(self) -> DmxFrame:
|
|
return DmxFrame(self.universe, self.values.copy(), self.source_map.copy())
|
|
|
|
|
|
def merge_layers(layers: list[FrameLayer], blackout: bool = False) -> dict[int, DmxFrame]:
|
|
universes = sorted(
|
|
{
|
|
int(universe)
|
|
for layer in layers
|
|
for universe in layer.values_by_universe
|
|
}
|
|
) or [1]
|
|
if blackout:
|
|
return {universe: DmxFrame(universe=universe) for universe in universes}
|
|
|
|
ordered = sorted(layers, key=lambda layer: (layer.priority, layer.created_at))
|
|
frames = {universe: DmxFrame(universe=universe) for universe in universes}
|
|
for universe, frame in frames.items():
|
|
for channel in range(1, 513):
|
|
channel_candidates: list[tuple[FrameLayer, int]] = []
|
|
for layer in ordered:
|
|
values = layer.values_by_universe.get(universe, {})
|
|
if channel in values:
|
|
channel_candidates.append((layer, clamp_channel(values[channel])))
|
|
if not channel_candidates:
|
|
continue
|
|
|
|
precedence_map = channel_candidates[-1][0].precedence_map_by_universe.get(universe, {})
|
|
precedence = precedence_map.get(channel, "ltp").lower()
|
|
if precedence == "htp":
|
|
chosen_layer, chosen_value = max(channel_candidates, key=lambda pair: pair[1])
|
|
else:
|
|
chosen_layer, chosen_value = channel_candidates[-1]
|
|
frame.set_channel(channel, chosen_value, chosen_layer.name)
|
|
return frames
|