summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rwxr-xr-xblc/audio/interface.py7
-rw-r--r--blc/audio/mpv.py8
-rw-r--r--blc/output.py4
-rwxr-xr-xblc/workspace.py217
4 files changed, 180 insertions, 56 deletions
diff --git a/blc/audio/interface.py b/blc/audio/interface.py
index fff19cf..a33b2fe 100755
--- a/blc/audio/interface.py
+++ b/blc/audio/interface.py
@@ -48,6 +48,13 @@ class AudioPlayer(ABC):
"""Return if the player is playing or not."""
return
+ def terminate(self):
+ """Conduct any necessary cleanup here.
+
+ No further methods of the player may be called after this one.
+ """
+ return
+
def __init__(self, fname, args=()):
self.fname = fname
self.args = args
diff --git a/blc/audio/mpv.py b/blc/audio/mpv.py
index da21494..a71adf7 100644
--- a/blc/audio/mpv.py
+++ b/blc/audio/mpv.py
@@ -13,6 +13,7 @@ class MPVPlayer(AudioPlayer):
def play(self, start=-1):
## TODO: Raise error if already playing?
+ self.started = True
self.player.pause = False
def pause(self):
@@ -35,9 +36,10 @@ class MPVPlayer(AudioPlayer):
@property
def playing(self) -> bool:
- if self.started:
- return not (self.player.pause or self.player.eof_reached)
- return False
+ return self.started and not (self.player.pause or self.player.eof_reached)
+
+ def terminate(self):
+ self.player.terminate()
def __init__(self, fname, args=()):
super().__init__(fname, args)
diff --git a/blc/output.py b/blc/output.py
index dfe9a3d..45a79a0 100644
--- a/blc/output.py
+++ b/blc/output.py
@@ -24,7 +24,9 @@ class LightingOutput(ABC):
channel entries may not be repeated. Each channel must be either an instance of
Workspace.channel or a 2-tuple of integers (dmx universe, dmx channel). value must be
between 0 and 255, inclusive.
- inclusive.
+
+ Note that this function need only be called when values change; the output is expected
+ to hold all channels at their last values.
"""
return
diff --git a/blc/workspace.py b/blc/workspace.py
index 1aa6712..eb17578 100755
--- a/blc/workspace.py
+++ b/blc/workspace.py
@@ -103,6 +103,7 @@ import os.path
import subprocess as subp
import xml.etree.ElementTree as etree
import random
+from typing import Any, List, Tuple, Sequence, Dict, Iterable, NewType, Callable, Union, Set
## BEGIN Constants
@@ -123,11 +124,17 @@ PINGPONG = "PingPong"
QXW = "{http://www.qlcplus.org/Workspace}"
+## A pair (channel, value)
+CVPair = NewType("CVPair", Tuple["Channel", int])
+
+AudioInstance = NewType("AudioInstance", Tuple[int, int, str, int, int, int])
+Render = NewType("Render", Tuple[Iterable[CVPair], Iterable[AudioInstance], int, Any])
+
## END Constants
## BEGIN Utility functions
-def ffprobe_audio_length(f, path="ffprobe"):
+def ffprobe_audio_length(f: str, path: str = "ffprobe") -> int:
"""Use ffprobe to check audio length in milliseconds.
Will always return the nearest whole millisecond greater than or equal to the duration.
@@ -150,13 +157,22 @@ class Fixture:
May be composed of multiple channels.
"""
+ name: str
+ address_start: int
+ channel_count: int
+ mode: Any
+ universe: "Universe"
+ id: int
+ channnels: Tuple["Channel"]
+
def __hash__(self):
return self._hash
def __repr__(self):
return "Fixture(id=%d, name=%s, universe=%d, start=%d, channels=%d)" % (self.id, self.name, self.universe.id, self.address_start, self.channel_count)
- def __init__(self, id_, name, address, universe, mode, channels=1):
+ def __init__(self, id_: int, name: str, address: int, universe: "Universe", mode: Any,
+ channels: int = 1):
self.name = name
self.address_start = address
self.channel_count = channels
@@ -165,10 +181,15 @@ class Fixture:
self.id = id_
self._hash = hash((self.name, self.address_start, self.channel_count, self.mode,
self.id, self.universe))
- self.channels = [Channel(self, i) for i in range(channels)]
+ self.channels = tuple((Channel(self, i) for i in range(channels)))
class Channel:
"""Class representing a single output channel."""
+ fixture: Fixture
+ offset: int
+ address: int
+ universe: "Universe"
+
def __hash__(self):
return self._hash
@@ -178,7 +199,7 @@ class Channel:
def __iter__(self):
return iter((self.universe.id, self.address,))
- def __init__(self, fixture, offset):
+ def __init__(self, fixture: Fixture, offset: int):
if offset >= fixture.channel_count or offset < 0:
raise ValueError("Invalid offset")
self.fixture = fixture
@@ -189,6 +210,10 @@ class Channel:
class ChannelGroup:
"""Class representing a group of output channels."""
+ id: int
+ name: str
+ channels: Tuple[Channel, ...]
+
def __hash__(self):
return self._hash
@@ -196,7 +221,7 @@ class ChannelGroup:
return "ChannelGroup(id=%d, name=%s, channels=(%s))" % (self.id, self.name,
", ".join((repr(c) for c in self.channels)))
- def __init__(self, id_, name, channels):
+ def __init__(self, id_: int, name: str, channels: Tuple[Channel]):
self.id = id_
self.name = name
self.channels = tuple(channels)
@@ -205,13 +230,16 @@ class ChannelGroup:
class Universe:
"""Class representing an output universe."""
+ id: int
+ name: str
+
def __hash__(self):
return self._hash
def __repr__(self):
return "Universe(id=%d, name=%s)" % (self.id, self.name)
- def __init__(self, id_, name):
+ def __init__(self, id_: int, name: str):
self.id = id_
self.name = name
@@ -250,24 +278,28 @@ class Function(ABC):
As id is not necessarily unique, hash may be used to differentiate functions: this is
guaranteed to be globally unique and is calculated once on instantiation.
"""
- repr_attr = ("id", "name",)
+ id: int
+ type: str
+ name: str
+ hidden: bool
+ duration: int
+ actual_duration: int
+ scope: Tuple[Channel, ...]
+
+ repr_attr: Tuple[Union[str, Tuple[str, Callable[[Any], str]]], ...] = ("id", "name",)
def __hash__(self):
return self._hash
@staticmethod
- def get_data():
+ def get_data() -> Any:
"""Return an initial state for the function.
This function may not be a static method. Always invoke on an instance.
"""
return None
- @staticmethod
- def copy_data(data):
- """Copy the given function state.
-
- This function may not be a static method. Always invoke on an instance.
- """
+ def copy_data(self, data: Any) -> Any:
+ """Copy the given function state."""
return None
def __repr__(self):
@@ -283,7 +315,7 @@ class Function(ABC):
return "%s(%s)" % (self.__class__.__name__, ", ".join(buff))
@abstractmethod
- def render(self, t: int, data=None):
+ def render(self, t: int, data: Any = None) -> Render:
"""Render the function at the given time.
Parameters:
@@ -316,9 +348,9 @@ class Function(ABC):
It is an error to call render with data that has been used to render a future time; this
is undefined behavior.
"""
- return
- def __init__(self, id_, type_, name, scope, hidden=False, duration=-1, actual_duration=-1):
+ def __init__(self, id_: int, type_: str, name: str, scope: Iterable[Channel],
+ hidden: bool = False, duration: int = -1, actual_duration: int = -1):
self.id = id_
self.type = type_
self.name = name
@@ -332,10 +364,16 @@ class Function(ABC):
class FadeFunction(Function):
"""QLC function that can fade in/out."""
- def __init__(self, id_, type_, name, scope, hidden=False, duration=-1, actual_duration=-1, fade_in=0, fade_out=0):
+ fade_in: int
+ fade_out: int
+
+ def __init__(self, id_: int, type_: str, name: str, scope: Sequence[Channel],
+ hidden: bool = False, duration: int = -1, actual_duration: int = -1,
+ fade_in: int = 0, fade_out: int = 0):
if fade_in >= QLC_INFTY or fade_out >= QLC_INFTY:
raise ValueError("Fades cannot be infinite")
- super().__init__(id_, type_, name, scope, hidden=hidden, duration=duration, actual_duration=actual_duration)
+ super().__init__(id_, type_, name, scope, hidden=hidden, duration=duration,
+ actual_duration=actual_duration)
self.fade_in = min(QLC_INFTY, fade_in)
self.fade_out = min(QLC_INFTY, fade_out)
@@ -343,15 +381,14 @@ class FadeFunction(Function):
class Advanceable(ABC):
"""Function that may be advanced."""
- @abstractstaticmethod
- def advance(t, data):
+ def advance(self, t: int, data: Any):
"""Advance the function."""
return data
class PreRenderable(ABC):
"""Function that may be pre-rendered."""
@abstractmethod
- def render_all(self, minnx=10):
+ def render_all(self, minnx: int = 10):
"""Render the entire function.
This may raise a ValueError if the Function is not actually PreRenderable; inheritance
@@ -366,14 +403,52 @@ class PreRenderable(ABC):
"""
return ()
+class CacheRenderable(PreRenderable, ABC):
+ """Function that may be cached.
+
+ Note that a function inheriting this need not be static under every possible definition of
+ that function, only some.
+ """
+ _render = None
+
+ class CacheRenderData:
+ last_idx: int
+ values: Dict[Channel, int]
+
+ def __init__(self, last_idx: int, values: Dict[Channel, int]):
+ self.last_idx = last_idx
+ self.values = values
+
+ def cache_render(self, minnx: int = 10) -> bool:
+ """Render the entire function.
+
+ The result is that the function is essentially pre-rendered but behaves identically to a
+ regular function, i.e. that rendering is done using the function's render function as
+ usual.
+
+ @returns True if the function was cached and False otherwise
+ """
+ try:
+ render = self.render_all(minnx=minnx)
+ except ValueError:
+ return False
+ self._render = render
+ return True
+
+ def _render_from_cache(self, t: int, data: CacheRenderData = None) -> Render:
+ raise NotImplementedError("Not ready yet")
+
## END Base classes
## BEGIN Function classes
class Audio(FadeFunction):
"""Class for a QLC+ audio function."""
+ fname: str
+ run_order: str
+
repr_attr = ("id", "fname", "fade_in", "fade_out",)
- def render(self, t, data=None):
+ def render(self, t: int, data = None):
"""Render the audio function.
We do not seek to do anything related to audio in this library: the responsibility for
@@ -385,7 +460,8 @@ class Audio(FadeFunction):
return (), ((0, self.id, self.fname, self.fade_in, self.fade_out, self.duration-self.fade_out),), self.duration+1-t, data
- def __init__(self, id_, name, fname, fade_in, fade_out, length, run_order=SINGLESHOT, hidden=False):
+ def __init__(self, id_: int, name: str, fname: str, fade_in: int, fade_out: int,
+ length: int, run_order: str = SINGLESHOT, hidden: bool = False):
super().__init__(id_, AUDIO, name, (), hidden=hidden, duration=length,
actual_duration=length, fade_in=fade_in, fade_out=fade_out)
self.fname = fname
@@ -404,29 +480,42 @@ class Scene(Function):
"""All arguments are unused."""
return self.values, (), QLC_INFTY, data
- def __init__(self, id_, name, values, hidden=False):
- super().__init__(id_, SCENE, name, (c for c,v in values), hidden=hidden, duration=-1, actual_duration=-1)
+ def __init__(self, id_: int, name: str, values: Iterable[Tuple[Channel, int]],
+ hidden: bool = False):
+ super().__init__(id_, SCENE, name, (c for c,v in values), hidden=hidden, duration=-1,
+ actual_duration=-1)
self.values = tuple(values)
self._hash = hash((self._hash, self.values))
class ChaserStep(FadeFunction):
- """A single step in a chaser."""
+ """A single step in a chaser.
+
+ This function's methods are not intended to be used outside of Chaser's methods.
+ """
+ function: Function
+ hold: int
+
repr_attr = ("id", "name", "hold", "fade_in", "fade_out", ("function", lambda f: f.id))
class ChaserStepData:
"""Data for the step."""
- def __init__(self, fd, start_time, end_time):
+ fd: Any
+ start_time: int
+ end_time: int
+
+ def __init__(self, fd: Any, start_time: int, end_time: int):
self.fd = fd
self.start_time = start_time
self.end_time = end_time
- def get_data(self, start_time=0):
- return self.ChaserStepData(fd=self.function.get_data(), start_time=start_time, end_time=self.duration)
+ def get_data(self, start_time: int = 0):
+ return self.ChaserStepData(fd=self.function.get_data(), start_time=start_time,
+ end_time=self.duration)
@classmethod
def copy_data(cls, data):
return cls.ChaserStepData(fd=data.fd, start_time=data.start_time, end_time=data.end_time)
- def render(self, t, data:ChaserStepData=None):
+ def render(self, t: int, data:ChaserStepData=None):
"""DO NOT CALL OUTSIDE OF Chaser.
The logic is different here: we never check the actual duration of this function and
@@ -443,7 +532,7 @@ class ChaserStep(FadeFunction):
## Render the function at time t
values, acues, nx, data.fd = self.function.render(t, data=data.fd)
## Determine the multiplier
- mul = 1
+ mul = 1.0
if self.fade_in > 0 and t < self.fade_in: ## Fade in first
mul = min(1,t/self.fade_in)
nx = 1
@@ -473,11 +562,10 @@ class ChaserStep(FadeFunction):
return (values, mul), tuple(nacues), nx, data
- def __init__(self, id_, fade_in, fade_out, hold, function):
+ def __init__(self, id_: int, fade_in: int, fade_out: int, hold: int, function: Function):
super().__init__(id_, STEP, function.name, function.scope, hidden=False,
duration=hold+fade_in, actual_duration=hold+fade_out+fade_in,
fade_in=fade_in, fade_out=fade_out)
- self.id = id_
self.hold = hold
self.function = function
self._hash = hash((self._hash, self.function, self.hold))
@@ -486,7 +574,13 @@ class Chaser(Function, Advanceable, PreRenderable):
"""Class for representing a QLC+ Chaser or Sequence.
Since they essentially do the same thing (Chaser being more general), they have only one
- class here."""
+ class here.
+ """
+ steps: Tuple[ChaserStep, ...]
+ run_order: str
+ direction: str
+ is_chaser: bool
+
repr_attr = ("id", "name", ("steps", lambda s: ",".join((str(i.id) for i in s))))
class ChaserData:
"""Current state of a chaser.
@@ -495,17 +589,23 @@ class Chaser(Function, Advanceable, PreRenderable):
[(step number, step data), ...]
"""
+ step_data: List[Tuple[int, ChaserStep.ChaserStepData]]
+ obey_loop: bool
+ forward: bool
+
@property
- def current_step(self):
+ def current_step(self) -> int:
+ """Returns the index of the current step."""
return self.step_data[-1][0] if self.step_data else -1
- def __init__(self, step_data, obey_loop, forward):
- self.step_data = step_data
+ def __init__(self, step_data: Iterable[Tuple[int, ChaserStep.ChaserStepData]], obey_loop: bool,
+ forward: bool = True):
+ self.step_data = list(step_data )
self.obey_loop = obey_loop
- self.forward = True
+ self.forward = forward
@staticmethod
- def advance(t, data):
+ def advance(t: int, data):
"""End the current chaser step.
After calling this function, the chaser must be rendered at a time at least t before
@@ -520,10 +620,12 @@ class Chaser(Function, Advanceable, PreRenderable):
return self.ChaserData(step_data=[], obey_loop=True, forward=True)
def copy_data(self, data):
+ if data is None:
+ return None
return self.ChaserData(step_data=[(a,self.steps[a].copy_data(b)) for a,b in data.step_data],
obey_loop=data.obey_loop, forward=data.forward)
- def next_step(self, data): ## TODO: Implement other chaser types
+ def next_step(self, data) -> Tuple[int, "Chaser.ChaserData"]: ## TODO: Implement other chaser types
"""Return the next step in the chaser.
Returns a tuple (next step number, data)
@@ -607,7 +709,7 @@ class Chaser(Function, Advanceable, PreRenderable):
return tuple(vals.items()), tuple(sorted(acues, key=lambda a: a[0])), nx, data
- def render_all(self, minnx=1):
+ def render_all(self, minnx: int = 1):
"""Render the entire Chaser."""
## Verify that we can render this one
if self.run_order != SINGLESHOT:
@@ -659,7 +761,8 @@ class Chaser(Function, Advanceable, PreRenderable):
return tuple(steps)
- def __init__(self, id_, name, steps, hidden=False, run_order=SINGLESHOT, direction=FORWARD):
+ def __init__(self, id_: int, name: str, steps: Iterable[ChaserStep], hidden: bool = False,
+ run_order: str = SINGLESHOT, direction: str = FORWARD, is_chaser: str = False):
if run_order not in (LOOP, SINGLESHOT):
raise NotImplementedError("Only Loop and SingleShot chasers are currently supported")
if direction not in (FORWARD,):
@@ -683,10 +786,14 @@ class Chaser(Function, Advanceable, PreRenderable):
self.steps = tuple(steps)
self.run_order = run_order
self.direction = direction
+ self.is_chaser = is_chaser
self._hash = hash((self._hash, self.steps, self.run_order, self.direction))
class ShowFunction(Function):
"""Class for representing a function in a show."""
+ function: Function
+ start_time: int
+
repr_attr = ("id", "name", "start_time", ("function", lambda f: f.id))
def render(self, t, data=None):
if data is None:
@@ -696,7 +803,7 @@ class ShowFunction(Function):
return values, tuple(((at+self.start_time,hash((self.id, self.start_time, aid)),
*others) for at,aid,*others in acues)), nx, data
- def __init__(self, id_, name, function, start_time):
+ def __init__(self, id_: int, name: str, function: Function, start_time: int):
if function.actual_duration >= QLC_INFTY:
raise ValueError("Cannot have infinite-length functions in shows")
super().__init__(id_, "ShowFunction", name, function.scope, duration=function.duration,
@@ -707,6 +814,7 @@ class ShowFunction(Function):
class ShowTrack(Function):
"""Class for representing a track in a show."""
+ functions: Tuple[ShowFunction, ...]
repr_attr = ("id", "name", ("functions", lambda fs: ','.join(("%d@%d" % (f.function.id, f.start_time) for f in fs))))
def get_data(self):
return tuple((f.function.get_data() for f in self.functions))
@@ -743,7 +851,7 @@ class ShowTrack(Function):
return tuple(values.items()), tuple(sorted(acues, key=lambda a: a[0])), nx, data
- def __init__(self, id_, name, functions):
+ def __init__(self, id_: int, name: str, functions: Iterable[ShowFunction]):
dur = -1
adur = -1
self.functions = tuple(sorted(functions, key=lambda f: f.start_time))
@@ -757,12 +865,16 @@ class ShowTrack(Function):
super().__init__(id_, "ShowTrack", name, scope, duration=dur, actual_duration=adur)
self._hash = hash((self._hash, self.functions))
-class Show(Function, PreRenderable):
+class Show(Function, CacheRenderable):
"""Class representing a QLC+ show."""
+ tracks: Tuple[ShowTrack, ...]
+
def get_data(self):
return tuple((t.get_data() for t in self.tracks))
def copy_data(self, data):
+ if isinstance(data, CacheRenderable.CacheRenderData):
+ return data.copy()
return tuple((t.copy_data(d) for t,d in zip(self.tracks, data)))
def render(self, t, data=None):
@@ -851,7 +963,7 @@ class Show(Function, PreRenderable):
return ((tuple(cues), tuple(sorted(acues, key=lambda a: a[1])),),)
- def __init__(self, id_, name, tracks):
+ def __init__(self, id_: int, name: str, tracks: Iterable[ShowTrack]):
scope = set()
dur = -1
adur = -1
@@ -875,7 +987,7 @@ class Workspace:
Should be created using Workspace.load and is assumed to be immutable.
"""
@classmethod
- def load(cls, wfname, audio_length=ffprobe_audio_length):
+ def load(cls, wfname: str, audio_length: Callable[[str], int] = ffprobe_audio_length):
"""Load a QLC+ workspace.
This function returns the created Workspace object.
@@ -891,8 +1003,9 @@ class Workspace:
wdir = os.path.dirname(os.path.abspath(wfname))
creator = ws.find(QXW+"Creator")
- self = cls(creator.find(QXW+"Name").text, creator.find(QXW+"Version").text,
- creator.find(QXW+"Author").text)
+ self = cls(creator.find(QXW+"Name").text or "No Name",
+ creator.find(QXW+"Version").text or "No Version",
+ creator.find(QXW+"Author").text or "No Author")
engine = ws.find(QXW+"Engine")
@@ -1086,7 +1199,7 @@ class Workspace:
steps.append(step)
func = Chaser(sid, name, steps, hidden=hidden,
run_order=func.find(QXW+"RunOrder").text,
- direction=func.find(QXW+"Direction").text)
+ direction=func.find(QXW+"Direction").text, is_chaser=True)
else:
raise ValueError("Unhandled type %s" % ftype)
@@ -1096,7 +1209,7 @@ class Workspace:
return self
- def __init__(self, creator, version, author):
+ def __init__(self, creator: str, version: str, author: str):
self.universes = {}
self.fixtures = {}
self.channel_groups = {}
generated by cgit v1.2.3 (git 2.25.1) at 2025-05-23 12:27:22 -0400