Implemented interpolation and EngineTracks

src/engine.py

@@ -1,21 +1,41 @@
 from decimal import Decimal
 import logging
 
-from models import EngineFixture, Flickr
+from models import EngineFixture, EngineTrack, Flickr
 
 
 class Engine:
     """Render a lightshow."""
 
     def __init__(self, flickr: Flickr, apis: dict[str, str]) -> None:
-        self.time: Decimal = Decimal.from_float(0)
-        self.fixtures: dict[str, EngineFixture] = {
-            f.id: EngineFixture(f) for s in flickr.spaces for f in s.fixtures
-        }
-        self.flickr = flickr
         self.apis = apis
+        self.flickr = flickr
+        self.tracks: list[EngineTrack] = []
+        self.fixtures: dict[str, EngineFixture] = {}
 
-        # Validate fixture APIs
+    def load_sequence(self, sequence_id: str):
+        """Loads a specific sequence and resets the render paremeters."""
+        sequence = next(
+            filter(lambda s: s.id == sequence_id, self.flickr.sequences), None
+        )
+        if sequence is None:
+            raise ValueError(f"Sequence with id [{sequence_id}] not found.")
+        self.load_tracks([t.id for t in sequence.tracks])
+
+    def load_tracks(self, track_ids: list[str]):
+        """Loads specific tracks and resets the render paremeters."""
+        self.render_time: Decimal = Decimal.from_float(0)
+        self.tracks = [
+            EngineTrack(t)
+            for s in self.flickr.sequences
+            for t in s.tracks
+            if t.id in track_ids
+        ]
+
+        # Create and validate fixtures
+        self.fixtures = {
+            f.id: EngineFixture(f) for s in self.flickr.spaces for f in s.fixtures
+        }
         for id, f in self.fixtures.items():
             if f.fixture.api not in self.apis.keys():
                 logging.warning(
@@ -25,12 +45,16 @@ class Engine:
 
     def next(self, step_size: Decimal):
         """Calculate light values for next step in time. Step size in seconds."""
-        prev_time = self.time
-        self.time += step_size
+        prev_time = self.render_time
+        self.render_time += step_size
 
-        for s in self.flickr.sequences:
-            for t in s.tracks:
-                # TODO: Calculate new state
+        for t in self.tracks:
+            if t.next is None:
+                continue
+
+            # Move keyframes forward
+            while step_size > 0:
+                # TODO
                 pass
 
     def propagate(self):

src/models.py

@@ -1,4 +1,5 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+import logging
 from dataclasses_json import dataclass_json
 from enum import Enum
 from decimal import Decimal
@@ -26,6 +27,18 @@ class Capabilities(Enum):
     COLOR = "color"
 
 
+class Parameter(Enum):
+    """Possible keyframe/fixture parameters."""
+
+    ON = "on"
+    HUE = "hue"
+    BRIGHTNESS = "brightness"
+    SATURATION = "saturation"
+    TEMPERATURE = "temperature"
+    TOPIC = "topic"
+    TRANSITION = "transition"
+
+
 @dataclass_json
 @dataclass
 class Fixture:
@@ -69,18 +82,44 @@ class EngineFixture:
     on: bool = False
     """On state."""
 
-    bri: float = 0
+    brightness: float = 0
     """Brightness state."""
 
     hue: float = 0
     """Hue state."""
 
-    sat: float = 0
+    saturation: float = 0
     """Saturation state."""
 
+    temperature: float = 0
+    """Temperature state."""
+
     transition: float = 0
     """Transition time in seconds."""
 
+    topic: str | None = None
+    """Topic fixture currently displays."""
+
+    def set(self, parameter: Parameter, value: None | bool | float | int | str):
+        """Set value of fixture based on parameter."""
+        match parameter:
+            case Parameter.ON:
+                self.on = value
+            case Parameter.HUE:
+                self.hue = value
+            case Parameter.BRIGHTNESS:
+                self.brightness = value
+            case Parameter.SATURATION:
+                self.saturation = value
+            case Parameter.TEMPERATURE:
+                self.temperature = value
+            case Parameter.TOPIC:
+                self.topic = value
+            case _:
+                logging.warning(
+                    f"Unknown parameter [{parameter}] with value [{value}] for fixture id [{self.fixture.id}]."
+                )
+
 
 @dataclass_json
 @dataclass
@@ -100,6 +139,16 @@ class Space:
     """Metatag to track version of datastructure."""
 
 
+class Interpolation(Enum):
+    STEP = "step"
+    """Keep value constant after keyframe, and jump to new value for the next keyframe. Required for bool and str value types."""
+
+    LINEAR = "linear"
+    EASE_IN_OUT = "ease-in-out"
+    EASE_IN = "ease-in"
+    EASE_OUT = "ease-out"
+
+
 @dataclass_json
 @dataclass
 class Keyframe:
@@ -108,12 +157,26 @@ class Keyframe:
     time: Decimal
     """Point of time of the keyframe. Stored as decimal for high precision."""
 
-    parameter: str
+    parameter: Parameter
     """Parameter targeted by frame value."""
 
     value: None | bool | float | int | str = None
     """Targeted state value reached by reaching the keyframe."""
 
+    interpolation: Interpolation = Interpolation.STEP
+    """Interpolation of value following the keyframe."""
+
+
+@dataclass
+class EngineKeyframe:
+    """A keyframe for engine use during rendering."""
+
+    keyframes: list[Keyframe]
+    """Reference keyframes."""
+
+    delta: Decimal
+    """Remaining time in seconds relative to previous keyframes."""
+
 
 @dataclass_json
 @dataclass
@@ -133,6 +196,45 @@ class Track:
     """Sequence of keyframes describing actions"""
 
 
+@dataclass
+class EngineTrack:
+    """A track for engine use during rendering."""
+
+    track: Track
+    """Reference track."""
+
+    keyframes: list[EngineKeyframe] = field(default_factory=list)
+    """Relative keyframes of track."""
+
+    previous: EngineKeyframe | None = None
+    """Keyframes before next keyframe."""
+
+    @property
+    def next(self) -> EngineKeyframe | None:
+        """Next keyframe in track, or None if empty."""
+        if len(self.keyframes) == 0:
+            return None
+        return self.keyframes[0]
+
+    def __post_init__(self):
+        # Calculate relative keyframes
+        sorted_keyframes = sorted(self.track.keyframes, key=lambda k: k.time)
+        prev_time: Decimal = Decimal.from_float(0)
+        curr_time: Decimal = Decimal.from_float(0)
+        keyframes = []
+        for k, next in zip(sorted_keyframes, [*sorted_keyframes[1:], None]):
+            curr_time = k.time
+            keyframes.append(k)
+
+            if next is None or k.time != next.time:
+                self.keyframes.append(EngineKeyframe(keyframes, curr_time - prev_time))
+
+                if next is not None:
+                    keyframes = []
+                    prev_time = curr_time
+                    curr_time = next.time
+
+
 @dataclass_json
 @dataclass
 class Sequence: