 | blc2 - Python library and frontend for running theatrical lighting and SFX. Written for the English 2041F fall 2019 theatre production of "The Cenci".
git clone https://benconnors.ca/git-repos/blc2
|
Log | Files | Refs
commit 2b8a53f98c44e6e78d49b7c246731deef75ed6d3
parent 7f85bd8ed84b23fc4e683ab90fc7babe288f1a27
Author: Ben Connors <benconnors@outlook.com>
Date: Wed, 25 Sep 2019 23:05:14 -0400
Change module layout; start chaser work
- Fix up callbacks
- Clean up function implementations
- More properties to prevent editing of attributes
- Start work on chasers
- Implement framework for chaser steps
Diffstat:
17 files changed, 931 insertions(+), 746 deletions(-)
diff --git a/__init__.py b/blc2/__init__.py
diff --git a/blc2/constants.py b/blc2/constants.py
@@ -0,0 +1,69 @@
+"""Constants module.
+
+Contains some constants used throughout.
+"""
+
+class _Infinity:
+ """Class for the singleton INFTY.
+
+ We assume that this is only used in time-related operations, i.e. that numbers involved
+ are nonnegative.
+ """
+ INFTY = None
+
+ def __repr__(self):
+ return "infty"
+
+ def __str__(self):
+ return "infty"
+
+ def __gt__(self, other):
+ return True
+
+ def __lt__(self, other):
+ return False
+
+ def __add__(self, other):
+ return self
+
+ def __radd__(self, other):
+ return self
+
+ def __sub__(self, other):
+ return self
+
+ def __rsub__(self, other):
+ return self
+
+ def __mul__(self, other):
+ if other == 0:
+ return 0
+ return self
+
+ def __rmul__(self, other):
+ if other == 0:
+ return 0
+ return self
+
+ def __init__(self):
+ if _Infinity.INFTY is not None:
+ raise ValueError("Cannot create two infinities")
+ _Infinity.INFTY = self
+
+INFTY = _Infinity()
+
+AUTO = -2
+
+BXW = "{http://unsuspicious.services/bxw}"
+
+SCENE = "Scene"
+AUDIO = "Audio"
+CHASER = "Chaser"
+CHASERSTEP = "ChaserStep"
+FUNCTION = "Function"
+
+INTERNAL = "Internal"
+EXTERNAL = "External"
+
+INHERIT = "Inherit"
+MANUAL = "Manual"
diff --git a/exceptions.py b/blc2/exceptions.py
diff --git a/functions/__init__.py b/blc2/functions/__init__.py
diff --git a/blc2/functions/audio.py b/blc2/functions/audio.py
@@ -0,0 +1,172 @@
+"""Audio function module.
+
+Contains the definition of the Audio, the audio primitive.
+"""
+
+import xml.etree.ElementTree as et
+
+from .function import Function
+
+from ..constants import AUDIO, BXW, INTERNAL
+from ..exceptions import LoadError
+
+class Audio(Function):
+ """Class representing an audio cue.
+
+ This is the primitive for audio, and all sound cues must be based in some manner off of
+ Audio. This function merely plays a single audio file once, starting at t=0.
+
+ The duration of the audio is automatically determined and is zero if the file does not
+ exist or is unsupported by ffmpeg.
+ """
+ type = AUDIO
+ fade_out_mode = INTERNAL
+
+ def __init__(self, w, id_ = None, name = None, fade_in = 0, fade_out = 0,
+ filename: str = None):
+ super().__init__(w=w, id_=id_, name=name)
+
+ self._filename = filename
+
+ if fade_in < 0 or fade_out < 0:
+ raise ValueError("Fades must be nonnegative")
+ self._fade_out = fade_out
+ self._fade_in = fade_in
+
+ if filename is not None:
+ self._duration = self.w.get_audio_length(filename)
+ self._audio_scope = frozenset(((filename,),))
+ else:
+ self._duration = 0
+ self._audio_scope = frozenset()
+
+ @property
+ def fade_in(self):
+ return self._fade_in
+
+ @fade_in.setter
+ def fade_in(self, v):
+ if v < 0:
+ raise ValueError("Fades must be nonnegative")
+
+ if v != self._fade_in:
+ self._fade_in = v
+ self.w.function_changed(self)
+
+ @property
+ def fade_out(self):
+ return self._fade_out
+
+ @fade_out.setter
+ def fade_out(self, v):
+ if v < 0:
+ raise ValueError("Fades must be nonnegative")
+
+ if v != self._fade_out:
+ self._fade_out = v
+ self.w.function_changed(self)
+
+ @property
+ def scope(self):
+ return ()
+
+ @property
+ def audio_scope(self):
+ return self._audio_scope.union()
+
+ @property
+ def actual_duration(self):
+ return self._duration
+
+ @property
+ def duration(self):
+ return self._duration
+
+ def _set_duration(self, value):
+ """Set the duration.
+
+ This is called by Workspace when the duration for the file has been rechecked; do
+ not call this elsewhere, duration is intended to be immutable from outside the
+ library.
+ """
+ if value != self._duration:
+ self._duration = value
+ self.w.function_changed(self)
+
+ def get_data(self):
+ return None
+
+ def copy_data(self, data):
+ return None
+
+ def render(self, t, data = None):
+ return ((),
+ ((self.id, self._filename, 0, self.fade_in, self.duration, self.fade_out),),
+ None)
+
+ @property
+ def filename(self):
+ """Return the current audio filename."""
+ return self._filename
+
+ @filename.setter
+ def filename(self, value: str):
+ """Set the current audio filename, updating the duration."""
+ if value is not None:
+ self.duration = self.w.get_audio_length(value)
+ self.audio_scope = frozenset(((value,),))
+ else:
+ self.duration = 0
+ self.audio_scope = frozenset()
+
+ self._filename = value
+ self.w.function_changed(self)
+
+ def serialize(self) -> et.Element:
+ e = et.Element(BXW+"function")
+ e.set("type", self.type)
+ e.set("id", str(self.id))
+ e.set("name", self.name)
+ e.set("fade-in", str(self.fade_in))
+ e.set("fade-out", str(self.fade_out))
+ if self.filename is not None:
+ filename = et.SubElement(e, BXW+"filename")
+ filename.text = self.filename
+
+ return e
+
+ @classmethod
+ def deserialize(cls, w, e):
+ if e.tag != BXW+"function":
+ raise LoadError("Invalid function tag")
+ elif e.get("type") != AUDIO:
+ raise LoadError("Load delegated to wrong class (this is a bug)")
+
+ id_ = cls.int_or_none(e.get("id"))
+ if id_ is None:
+ raise LoadError("Function tag has invalid/missing ID")
+
+ name = e.get("name")
+
+ fade_in = e.get("fade-in")
+ try:
+ fade_in = int(fade_in) if fade_in else 0
+ except ValueError:
+ raise LoadError("Invalid fade in")
+
+ fade_out = e.get("fade-out")
+ try:
+ fade_out = int(fade_out) if fade_out else 0
+ except ValueError:
+ raise LoadError("Invalid fade out")
+
+ if len(e) > 1:
+ raise LoadError("Audio tag can have at most one filename")
+ elif len(e) == 1:
+ filename, = e
+ filename = filename.text
+ else:
+ filename = None
+
+ return cls(w=w, id_=id_, name=name, filename=filename, fade_in=fade_in,
+ fade_out=fade_out)
diff --git a/blc2/functions/function.py b/blc2/functions/function.py
@@ -0,0 +1,114 @@
+"""Base function module.
+
+Contains the generic Function interface.
+"""
+
+from abc import ABCMeta, abstractmethod, abstractproperty
+from typing import Set, Any
+
+from ..topology import Fixture
+from ..constants import INFTY, EXTERNAL, FUNCTION
+from ..interfaces import XMLSerializable
+
+class Function(XMLSerializable, metaclass=ABCMeta):
+ """Class representing a generic function.
+
+ Many of the properties here should not be implemented as properties for performance
+ reasons.
+
+ Functions and properties required for rendering, e.g. ``Function.render`` and
+ ``Function.actual_duration``, should be written to be performant.
+
+ Any change to the scope, audio scope, or actual duration of the function must be
+ reported using ``Workspace.function_changed`` with the new values. This should also be
+ extended to any change in e.g. values for lighting and audio primitives.
+ """
+ type: FUNCTION
+ fade_out_mode: EXTERNAL
+
+ def __init__(self, w: "Workspace", id_: int = None, name: str = None):
+ self.w = w
+ self._id = id_ if id_ is not None else w.next_function_id
+ self._name = name if name else "%s %s" % (self.type, self.id)
+
+ self.w.register_function(self)
+
+ @property
+ def id(self):
+ """Return the function's ID."""
+ return self._id
+
+ @property
+ def name(self):
+ """Return the function's name."""
+ return self._name
+
+ @name.setter
+ def name(self, v):
+ if v != self._name:
+ self._name = v if v else "%s %s" % (self.type, self.id)
+ self.w.function_changed(self)
+
+ def delete(self):
+ """Delete the function from the Workspace."""
+ self.w.function_deleted(self)
+
+ @abstractproperty
+ def duration(self) -> int:
+ """Return the function's duration (excluding fades)."""
+
+ @abstractproperty
+ def fade_in(self) -> int:
+ """Return the function's fade in time."""
+
+ @abstractproperty
+ def fade_out(self) -> int:
+ """Return the function's fade out time."""
+
+ @abstractproperty
+ def scope(self) -> Set[Fixture.Channel]:
+ """Return the set of channels affected by this function."""
+
+ @abstractproperty
+ def audio_scope(self) -> Set[str]:
+ """Return the set of audio filenames that may be used by this function."""
+
+ @abstractmethod
+ def get_data(self) -> Any:
+ """Return the default data for this function."""
+
+ @abstractmethod
+ def copy_data(self, data: Any) -> Any:
+ """Duplicate the given data."""
+
+ @abstractproperty
+ def actual_duration(self):
+ """Return the actual duration of the function, including all fades."""
+
+ @abstractmethod
+ def render(self, t: int, data: Any = None):
+ """Render the function at the given time.
+
+ This function must return a 3-tuple:
+
+ (light_cues, audio_cues, new_data)
+
+ Where ``light_cues`` is a iterable of (channel, value) pairs. It is an error for
+ ``light_cues`` to contain channels other than exactly this Function's scope.
+ ``audio_cues`` is an iterable of audio cues of the form:
+
+ (guid, filename, start_time, fade_in, end_time, fade_out)
+
+ The ``guid`` is globally unique: that is, it is unique to that specific audio cue,
+ even if the same file is played multiple times or in different places. Note that
+ ``end_time`` may be ``INFTY``, in which case the file should be played to its end.
+ In the case of infinite-duration chasers, the ``end_time`` may change over
+ subsequent calls to this function.
+
+ Note that ``data`` is not necessarily mutable: ``render`` should be called like:
+
+ lights, sound, data = f.render(t, data)
+
+ :param t: the time to render at, in milliseconds
+ :param data: the function data to use
+ """
diff --git a/blc2/functions/scene.py b/blc2/functions/scene.py
@@ -0,0 +1,159 @@
+"""Scene function module.
+
+Contains the definition of the Scene, the lighting primitive.
+"""
+
+import xml.etree.ElementTree as et
+from typing import Mapping
+
+from .function import Function
+from ..topology import Fixture
+from ..constants import INFTY, SCENE, BXW
+from ..exceptions import LoadError
+
+class Scene(Function):
+ """Class representing a lighting scene.
+
+ This is the primitive for lighting, and all light cues must be based in some manner off
+ of Scene. Scenes are simple, having no fading and infinite duration.
+
+ Modifying the scene can be done in a few ways: the easiest is using the dictionary-
+ style accesses, e.g. setting the values using:
+
+ scene[channel] = value
+
+ Values may be removed from the scene using the ``del`` operator and retrieved in the
+ same fashion. Alternatively, values can be updated in bulk using ``Scene.update`` or
+ set in bulk with ``Scene.set``.
+ """
+ audio_scope = ()
+ scope = frozenset()
+ values = frozenset()
+ type = SCENE
+
+ def __init__(self, w, id_ = None, name = None, values: Mapping[Fixture.Channel, int] = None):
+ super().__init__(w=w, id_=id_, name=name)
+ self._values = {}
+ self._render = ()
+
+ self._update(values, changed=False)
+
+ @property
+ def fade_in(self):
+ return 0
+
+ @property
+ def fade_out(self):
+ return 0
+
+ @property
+ def duration(self):
+ return INFTY
+
+ @property
+ def actual_duration(self):
+ return INFTY
+
+ def _update_render(self, changed=True):
+ self._render = tuple(self._values.items())
+ self.scope = frozenset(self._values.keys())
+ self.values = self._render
+
+ if changed:
+ self.w.function_changed(self)
+
+ def set(self, v):
+ """Set the scene's values to the given ones.
+
+ Equivalent to update, except that all non-given values are deleted from the scene.
+ """
+ self._values = {}
+ self.update(v)
+
+ def _update(self, v, changed=True):
+ if isinstance(v, dict):
+ v = v.items()
+ vn = []
+ for c, val in v:
+ val = int(val)
+ if val < 0 or val > 255:
+ raise ValueError("Values must be integers on [0,256)")
+ vn.append((c, val))
+ for c, val in vn:
+ self._values[c] = val
+ self._update_render(changed=changed)
+
+ def update(self, v):
+ """Update the scene's values."""
+ self._update(v=v, changed=True)
+
+ def __getitem__(self, c: Fixture.Channel):
+ return self._values[c] if c in self._values else None
+
+ def __setitem__(self, c: Fixture.Channel, v: int):
+ self.update(((c, v),))
+
+ def __delitem__(self, c: Fixture.Channel):
+ if c in self._values:
+ del self._values[c]
+ self._update_render()
+
+ def get_data(self):
+ return None
+
+ def copy_data(self, data):
+ return None
+
+ def render(self, t, data = None):
+ return (self._render, (), None)
+
+ def serialize(self):
+ e = et.Element(BXW+"function")
+ e.set("type", self.type)
+ e.set("id", str(self.id))
+ e.set("name", self.name)
+ for c, v in self.values:
+ ce = et.SubElement(e, BXW+"value")
+ ce.set("fixture", str(c.f.id))
+ ce.set("channel", str(c.id))
+ ce.text = str(v)
+
+ return e
+
+ @classmethod
+ def deserialize(cls, w, e):
+ if e.tag != BXW+"function":
+ raise LoadError("Invalid function tag")
+ elif e.get("type") != SCENE:
+ raise LoadError("Load delegated to wrong class (this is a bug)")
+
+ id_ = cls.int_or_none(e.get("id"))
+ if id_ is None:
+ raise LoadError("Function tag has invalid/missing ID")
+
+ name = e.get("name")
+
+ values = {}
+ for ve in e:
+ if ve.tag != BXW+"value":
+ raise LoadError("Invalid value tag")
+
+ fixture = cls.int_or_none(ve.get("fixture"))
+ channel = cls.int_or_none(ve.get("channel"))
+
+ if None in (fixture, channel):
+ raise LoadError("Missing/invalid fixture/channel value")
+ elif fixture not in w.fixtures:
+ raise LoadError("Missing fixture ID %d" % fixture)
+ elif channel >= len(w.fixtures[fixture].channels):
+ raise LoadError("Fixture %d missing channel ID %d" % (fixture, channel))
+
+ channel = w.fixtures[fixture].channels[channel]
+
+ value = cls.int_or_none(ve.text)
+ if value is None:
+ raise LoadError("Missing/invalid value for channel")
+
+ values[channel] = value
+
+ return cls(w=w, id_=id_, name=name, values=values)
diff --git a/interfaces.py b/blc2/interfaces.py
diff --git a/blc2/topology.py b/blc2/topology.py
@@ -0,0 +1,137 @@
+"""Topology module.
+
+Contains the topology classes for representing the physical and virtual topology of the
+lighting setup.
+"""
+
+import xml.etree.ElementTree as et
+
+from .constants import BXW
+from .interfaces import XMLSerializable
+from .exceptions import LoadError
+
+## TODO: Call Workspace.topology_changed when changed
+
+class Fixture(XMLSerializable):
+ """Class representing a lighting fixture.
+
+ Each lighting fixture has a number of channels, which are mapped to the physical
+ topology.
+
+ Channels must be added by changing the value of ``Fixture.channel_count``, which will
+ create the necessary channel objects and handle deletion.
+ """
+ def __init__(self, w: "Workspace", id_: int = None, name: str = None, channel_count: int = 0):
+ self.w = w
+ self.id = id_ if id_ is not None else w.next_fixture_id
+ self.name = name if name else "Fixture %d" % self.id
+
+ self.channels = ()
+ self.channel_count = channel_count
+
+ self.w.register_fixture(self)
+
+ class Channel:
+ """Class representing a single channel on a Fixture.
+
+ The physical address of the channel is stored in the ``Channel.address`` attribute
+ and may be changed at will.
+
+ This program takes a very lax approach to channel addressing: the address is not
+ parsed in any way and duplicate addresses are not handled. For convenience with
+ OLA, Channel is iterable and iteration returns an iterator over its ``address``
+ attribute.
+ """
+ def __init__(self, f: "Fixture", id_: int, name: str = "Intensity", address = (-1,-1)):
+ self.f = f
+ self.id = id_
+ self.name = name
+ self.address = address
+ self._hash = hash((f.id, id_))
+
+ def __hash__(self):
+ return self._hash
+
+ def __iter__(self):
+ return iter(self.address)
+
+ def __repr__(self):
+ return "Channel(fixture={c.f.id}, index={c.id}, name={c.name})".format(c=self)
+
+ @property
+ def channel_count(self):
+ """Return the current number of channels on the fixture."""
+ return len(self.channels)
+
+ @channel_count.setter
+ def channel_count(self, value):
+ """Change the number of channels on the fixture.
+
+ This function handles deletion of removed channels from functions as well.
+ """
+ if value < 0:
+ raise ValueError("Number of channels must be nonnegative")
+ elif value < len(self.channels):
+ for i in range(value, len(self.channels)):
+ self.w.delete_channel(self.channels[i])
+ self.channels = self.channels[:value]
+ elif value > len(self.channels):
+ self.channels += tuple((Fixture.Channel(self, i) for i in range(len(self.channels), value)))
+
+ def __repr__(self):
+ return "Fixture({f.name}, id={f.id}, channels={f.channel_count})".format(f=self)
+
+ def serialize(self):
+ e = et.Element(BXW+"fixture")
+ e.set("name", self.name)
+ e.set("id", str(self.id))
+
+ for c in self.channels:
+ ce = et.SubElement(e, BXW+"channel")
+ ce.set("name", c.name)
+ if c.address is not None:
+ ## TODO: Other addressing modes
+ try:
+ univ, addr = c.address
+ ae = et.SubElement(ce, BXW+"ola")
+ ae.set("universe", str(univ))
+ ae.set("address", str(addr))
+ except ValueError:
+ pass
+
+ return e
+
+ @classmethod
+ def deserialize(cls, w, e):
+ if e.tag != BXW+"fixture":
+ raise LoadError("Invalid fixture tag")
+
+ id_ = cls.int_or_none(e.get("id"))
+ if id_ is None:
+ raise LoadError("Fixture tag has invalid/missing ID")
+
+ name = e.get("name")
+
+ f = cls(w, id_=id_, name=name, channel_count=len(e))
+ for n, channel in enumerate(e):
+ if channel.tag != BXW+"channel":
+ raise LoadError("Invalid channel tag")
+
+ name = channel.get("name")
+ if name is not None:
+ f.channels[n].name = name
+
+ if len(channel) > 1:
+ raise LoadError("Channel can have at most one address")
+ elif len(channel) == 1:
+ address, = channel
+ if address.tag == BXW+"ola":
+ try:
+ address = (int(address.get("universe")), int(address.get("address")),)
+ except (ValueError, TypeError):
+ raise LoadError("Invalid OLA address on channel")
+ else:
+ raise LoadError("Unknown address tag \"%s\"" % address.tag)
+ f.channels[n].address = address
+
+ return f
diff --git a/blc2/utility.py b/blc2/utility.py
@@ -0,0 +1,6 @@
+"""Module containing various utility functions."""
+
+import datetime as dt
+
+def fromisoformat(s):
+ return dt.datetime.strptime(s, "%Y-%m-%dT%H:%M:%S.%f")
diff --git a/blc2/workspace.py b/blc2/workspace.py
@@ -0,0 +1,274 @@
+"""Workspace module.
+
+Contains the main Workspace implementation.
+"""
+
+import datetime as dt
+import json
+import subprocess as subp
+import xml.etree.ElementTree as et
+
+from .constants import AUDIO, SCENE, BXW
+from .functions.function import Function
+from .exceptions import LoadError
+from .interfaces import XMLSerializable
+from .utility import fromisoformat
+
+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.
+
+ Parameters:
+ f: the path to check
+ path: the path of ffprobe
+ """
+ try:
+ a = subp.check_output([path, "-show_format", "-print_format", "json", f], stderr=subp.DEVNULL)
+ except subp.CalledProcessError:
+ return 0
+ return int(1000*float(json.loads(a)["format"]["duration"])+0.5)
+
+class Workspace(XMLSerializable):
+ """Class representing a audiovisual workspace.
+
+ Note that all callbacks are executed synchronously: if they require long periods of
+ time to execute, the callback should handle scheduling the actual work on a different
+ thread.
+ """
+ def __init__(self, name: str, author: str, version: int, modified: dt.datetime):
+ self.name = name
+ self.author = author
+ self.version = version
+ self.modified = modified
+
+ self.fixtures = {}
+ self._last_fixture_id = -1
+ self.functions = {}
+ self._last_function_id = -1
+
+ self._audio_lengths = {}
+
+ self._change_callbacks = {}
+ self._delete_callbacks = {}
+ self._topology_callbacks = []
+
+ def get_audio_length(self, filename: str) -> int:
+ """Determine the audio length of the given file.
+
+ This value is returned from the cache, if available.
+ """
+ if filename not in self._audio_lengths:
+ self._audio_lengths[filename] = ffprobe_audio_length(filename)
+ return self._audio_lengths[filename]
+
+ def recheck_audio_length(self, filename: str) -> int:
+ """Determine the audio length of the given file.
+
+ This function re-probes the value, updating the cache result and the durations of
+ Audio functions.
+ """
+ self._audio_lengths[filename] = ffprobe_audio_length(filename)
+ for f in self.functions.values():
+ if f.type == AUDIO and f.filename == filename:
+ f._set_duration(self._audio_lengths[f.filename]) #pylint: disable=protected-access
+ return self._audio_lengths[filename]
+
+ def recheck_audio_lengths(self):
+ """Recheck and update all audio lengths."""
+ for filename in self._audio_lengths:
+ self._audio_lengths[filename] = ffprobe_audio_length(filename)
+ for f in self.functions.values():
+ if f.type == AUDIO and f.filename in self._audio_lengths:
+ f._set_duration(self._audio_lengths[f.filename]) #pylint: disable=protected-access
+
+ def register_fixture(self, f: "Fixture"):
+ """Register the fixture in the Workspace.
+
+ Always called when the fixture is instantiated.
+ """
+ if f.id in self.fixtures:
+ raise ValueError("A fixture with that ID already exists")
+ self.fixtures[f.id] = f
+
+ def register_function(self, f: Function):
+ """Register the function in the Workspace.
+
+ Always called when the function is instantiated.
+ """
+ if f.id in self.functions:
+ raise ValueError("A function with that ID already exists")
+ self.functions[f.id] = f
+
+ @property
+ def next_fixture_id(self):
+ """Return the next fixture ID."""
+ return self._last_fixture_id + 1
+
+ @property
+ def next_function_id(self):
+ """Return the next function ID."""
+ return self._last_function_id + 1
+
+ def delete_channel(self, c: "Fixture.Channel"):
+ """Notify that the given channel was deleted.
+
+ This is used for removing deleted channels from functions.
+ """
+ for f in self.functions.values():
+ if f.type == SCENE and c in f.scope:
+ f.delete_channel(c)
+
+ def delete_callbacks(self, owner, f: int = None):
+ """Remove all callbacks registered by the owner.
+
+ :param f: the function to remove from (all if None)
+ """
+ if isinstance(f, Function):
+ f = f.id
+ for g in self._change_callbacks:
+ if f is None or g == f:
+ self._change_callbacks[g] = [i for i in self._change_callbacks[g] if i[0] != owner]
+ for g in self._delete_callbacks:
+ if f is None or g == f:
+ self._delete_callbacks[g] = [i for i in self._delete_callbacks[g] if i[0] != owner]
+ self._topology_callbacks = [i for i in self._topology_callbacks if i[0] != owner]
+
+ def register_function_change_callback(self, f: int, callback, owner = None):
+ """Register a callback for a function change.
+
+ :param f: the ID of the function to monitor
+ :param callback: a function to call, accepting the Function
+ :param owner: optional ID to use for removal of the callback later
+ """
+ if isinstance(f, Function):
+ f = f.id
+ if f not in self._change_callbacks:
+ self._change_callbacks[f] = []
+ self._change_callbacks[f].append((owner, callback))
+
+ def register_function_delete_callback(self, f: int, callback, owner = None):
+ """Register a callback for a function deletion.
+
+ :param f: the ID of the function to monitor
+ :param callback: a function to call, accepting the Function
+ :param owner: optional ID to use for removal of the callback later
+ """
+ if isinstance(f, Function):
+ f = f.id
+ if f not in self._delete_callbacks:
+ self._delete_callbacks[f] = []
+ self._delete_callbacks[f].append((owner, callback))
+
+ def register_topology_callback(self, callback, owner = None):
+ """Register a callback for a topology change.
+
+ For simplicity, this is fired whenever any change to the topology is made. As
+ functions is general do not care about the topology (only the channels, the
+ deletion of which is handled elsewhere) this function is not concerned about
+ performance.
+
+ :param callback: the function to call, accepting the Workspace
+ :param owner: optional ID to use for removal of the callback later
+ """
+ self._topology_callbacks.append((owner, callback))
+
+ def topology_changed(self):
+ """Call when a topology change has been made."""
+ for f in self._topology_callbacks.values():
+ f(self)
+
+ def function_changed(self, f: Function):
+ """Called when a function is changed.
+
+ :param f: the changed function
+ """
+ for callback in self._change_callbacks[f.id]:
+ callback(f)
+
+ def function_deleted(self, f: Function):
+ """Called when a function is deleted.
+
+ This also handles removing the function from the Workspace.
+ """
+ for callback in self._delete_callbacks[f.id]:
+ callback(f)
+
+ self.delete_callbacks(f)
+
+ del self.functions[f.id]
+
+ @classmethod
+ def load(cls, filename):
+ """Load the workspace from a file."""
+ tree = et.parse(filename)
+ root = tree.getroot()
+
+ return cls.deserialize(None, root)
+
+ @classmethod
+ def deserialize(cls, w, e):
+ from .topology import Fixture
+ from .functions.scene import Scene
+ from .functions.audio import Audio
+
+ if e.tag != BXW+"workspace":
+ raise LoadError("Root tag must be workspace")
+
+ try:
+ name, author, version, modified, fixtures, functions = e
+ except ValueError:
+ raise LoadError("Invalid workspace layout")
+
+ ## First load the metadata so we can create the workspace
+ name = name.text
+ author = author.text
+ version = int(version.text)
+ modified = fromisoformat(modified.text)
+
+ w = cls(name=name, author=author, version=version, modified=modified)
+
+ ## Now load the fixtures
+ for fixture in fixtures:
+ Fixture.deserialize(w, fixture)
+
+ ## Finally, load the functions
+ for function in functions:
+ type_ = function.get("type")
+ if type_ == AUDIO:
+ Audio.deserialize(w, function)
+ elif type_ == SCENE:
+ Scene.deserialize(w, function)
+ else:
+ raise LoadError("Unknown function type \"%s\"" % type_)
+
+ return w
+
+ def serialize(self) -> et.Element:
+ root = et.Element(BXW+"workspace")
+
+ et.SubElement(root, BXW+"name").text = self.name
+ et.SubElement(root, BXW+"author").text = self.author
+ et.SubElement(root, BXW+"version").text = str(self.version)
+ et.SubElement(root, BXW+"modified").text = dt.datetime.now().isoformat()
+
+ fixtures = et.SubElement(root, BXW+"fixtures")
+ for n, fixture in enumerate(self.fixtures.values()):
+ fe = fixture.serialize()
+ fixtures.insert(n, fe)
+
+ functions = et.SubElement(root, BXW+"functions")
+ for n, function in enumerate(self.functions.values()):
+ fe = function.serialize()
+ functions.insert(n, fe)
+
+ return root
+
+
+ def save(self, filename):
+ """Save the workspace to a file."""
+ et.register_namespace("", BXW.strip("{}"))
+ root = self.serialize()
+ XMLSerializable.indent(root)
+ tree = et.ElementTree(element=root)
+ tree.write(filename, encoding="unicode")
diff --git a/constants.py b/constants.py
@@ -1,13 +0,0 @@
-"""Constants module.
-
-Contains some constants used throughout.
-"""
-
-INFTY = -1
-AUTO = -2
-
-BXW = "{http://unsuspicious.services/bxw}"
-
-SCENE = "Scene"
-AUDIO = "Audio"
-CHASER = "Chaser"
diff --git a/functions/audio.py b/functions/audio.py
@@ -1,118 +0,0 @@
-"""Audio function module.
-
-Contains the definition of the Audio, the audio primitive.
-"""
-
-import xml.etree.ElementTree as et
-
-from .function import Function
-
-from ..constants import AUDIO, BXW
-from ..exceptions import LoadError
-
-class Audio(Function):
- """Class representing an audio cue.
-
- This is the primitive for audio, and all sound cues must be based in some manner off of
- Audio. This function merely plays a single audio file once, starting at t=0.
-
- The duration of the audio is automatically determined and is zero if the file does not
- exist or is unsupported by ffmpeg.
- """
- scope = ()
- audio_scope = frozenset()
- actual_duration = 0
-
- def __init__(self, w, id_ = None, name = None, fade_in = 0, fade_out = 0,
- filename: str = None):
- super().__init__(w=w, type_=AUDIO, id_=id_, name=name, duration=0,
- fade_in=fade_in, fade_out=fade_out)
-
- self._filename = filename
-
- if filename is not None:
- self.duration = self.w.get_audio_length(filename)
- self.audio_scope = frozenset(((filename,),))
- else:
- self.duration = 0
- self.audio_scope = frozenset()
-
- self.actual_duration = self.duration
-
- def get_data(self):
- return None
-
- def copy_data(self, data):
- return None
-
- def render(self, t, data = None):
- return ((),
- ((self.id, self._filename, 0, self.fade_in, self.duration, self.fade_out),),
- None)
-
- @property
- def filename(self):
- """Return the current audio filename."""
- return self._filename
-
- @filename.setter
- def filename(self, value: str):
- """Set the current audio filename, updating the duration."""
- if value is not None:
- self.duration = self.w.get_audio_length(value)
- self.audio_scope = frozenset(((value,),))
- else:
- self.duration = 0
- self.audio_scope = frozenset()
-
- self._filename = value
- self.w.function_changed(self)
-
- def serialize(self) -> et.Element:
- e = et.Element(BXW+"function")
- e.set("type", self.type)
- e.set("id", str(self.id))
- e.set("name", self.name)
- e.set("fade-in", str(self.fade_in))
- e.set("fade-out", str(self.fade_out))
- if self.filename is not None:
- filename = et.SubElement(e, BXW+"filename")
- filename.text = self.filename
-
- return e
-
- @classmethod
- def deserialize(cls, w, e):
- if e.tag != BXW+"function":
- raise LoadError("Invalid function tag")
- elif e.get("type") != AUDIO:
- raise LoadError("Load delegated to wrong class (this is a bug)")
-
- id_ = cls.int_or_none(e.get("id"))
- if id_ is None:
- raise LoadError("Function tag has invalid/missing ID")
-
- name = e.get("name")
-
- fade_in = e.get("fade-in")
- try:
- fade_in = int(fade_in) if fade_in else 0
- except ValueError:
- raise LoadError("Invalid fade in")
-
- fade_out = e.get("fade-out")
- try:
- fade_out = int(fade_out) if fade_out else 0
- except ValueError:
- raise LoadError("Invalid fade out")
-
- if len(e) > 1:
- raise LoadError("Audio tag can have at most one filename")
- elif len(e) == 1:
- filename, = e
- filename = filename.text
- else:
- filename = None
-
- return cls(w=w, id_=id_, name=name, filename=filename, fade_in=fade_in,
- fade_out=fade_out)
diff --git a/functions/function.py b/functions/function.py
@@ -1,88 +0,0 @@
-"""Base function module.
-
-Contains the generic Function interface.
-"""
-
-from abc import ABCMeta, abstractmethod, abstractproperty
-from typing import Set, Any
-
-from ..topology import Fixture
-from ..constants import INFTY
-from ..interfaces import XMLSerializable
-
-class Function(XMLSerializable, metaclass=ABCMeta):
- """Class representing a generic function.
-
- Many of the properties here should not be implemented as properties for performance
- reasons.
-
- Functions and properties required for rendering, e.g. ``Function.render`` and
- ``Function.actual_duration``, should be written to be performant.
-
- Any change to the scope, audio scope, or actual duration of the function must be
- reported using ``Workspace.function_changed`` with the new values. This should also be
- extended to any change in e.g. values for lighting and audio primitives.
- """
- def __init__(self, w: "Workspace", type_: str, id_: int = None, name: str = None,
- duration: int = INFTY, fade_in: int = 0, fade_out: int = 0):
- self.w = w
- self.id = id_ if id_ is not None else w.next_function_id
- self.name = name if name else "%s %s" % (type_, self.id)
- self.type = type_
- self.duration = duration
- self.fade_in = fade_in
- self.fade_out = fade_out
-
- self.w.register_function(self)
-
- def delete(self):
- """Delete the function from the Workspace."""
- self.w.function_deleted(self)
-
- @abstractproperty
- def scope(self) -> Set[Fixture.Channel]:
- """Return the set of channels affected by this function."""
-
- @abstractproperty
- def audio_scope(self) -> Set[str]:
- """Return the set of audio filenames that may be used by this function."""
-
- @abstractmethod
- def get_data(self) -> Any:
- """Return the default data for this function."""
-
- @abstractmethod
- def copy_data(self, data: Any) -> Any:
- """Duplicate the given data."""
-
- @abstractproperty
- def actual_duration(self):
- """Return the actual duration of the function, including all fades."""
-
- @abstractmethod
- def render(self, t: int, data: Any = None):
- """Render the function at the given time.
-
- This function must return a 3-tuple:
-
- (light_cues, audio_cues, new_data)
-
- Where ``light_cues`` is a iterable of (channel, value) pairs. It is an error for
- ``light_cues`` to contain channels other than exactly this Function's scope.
- ``audio_cues`` is an iterable of audio cues of the form:
-
- (guid, filename, start_time, fade_in, end_time, fade_out)
-
- The ``guid`` is globally unique: that is, it is unique to that specific audio cue,
- even if the same file is played multiple times or in different places. Note that
- ``end_time`` may be ``INFTY``, in which case the file should be played to its end.
- In the case of infinite-duration chasers, the ``end_time`` may change over
- subsequent calls to this function.
-
- Note that ``data`` is not necessarily mutable: ``render`` should be called like:
-
- lights, sound, data = f.render(t, data)
-
- :param t: the time to render at, in milliseconds
- :param data: the function data to use
- """
diff --git a/functions/scene.py b/functions/scene.py
@@ -1,144 +0,0 @@
-"""Scene function module.
-
-Contains the definition of the Scene, the lighting primitive.
-"""
-
-import xml.etree.ElementTree as et
-from typing import Mapping
-
-from .function import Function
-from ..topology import Fixture
-from ..constants import INFTY, SCENE, BXW
-from ..exceptions import LoadError
-
-class Scene(Function):
- """Class representing a lighting scene.
-
- This is the primitive for lighting, and all light cues must be based in some manner off
- of Scene. Scenes are simple, having no fading and infinite duration.
-
- Modifying the scene can be done in a few ways: the easiest is using the dictionary-
- style accesses, e.g. setting the values using:
-
- scene[channel] = value
-
- Values may be removed from the scene using the ``del`` operator and retrieved in the
- same fashion. Alternatively, values can be updated in bulk using ``Scene.update`` or
- set in bulk with ``Scene.set``.
- """
- audio_scope = ()
- scope = frozenset()
- actual_duration = INFTY
- values = frozenset()
-
- def __init__(self, w, id_ = None, name = None, values: Mapping[Fixture.Channel, int] = None):
- super().__init__(w=w, type_=SCENE, id_=id_, name=name, duration=INFTY, fade_in=0,
- fade_out=0)
- self._values = {}
- self._render = ()
-
- self._update(values, changed=False)
-
- def _update_render(self, changed=True):
- self._render = tuple(self._values.items())
- self.scope = frozenset(self._values.keys())
- self.values = self._render
-
- if changed:
- self.w.function_changed(self)
-
- def set(self, v):
- """Set the scene's values to the given ones.
-
- Equivalent to update, except that all non-given values are deleted from the scene.
- """
- self._values = {}
- self.update(v)
-
- def _update(self, v, changed=True):
- if isinstance(v, dict):
- v = v.items()
- vn = []
- for c, val in v:
- val = int(val)
- if val < 0 or val > 255:
- raise ValueError("Values must be integers on [0,256)")
- vn.append((c, val))
- for c, val in vn:
- self._values[c] = val
- self._update_render(changed=changed)
-
- def update(self, v):
- """Update the scene's values."""
- self._update(v=v, changed=True)
-
- def __getitem__(self, c: Fixture.Channel):
- return self._values[c] if c in self._values else None
-
- def __setitem__(self, c: Fixture.Channel, v: int):
- self.update(((c, v),))
-
- def __delitem__(self, c: Fixture.Channel):
- if c in self._values:
- del self._values[c]
- self._update_render()
-
- def get_data(self):
- return None
-
- def copy_data(self, data):
- return None
-
- def render(self, t, data = None):
- return (self._render, (), None)
-
- def serialize(self):
- e = et.Element(BXW+"function")
- e.set("type", self.type)
- e.set("id", str(self.id))
- e.set("name", self.name)
- for c, v in self.values:
- ce = et.SubElement(e, BXW+"value")
- ce.set("fixture", str(c.f.id))
- ce.set("channel", str(c.id))
- ce.text = str(v)
-
- return e
-
- @classmethod
- def deserialize(cls, w, e):
- if e.tag != BXW+"function":
- raise LoadError("Invalid function tag")
- elif e.get("type") != SCENE:
- raise LoadError("Load delegated to wrong class (this is a bug)")
-
- id_ = cls.int_or_none(e.get("id"))
- if id_ is None:
- raise LoadError("Function tag has invalid/missing ID")
-
- name = e.get("name")
-
- values = {}
- for ve in e:
- if ve.tag != BXW+"value":
- raise LoadError("Invalid value tag")
-
- fixture = cls.int_or_none(ve.get("fixture"))
- channel = cls.int_or_none(ve.get("channel"))
-
- if None in (fixture, channel):
- raise LoadError("Missing/invalid fixture/channel value")
- elif fixture not in w.fixtures:
- raise LoadError("Missing fixture ID %d" % fixture)
- elif channel >= len(w.fixtures[fixture].channels):
- raise LoadError("Fixture %d missing channel ID %d" % (fixture, channel))
-
- channel = w.fixtures[fixture].channels[channel]
-
- value = cls.int_or_none(ve.text)
- if value is None:
- raise LoadError("Missing/invalid value for channel")
-
- values[channel] = value
-
- return cls(w=w, id_=id_, name=name, values=values)
diff --git a/topology.py b/topology.py
@@ -1,135 +0,0 @@
-"""Topology module.
-
-Contains the topology classes for representing the physical and virtual topology of the
-lighting setup.
-"""
-
-import xml.etree.ElementTree as et
-
-from .constants import BXW
-from .interfaces import XMLSerializable
-from .exceptions import LoadError
-
-class Fixture(XMLSerializable):
- """Class representing a lighting fixture.
-
- Each lighting fixture has a number of channels, which are mapped to the physical
- topology.
-
- Channels must be added by changing the value of ``Fixture.channel_count``, which will
- create the necessary channel objects and handle deletion.
- """
- def __init__(self, w: "Workspace", id_: int = None, name: str = None, channel_count: int = 0):
- self.w = w
- self.id = id_ if id_ is not None else w.next_fixture_id
- self.name = name if name else "Fixture %d" % self.id
-
- self.channels = ()
- self.channel_count = channel_count
-
- self.w.register_fixture(self)
-
- class Channel:
- """Class representing a single channel on a Fixture.
-
- The physical address of the channel is stored in the ``Channel.address`` attribute
- and may be changed at will.
-
- This program takes a very lax approach to channel addressing: the address is not
- parsed in any way and duplicate addresses are not handled. For convenience with
- OLA, Channel is iterable and iteration returns an iterator over its ``address``
- attribute.
- """
- def __init__(self, f: "Fixture", id_: int, name: str = "Intensity", address = (-1,-1)):
- self.f = f
- self.id = id_
- self.name = name
- self.address = address
- self._hash = hash((f.id, id_))
-
- def __hash__(self):
- return self._hash
-
- def __iter__(self):
- return iter(self.address)
-
- def __repr__(self):
- return "Channel(fixture={c.f.id}, index={c.id}, name={c.name})".format(c=self)
-
- @property
- def channel_count(self):
- """Return the current number of channels on the fixture."""
- return len(self.channels)
-
- @channel_count.setter
- def channel_count(self, value):
- """Change the number of channels on the fixture.
-
- This function handles deletion of removed channels from functions as well.
- """
- if value < 0:
- raise ValueError("Number of channels must be nonnegative")
- elif value < len(self.channels):
- for i in range(value, len(self.channels)):
- self.w.delete_channel(self.channels[i])
- self.channels = self.channels[:value]
- elif value > len(self.channels):
- self.channels += tuple((Fixture.Channel(self, i) for i in range(len(self.channels), value)))
-
- def __repr__(self):
- return "Fixture({f.name}, id={f.id}, channels={f.channel_count})".format(f=self)
-
- def serialize(self):
- e = et.Element(BXW+"fixture")
- e.set("name", self.name)
- e.set("id", str(self.id))
-
- for c in self.channels:
- ce = et.SubElement(e, BXW+"channel")
- ce.set("name", c.name)
- if c.address is not None:
- ## TODO: Other addressing modes
- try:
- univ, addr = c.address
- ae = et.SubElement(ce, BXW+"ola")
- ae.set("universe", str(univ))
- ae.set("address", str(addr))
- except ValueError:
- pass
-
- return e
-
- @classmethod
- def deserialize(cls, w, e):
- if e.tag != BXW+"fixture":
- raise LoadError("Invalid fixture tag")
-
- id_ = cls.int_or_none(e.get("id"))
- if id_ is None:
- raise LoadError("Fixture tag has invalid/missing ID")
-
- name = e.get("name")
-
- f = cls(w, id_=id_, name=name, channel_count=len(e))
- for n, channel in enumerate(e):
- if channel.tag != BXW+"channel":
- raise LoadError("Invalid channel tag")
-
- name = channel.get("name")
- if name is not None:
- f.channels[n].name = name
-
- if len(channel) > 1:
- raise LoadError("Channel can have at most one address")
- elif len(channel) == 1:
- address, = channel
- if address.tag == BXW+"ola":
- try:
- address = (int(address.get("universe")), int(address.get("address")),)
- except (ValueError, TypeError):
- raise LoadError("Invalid OLA address on channel")
- else:
- raise LoadError("Unknown address tag \"%s\"" % address.tag)
- f.channels[n].address = address
-
- return f
diff --git a/workspace.py b/workspace.py
@@ -1,248 +0,0 @@
-"""Workspace module.
-
-Contains the main Workspace implementation.
-"""
-
-import datetime as dt
-import json
-import subprocess as subp
-import xml.etree.ElementTree as et
-
-from .constants import AUDIO, SCENE, BXW
-from .functions.function import Function
-from .exceptions import LoadError
-from .interfaces import XMLSerializable
-
-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.
-
- Parameters:
- f: the path to check
- path: the path of ffprobe
- """
- try:
- a = subp.check_output([path, "-show_format", "-print_format", "json", f], stderr=subp.DEVNULL)
- except subp.CalledProcessError:
- return 0
- return int(1000*float(json.loads(a)["format"]["duration"])+0.5)
-
-class Workspace(XMLSerializable):
- """Class representing a audiovisual workspace."""
- def __init__(self, name: str, author: str, version: int, modified: dt.datetime):
- self.name = name
- self.author = author
- self.version = version
- self.modified = modified
-
- self.fixtures = {}
- self._last_fixture_id = -1
- self.functions = {}
- self._last_function_id = -1
-
- self._audio_lengths = {}
-
- self._change_callbacks = {}
- self._delete_callbacks = {}
-
- def get_audio_length(self, filename: str) -> int:
- """Determine the audio length of the given file.
-
- This value is returned from the cache, if available.
- """
- if filename not in self._audio_lengths:
- self._audio_lengths[filename] = ffprobe_audio_length(filename)
- return self._audio_lengths[filename]
-
- def recheck_audio_length(self, filename: str) -> int:
- """Determine the audio length of the given file.
-
- This function re-probes the value, updating the cache result and the durations of
- Audio functions.
- """
- self._audio_lengths[filename] = ffprobe_audio_length(filename)
- for f in self.functions.values():
- if f.type == AUDIO and f.filename == filename:
- f.duration = self._audio_lengths[f.filename]
- return self._audio_lengths[filename]
-
- def recheck_audio_lengths(self):
- """Recheck and update all audio lengths."""
- for filename in self._audio_lengths:
- self._audio_lengths[filename] = ffprobe_audio_length(filename)
- for f in self.functions.values():
- if f.type == AUDIO and f.filename in self._audio_lengths:
- f.filename = f.filename
-
- def register_fixture(self, f: "Fixture"):
- """Register the fixture in the Workspace.
-
- Always called when the fixture is instantiated.
- """
- if f.id in self.fixtures:
- raise ValueError("A fixture with that ID already exists")
- self.fixtures[f.id] = f
-
- def register_function(self, f: Function):
- """Register the function in the Workspace.
-
- Always called when the function is instantiated.
- """
- if f.id in self.functions:
- raise ValueError("A function with that ID already exists")
- self.functions[f.id] = f
-
- @property
- def next_fixture_id(self):
- """Return the next fixture ID."""
- return self._last_fixture_id + 1
-
- @property
- def next_function_id(self):
- """Return the next function ID."""
- return self._last_function_id + 1
-
- def delete_channel(self, c: "Fixture.Channel"):
- """Notify that the given channel was deleted.
-
- This is used for removing deleted channels from functions.
- """
- for f in self.functions.values():
- if f.type == SCENE and c in f.scope:
- f.delete_channel(c)
-
- def delete_callbacks(self, owner, f: int = None):
- """Remove all callbacks registered by the owner.
-
- :param f: the function to remove from (all if None)
- """
- if isinstance(f, Function):
- f = f.id
- for g in self._change_callbacks:
- if f is None or g == f:
- self._change_callbacks[g] = [i for i in self._change_callbacks[g] if i[0] != owner]
- for g in self._delete_callbacks:
- if f is None or g == f:
- self._delete_callbacks[g] = [i for i in self._delete_callbacks[g] if i[0] != owner]
-
- def register_function_change_callback(self, f: int, callback, owner = None):
- """Register a callback for a function change.
-
- :param f: the ID of the function to monitor
- :param callback: a function to call, accepting the Function
- :param owner: optional ID to use for removal of the callback later
- """
- if isinstance(f, Function):
- f = f.id
- if f not in self._change_callbacks:
- self._change_callbacks[f] = []
- self._change_callbacks[f].append((owner, callback))
-
- def register_function_delete_callback(self, f: int, callback, owner = None):
- """Register a callback for a function deletion.
-
- :param f: the ID of the function to monitor
- :param callback: a function to call, accepting the Function
- :param owner: optional ID to use for removal of the callback later
- """
- if isinstance(f, Function):
- f = f.id
- if f not in self._delete_callbacks:
- self._delete_callbacks[f] = []
- self._delete_callbacks[f].append((owner, callback))
-
- def function_changed(self, f: Function):
- """Called when a function is changed.
-
- :param f: the changed function
- """
- for callback in self._change_callbacks[f.id]:
- callback(f)
-
- def function_deleted(self, f: Function):
- """Called when a function is deleted.
-
- This also handles removing the function from the Workspace.
- """
- for callback in self._delete_callbacks[f.id]:
- callback(f)
-
- self.delete_callbacks(f)
-
- del self.functions[f.id]
-
- @classmethod
- def load(cls, filename):
- """Load the workspace from a file."""
- tree = et.parse(filename)
- root = tree.getroot()
-
- return cls.deserialize(None, root)
-
- @classmethod
- def deserialize(cls, w, e):
- from .topology import Fixture
- from .functions.scene import Scene
- from .functions.audio import Audio
-
- if e.tag != BXW+"workspace":
- raise LoadError("Root tag must be workspace")
-
- try:
- name, author, version, modified, fixtures, functions = e
- except ValueError:
- raise LoadError("Invalid workspace layout")
-
- ## First load the metadata so we can create the workspace
- name = name.text
- author = author.text
- version = int(version.text)
- modified = dt.datetime.fromisoformat(modified.text)
-
- w = cls(name=name, author=author, version=version, modified=modified)
-
- ## Now load the fixtures
- for fixture in fixtures:
- Fixture.deserialize(w, fixture)
-
- ## Finally, load the functions
- for function in functions:
- type_ = function.get("type")
- if type_ == AUDIO:
- Audio.deserialize(w, function)
- elif type_ == SCENE:
- Scene.deserialize(w, function)
- else:
- raise LoadError("Unknown function type \"%s\"" % type_)
-
- return w
-
- def serialize(self) -> et.Element:
- root = et.Element(BXW+"workspace")
-
- et.SubElement(root, BXW+"name").text = self.name
- et.SubElement(root, BXW+"author").text = self.author
- et.SubElement(root, BXW+"version").text = str(self.version)
- et.SubElement(root, BXW+"modified").text = dt.datetime.now().isoformat()
-
- fixtures = et.SubElement(root, BXW+"fixtures")
- for n, fixture in enumerate(self.fixtures.values()):
- fe = fixture.serialize()
- fixtures.insert(n, fe)
-
- functions = et.SubElement(root, BXW+"functions")
- for n, function in enumerate(self.functions.values()):
- fe = function.serialize()
- functions.insert(n, fe)
-
- return root
-
-
- def save(self, filename):
- """Save the workspace to a file."""
- et.register_namespace("", BXW.strip("{}"))
- root = self.serialize()
- XMLSerializable.indent(root)
- tree = et.ElementTree(element=root)
- tree.write(filename, encoding="unicode")