diff options
Diffstat (limited to 'workspace.py')
| -rw-r--r-- | workspace.py | 248 |
1 files changed, 248 insertions, 0 deletions
diff --git a/workspace.py b/workspace.py new file mode 100644 index 0000000..add047b --- /dev/null +++ b/workspace.py @@ -0,0 +1,248 @@ +"""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") |