From 8fa467cb93c09bedd4bd0a557af67530c2107c50 Mon Sep 17 00:00:00 2001 From: Gavan Fantom Date: Tue, 5 Aug 2025 01:10:28 +0100 Subject: [PATCH] Initial code commit --- README.md | 111 ++++++++++++++- display.py | 31 +++++ events.py | 155 +++++++++++++++++++++ factory.py | 47 +++++++ gig.py | 398 +++++++++++++++++++++++++++++++++++++++++++++++++++++ item.py | 53 +++++++ timings.py | 279 +++++++++++++++++++++++++++++++++++++ 7 files changed, 1073 insertions(+), 1 deletion(-) create mode 100755 display.py create mode 100644 events.py create mode 100644 factory.py create mode 100755 gig.py create mode 100644 item.py create mode 100644 timings.py diff --git a/README.md b/README.md index b45346c..485a66f 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,111 @@ -# gig +# gig.py + + ./gig.py [-w] show.yaml + +Use the -w flag to run in regular windows rather than full screen. + +## YAML + +Define your set with: + + set: + - item1 + - item2 + - item3 + +Define your preshow with: + + preshow: + - item4 + +Define your items: + + items: + item1: + title: Title + events: + - Display: + media: File.jpg + at: select + - Lyrics: + text: Text + at: start + +## Time markers + +Events are scheduled to happen at times. Times can be specified in terms of markers, and modified by bars, beats and hours/minutes/seconds. + +There are two predefined markers: + +* select - defined as the time the item is selected. This is typically used for an introduction to a piece. +* start - defined as the time the item is started. This is typically when backing tracks are played. + +Other markers can be defined for convenience. + +Every time specification must include at least one marker. The marker may be modified using addition or subtraction. + +The following expressions can be combined: + +* ` bars` - number of bars at the current tempo. The keywords `bar` and `bars` are both accepted. +* ` beats` - number of beats at the current tempo. The keywords `beat` and `beats` are both accepted. +* `` - number of seconds +* `:` - number of minutes and seconds +* `::` - number of hours, minutes and seconds + +Examples: + + at: start + 4 bars - 1 beat + at: select + 1:23 + +## Event types + +### Display + +Displays image for the audience +(not yet implemented) + +* at - time to display the image +* media - filename of the image + +### Lyrics + +Display lyrics for the audience + +* at - time to display the text +* text - text to display +* duration - (not yet implemented) + +### Playback + +Play audio + +* at - time to display the text +* media - filename to play +* device - output device to use (optional) +* channels - output channel map (optional) +* tempo - initial tempo (bpm) (optional) +* beats - initials beats in a bar (optional) +* leadin - time in seconds before the first beat (optional) (not yet implemented) +* loop - (True/False) loop at end of playback (optional) (not yet implemented) + +### SetTempo + +Set the tempo at a given time + +* at - time at which to set the tempo +* tempo - new tempo (bpm) (optional) +* beats - new beats in a bar (optional) + +### SetMark + +Set time marker + +* at - time to set the marker to +* name - name of marker + +### ButtonMark + +Set time marker based on external keypress + +* name - name of marker diff --git a/display.py b/display.py new file mode 100755 index 0000000..ac1561f --- /dev/null +++ b/display.py @@ -0,0 +1,31 @@ +# display.py + +import pyglet + +class DisplayContent: + def __init__(self): + pass + + def create(self, obj, width, height, x, y): + obj.audiencebatch = pyglet.graphics.Batch() + obj.audiencecontent = [] + + +class DisplayLyrics(DisplayContent): + def __init__(self, text): + self._text = text + + def create(self, obj, width, height, x, y): + scale = height / 600 + + super().create(obj, width, height, x, y) + + obj.audiencecontent.append(pyglet.text.Label(self._text, + font_name='Ariel', + font_size=36 * scale, + x=x + width//2, y=y + height//2, + anchor_x='center', anchor_y='center', + width=width, align='center', + multiline=True, + batch=obj.audiencebatch)) + diff --git a/events.py b/events.py new file mode 100644 index 0000000..d476818 --- /dev/null +++ b/events.py @@ -0,0 +1,155 @@ +# events.py + +import soundfile +import sounddevice + +import timings +import display + +import numpy as np + +class Event: + def __init__(self, at='start'): + self._at = timings.Timing(at) + + def set_timings(self, timings): + self._at.set_timings(timings) + + def set_audiencecontent(self, audiencecontent): + self.audiencecontent = audiencecontent + + @property + def scheduled(self): + return self._at.scheduled + + @scheduled.setter + def scheduled(self, value): + self._at.scheduled = value + + @property + def at(self): + return self._at.at + + def prepare(self): + pass + + def start(self): + pass + + def stop(self): + pass + +class Display(Event): + def __init__(self, media, *args, **kwargs): + super().__init__(*args, **kwargs) + self._media = media + + +class Playback(Event): + def __init__(self, media, device=None, channels=[1,2], tempo=None, beats=None, leadin=0, loop=False, *args, **kwargs): + super().__init__(*args, **kwargs) + self._media = media + self._device = device + self._mapping = np.atleast_1d(np.array(channels, copy=True)) + self._channels = self._mapping.max() + self._mapping -= 1 + self._tempo = tempo + self._beats = beats + self._leadin = leadin + self._loop = loop + + def prepare(self): + self._at._timings.register_tempo(self._tempo, self._beats, self._at) + +# def start(self): +# print("Start playback") +# data, fs = soundfile.read(self._media) +# sounddevice.play(data, fs, device=self._device, mapping=self._mapping) + + def start(self): + print("Start playback") + self._data, fs = soundfile.read(self._media) + self._current_frame = 0 + + self._stream = sounddevice.OutputStream( + samplerate=fs, device=self._device, channels=self._channels, + callback=self.callback, finished_callback=self.finished_callback) + self._stream.start() + + def callback(self, outdata, frames, time, status): + if status: + print(status) + chunksize = min(len(self._data) - self._current_frame, frames) + outdata[:chunksize, self._mapping] = self._data[self._current_frame:self._current_frame + chunksize] + if chunksize < frames: + outdata[chunksize:, self._mapping] = 0 + raise sounddevice.CallbackStop() + self._current_frame += chunksize + + def finished_callback(self): + self._data = None + print("Playback finished") + + def stop(self): + print("Stop playback") + self._stream.stop(True) + self._stream.close(True) + self._stream = None + +class SetTempo(Event): + def __init__(self, tempo=None, beats=None, *args, **kwargs): + super().__init__(*args, **kwargs) + self._tempo = tempo + self._beats = beats + + def prepare(self): + self._at._timings.register_tempo(self._tempo, self._beats, self._at) + + def start(self): + pass + + def stop(self): + pass + +class SetMark(Event): + def __init__(self, name, *args, **kwargs): + super().__init__(*args, **kwargs) + self._name = name + + def prepare(self): + self._at._timings.register_mark(self._name, self._at) + + def start(self): + pass + + def stop(self): + pass + +class ButtonMark(Event): + def __init__(self, name, *args, **kwargs): + super().__init__(*args, **kwargs) + self._name = name + + def prepare(self): + self._at._timings.register_mark(self._name, None) + + def start(self): + pass + + def stop(self): + pass + +class Lyrics(Event): + def __init__(self, text, duration=None, *args, **kwargs): + super().__init__(*args, **kwargs) + self._text = text + self._duration = duration + + def start(self): + self.audiencecontent.new_content(display.DisplayLyrics(self._text)) + print("Started displaying lyrics") + print(self._text) + + def stop(self): + self.audiencecontent.new_content(display.DisplayContent()) + print("Stopped displaying lyrics") diff --git a/factory.py b/factory.py new file mode 100644 index 0000000..0f97be2 --- /dev/null +++ b/factory.py @@ -0,0 +1,47 @@ +# factory.py + +import events +from item import Item + +class MakeItems: + def __new__(cls, config): + items = {} + for item in config: + kwargs = {} + for key in config[item].keys(): + if key == 'events': + kwargs[key] = MakeEvents(config[item][key]) + else: + kwargs[key] = config[item][key] + items[item] = Item(**kwargs) + if hasattr(items[item], 'prepare'): + items[item].prepare() + return items + +class MakeEvents: + def __new__(cls, config): + objects = [] + for x in config: + for key in x: + args = [] + kwargs = {} + if isinstance(x[key], list): + args = x[key] + elif isinstance(x[key], dict): + kwargs = x[key] + elif x[key] is None: + pass + else: + args[0] = x[key] + obj = getattr(events, key)(*args, **kwargs) + objects.append(obj) + return objects + +# This just returns an array for now, we'll see if we need to make it into a class later. +class MakeSet: + def __new__(cls, config, items): + for item in config: + if item not in items: + raise KeyError('{} not in items'.format(item)) + return config + diff --git a/gig.py b/gig.py new file mode 100755 index 0000000..13ac216 --- /dev/null +++ b/gig.py @@ -0,0 +1,398 @@ +#!/usr/bin/python3 + +import argparse + +import sounddevice +import soundfile + +import pyglet +from pyglet.window import key + +import factory +import display + +import yaml +import time + +import dbus + +class EventRunner: + def __init__(self, gigconfig): + self._gigconfig = gigconfig + self._running = False + self._selected = None + self._playing = None + self._active = [] + self._currentitem = None + + def set_audiencecontent(self, audiencecontent): + for item in self._gigconfig['items']: + self._gigconfig['items'][item].set_audiencecontent(audiencecontent) + + def select(self, pos): + self._selected = pos + self._playing = None + self.activate('select') + + def play(self): + self._playing = self._selected + self._selected = None + self.activate('start') + + def advance(self): + if self._playing is not None: + return self.activate() + elif self._selected is not None: + self.play() + return True + return False + + def activate(self, marker=None): + newitem = self.get_item() + if self._currentitem != newitem: + selected = self._selected + self.stop() + newitem.reset() + self._currentitem = newitem + if marker is None: + marker = 'select' + elif marker != 'select': + return False + self._selected = selected + now = time.time() + if not self._currentitem.activate(marker, now): + return False + for event in self._currentitem._events: + if event.at is not None and not event.scheduled: + pyglet.clock.schedule_once(self.update, event.at - now, event) + print("Scheduled event {} at {} ({} from now)".format(event, event.at, event.at - now)) + event.scheduled = True + self._next_button = self._currentitem.get_next_button() + return True + + def stop(self): + pyglet.clock.unschedule(self.update) + self._playing = None + self._selected = None + for event in self._active: + event.stop() + self._active = [] + if self._currentitem is not None: + self._currentitem.reset() + + def update(self, dt, event): + event.start() + self._active.append(event) + + def pause(self): + pass + + def get_item(self, pos=None): + if pos is None: + pos = self._playing + if pos is None: + pos = self._selected + if pos is None: + return None + item = self._gigconfig['setlist'][pos] + return self._gigconfig['items'][item] + + def status(self): + if self._playing is not None: + status = 'Playing {}'.format(self.get_item().get_title()) + if self._next_button is not None: + button = "Marker {}".format(self._next_button) + else: + button = "Select next" + elif self._selected is not None: + status = 'Selected {}'.format(self.get_item().get_title()) + button = "Play" + else: + status = 'Waiting' + button = "Select" + return { + 'selected' : self._selected, + 'playing' : self._playing, + 'running' : self._running, + 'status' : status, + 'button' : button, + } + +class MainWindow(pyglet.window.Window): + def __init__(self, gigconfig, eventrunner, audiencecontent, *args, **kwargs): + super().__init__(*args, **kwargs) + self._gigconfig = gigconfig + self._eventrunner = eventrunner + self._audiencecontent = audiencecontent + self._audiencecontent.register(self) + + self._cursorpos = 0 + self._playpos = None + self._selectpos = None + + self.on_resize(self.width, self.height) + + self.update_cursor(0) + + def on_resize(self, width, height): + + self._scale = self.height / 600 + + self.batch = pyglet.graphics.Batch() + + self.status = pyglet.text.Label('Status bar', + font_name='Ariel', + font_size=18 * self._scale, + x=self.width//2, y=0, + anchor_x='left', anchor_y='bottom', + batch=self.batch) + + self.next_action = pyglet.text.Label('Next:', + font_name='Ariel', + font_size=18 * self._scale, + x=self.width//2, y=40 * self._scale, + anchor_x='left', anchor_y='bottom', + batch=self.batch) + + self._spacing = 40 * self._scale + self._cursorspread = 300 * self._scale + self._cursorwidth = 15 * self._scale + self._cursorheight = 20 * self._scale + + y_increment = self._spacing + y_counter = self.height - y_increment + + # They're really buttons, but... + self.buttons = [] + for item in self._gigconfig['setlist']: + self.buttons.append(pyglet.text.Label(self._gigconfig['items'][item].get_title(), + font_name='Ariel', + font_size=14 * self._scale, + width=self._cursorspread, + x=self.width//4, y=y_counter, + anchor_x='center', anchor_y='center', + batch=self.batch)) + y_counter -= y_increment + + self._cursor1x = self.width//4 - self._cursorspread/2 + self._cursor2x = self.width//4 + self._cursorspread/2 + + self._cursory = self.height//2 + + self.cursor1 = pyglet.shapes.Triangle(self._cursor1x, self._cursory, + self._cursor1x-self._cursorwidth, self._cursory-self._cursorheight/2, + self._cursor1x-self._cursorwidth, self._cursory+self._cursorheight/2, + (255, 255, 255, 255), + batch=self.batch) + + self.cursor2 = pyglet.shapes.Triangle(self._cursor2x, self._cursory, + self._cursor2x+self._cursorwidth, self._cursory-self._cursorheight/2, + self._cursor2x+self._cursorwidth, self._cursory+self._cursorheight/2, + (255, 255, 255, 255), + batch=self.batch) + + self.update_objects() + self.audience_update() + + def audience_update(self): + self.switch_to() + self._audiencecontent.create(self, self.width // 2, self.height // 2, x=self.width//2, y=self.height//4) + + def update_objects(self): + status = self._eventrunner.status() + self._selectpos = status['selected'] + self._playpos = status['playing'] + self.set_status(status['status']) + self.set_next_action(status['button']) + self.update_cursor() + + def update_cursor(self, pos=None): + if pos is not None: + self._cursorpos = pos + y_increment = self._spacing + y_counter = self.height // 2 + y_increment * self._cursorpos + for i, button in enumerate(self.buttons): + button.y = y_counter + if i == self._playpos: + button.color = (0, 255, 0, 255) + elif i == self._selectpos: + button.color = (255, 255, 0, 255) + else: + button.color = (255, 255, 255, 255) + y_counter -= y_increment + + def select(self): + if self._playpos == self._cursorpos: + if not self._eventrunner.advance(): + if self._cursorpos < len(self.buttons) - 1: + self.update_cursor(self._cursorpos + 1) + self._eventrunner.select(self._cursorpos) + elif self._selectpos == self._cursorpos: + self._eventrunner.play() + else: + self._eventrunner.select(self._cursorpos) + self.update_objects() + + def set_status(self, text): + self.status.text = text + + def set_next_action(self, text): + self.next_action.text = "Next: {}".format(text) + + def action_up(self): + if self._cursorpos > 0: + self.update_cursor(self._cursorpos - 1) + + def action_down(self): + if self._cursorpos < len(self.buttons) - 1: + self.update_cursor(self._cursorpos + 1) + + def action_select(self): + self.select() + + def action_stop(self): + if self._playpos is not None: + self._eventrunner.stop() + elif self._selectpos is not None: + self._eventrunner.stop() + else: + self.action_up() + self.update_objects() + + def on_key_press(self, symbol, modifiers): + if symbol == key.UP: + self.action_up() + elif symbol == key.DOWN: + self.action_down() + elif symbol == key.ENTER: + self.action_select() + elif symbol == key.SPACE: + self.action_stop() + else: + super().on_key_press(symbol, modifiers) + + def on_draw(self): + self.clear() + self.batch.draw() + self.audiencebatch.draw() + + +class AudienceContent: + def __init__(self, gigconfig, eventrunner): + self._gigconfig = gigconfig + self._eventrunner = eventrunner + self._objects = [] + self._eventrunner.set_audiencecontent(self) + self._display = display.DisplayContent() + + def register(self, obj): + self._objects.append(obj) + + def update(self): + for obj in self._objects: + obj.audience_update() + + def create(self, obj, width, height, x=0, y=0): + self._display.create(obj, width, height, x, y) + + def new_content(self, obj): + self._display = obj + self.update() + +class AudienceWindow(pyglet.window.Window): + def __init__(self, audiencecontent, *args, **kwargs): + super().__init__(*args, **kwargs) + + self._audiencecontent = audiencecontent + self._audiencecontent.register(self) + + self._audiencecontent.create(self, self.width, self.height) + + def audience_update(self): + self.switch_to() + self._audiencecontent.create(self, self.width, self.height) + + def on_resize(self, width, height): + print("New size is: {} x {}".format(width, height)) + self._audiencecontent.create(self, width, height) + + def on_draw(self): + self.clear() + self.audiencebatch.draw() + +def process_config(config): + items = factory.MakeItems(config['items']) + setlist = factory.MakeSet(config['set'], items) + preshow = factory.MakeSet(config['preshow'], items) + if 'defaults' in config: + defaults = config['defaults'] + else: + defaults = {} + return {'items' : items, 'setlist' : setlist, 'preshow' : preshow} + +def load_config(filename): + with open(filename, encoding='utf8') as file: + y = yaml.safe_load(file) + config = process_config(y) + return config + +def playback(filename): + data, fs = soundfile.read(filename) + #sounddevice.play(data, fs, device='alsa,hw:CARD=QU16') + sounddevice.play(data, fs, device='QU-16 alsa', mapping=[21,22]) + sounddevice.wait() + +def main(): + parser = argparse.ArgumentParser(prog='gig.py', description='Gig showrunning') + parser.add_argument('filename') + parser.add_argument('-w', '--window', action='store_true', help='Run in a window') + args = parser.parse_args() + config = load_config(args.filename) + display = pyglet.display.get_display() + screens = display.get_screens() +# print(screens) + mainscreen = screens[0] + if len(screens) > 1: + audiencescreen = screens[1] + else: + audiencescreen = None + print(mainscreen) + print(audiencescreen) + eventrunner = EventRunner(gigconfig=config) + + audiencecontent = AudienceContent(gigconfig=config, eventrunner=eventrunner) + + if audiencescreen is not None: + window2 = AudienceWindow(audiencecontent=audiencecontent, fullscreen=False, screen=audiencescreen, resizable=True) + if not args.window: + window2.set_location(audiencescreen.x, audiencescreen.y) + window2.set_size(audiencescreen.width, audiencescreen.height) + window2.set_mouse_position(audiencescreen.x, audiencescreen.y) + window2.set_fullscreen(True) + #window2.set_location(audiencescreen.x, audiencescreen.y) + #event_logger = pyglet.window.event.WindowEventLogger() + #window2.push_handlers(event_logger) + + window = MainWindow(gigconfig=config, eventrunner=eventrunner, audiencecontent=audiencecontent, fullscreen=False, screen=mainscreen, resizable=True) + if not args.window: + window.set_location(mainscreen.x, mainscreen.y) + window.set_size(mainscreen.width, mainscreen.height) + window.set_mouse_position(mainscreen.x, mainscreen.y) + window.set_fullscreen(True) + #window.set_location(mainscreen.x, mainscreen.y) + + try: + # Let's set up some stuff to inhibit the screensaver + bus = dbus.SessionBus() + saver = bus.get_object('org.freedesktop.ScreenSaver', '/ScreenSaver') + saver_interface = dbus.Interface(saver, dbus_interface='org.freedesktop.ScreenSaver') + + # now we can inhibit the screensaver + cookie=saver_interface.Inhibit("gig.py", "Don't blank the screen during the gig") + except Exception: + pass + + pyglet.app.run() + +if __name__ == "__main__": + main() diff --git a/item.py b/item.py new file mode 100644 index 0000000..10a20d3 --- /dev/null +++ b/item.py @@ -0,0 +1,53 @@ +# item.py + +import timings + +class Item: + def __init__(self, title, events): + self._title = title + self._events = events + self._timings = timings.Timings() + for event in self._events: + event.set_timings(self._timings) + event.prepare() + self.resolve() + + def reset(self): + self._timings.reset() + self.resolve() + + def resolve(self): + self._timings.resolve() + self._timings.resolve(final=True) + def eventkey(x): + x = x.at + if x is None: + return (1, x) + else: + return (0, x) + self._events.sort(key=eventkey) +# print("Item {}".format(self._title)) +# for event in self._events: +# print("Event at time {}".format(event.at)) + + def activate(self, marker, time): + if marker is None: + marker = self._timings.get_button() + if marker is None: + return False + self._timings.set_marker(marker, time) + self.resolve() + return True + + def get_next_button(self): + return self._timings.get_button() + + def get_title(self): + return self._title + + def get_timings(self): + return self._timings + + def set_audiencecontent(self, audiencecontent): + for event in self._events: + event.set_audiencecontent(audiencecontent) diff --git a/timings.py b/timings.py new file mode 100644 index 0000000..5a06ca3 --- /dev/null +++ b/timings.py @@ -0,0 +1,279 @@ +# timings.py + +import re + +class TimingError(Exception): + def __init__(self, message, at=None): + self.message = message + self.at = at + super().__init__(self.message) + + def __str__(self): + return f'{self.message} in string "{self.at}"' + +class Timings: + def __init__(self): + self._timings = [] + self._markers = {} + self._marker_objects = {} + self._marker_buttons = [] + self._tempo_objects = [] + self._tempo = [] + self._resolved = False + + def reset(self): + self._markers = {} + self._resolved = False + for timing in self._timings: + timing.reset() + + def set_marker(self, marker, time): + self._markers[marker] = time + + def get_button(self): + for button in self._marker_buttons: + if button not in self._markers: + return button + return None + + def add_timing(self, timing): + self._timings.append(timing) + self._resolved = False + + def lookup(self, marker): + return self._markers[marker] + + def evaluate(self, value, modifier, context): + if modifier in {'bar', 'bars'}: + return self.bar_at(context) * value + elif modifier in {'beat', 'beats'}: + return self.beat_at(context) * value + else: + raise ValueError('Invalid modifier: {}'.format(modifier)) + + def tempo_at(self, at): + # Some nice defaults. Maybe these should live elsewhere? + tempo = 120 + beats = 4 + if at is not None: + for (t, b, tat) in self._tempo: + if tat is not None and tat <= at: + tempo = t + beats = b + #print("tempo_at({}) = ({}, {})".format(at, tempo, beats)) + return (tempo, beats) + + def bar_at(self, at): + (tempo, beats) = self.tempo_at(at) + #print("bar_at({}) = {}".format(at, 60 * beats / tempo)) + return 60 * beats / tempo + + def beat_at(self, at): + (tempo, beats) = self.tempo_at(at) + #print("beat_at({}) = {}".format(at, 60 / tempo)) + return 60 / tempo + + def register_mark(self, name, at): + self._marker_objects[name] = at + if at is None: + self._marker_buttons.append(name) + self._resolved = False + + def register_tempo(self, tempo, beats, at): + self._tempo_objects.append((tempo, beats, at)) + self._resolved = False + + def resolve(self, final=False): + try: + def namespaced_key(x): + if x is None: + return (1, x) + else: + return (0, x) + self._tempo_objects.sort(key=lambda x: namespaced_key(x[2].evaluate())) + self._tempo = [(t, b, a.evaluate()) for (t, b, a) in self._tempo_objects] + for name in self._marker_objects.keys(): + if self._marker_objects[name] is not None: + self._markers[name] = self._marker_objects[name].evaluate() + self._timings.sort(key=lambda x: namespaced_key(x.evaluate())) + except Exception: + if final: + raise + self._resolved = True + +class BinOp: + def __init__(self, op, a=None, b=None): + self.op = op + self.a = a + self.b = b + + def combine(self, b): + if self.b is not None: + raise TimingError('Binary operation already has a rhs') + self.b = b + + def __str__(self): + return 'BinOp({} {} {})'.format(str(self.a), self.op, str(self.b)) + +class ImmediateNumber: + def __init__(self, n): + self._n = [float(n)] + self._modifier = None + + def add_part(self, n): + if len(self._n) >= 2: + raise TimingError('Too many parts in HH:MM:SS') + if self._modifier is not None: + raise TimingError('Can\'t modify HH:MM:SS with bar/beats') + self._n.append(float(n)) + + def add_modifier(self, modifier): + if len(self._n) > 1: + raise TimingError('bar/beat modifier can not be applied to HH:MM:SS') + if self._modifier is not None: + raise TimingError('bar/beat modifier can not be applied twice to the same number') + self._modifier = modifier + + def type(self): + if self._modifier is not None: + return self._modifier + else: + return 'time' + + def value(self): + if self._modifier is not None: + return (self._n[0], self._modifier) + else: + value = 0 + for n in self._n: + value *= 60 + value += n + return value + + def __str__(self): + if self._modifier is not None: + return 'ImmediateNumber({} {})'.format(':'.join(self._n), self._modifier) + else: + return 'ImmediateNumber({})'.format(':'.join(self._n)) + + +class Timing: + def __init__(self, at): + self._at = str(at) + self._at_cached = None + self._scheduled = False + self.parse() + + def reset(self): + self._at_cached = None + self._scheduled = False + + @property + def scheduled(self): + return self._scheduled + + @scheduled.setter + def scheduled(self, value): + self._scheduled = value + + def set_timings(self, timings): + self._timings = timings + self._timings.add_timing(self) + + def parse(self): + def collapse(stack): + if len(stack) == 0: + return stack + if len(stack) > 2: + raise TimingError('Invalid expression') + if isinstance(stack[0], BinOp): + if stack[1] == ':': + raise TimingError('Invalid expression') + stack[0].combine(stack[1]) + stack.pop(1) + if stack[0] == ':': + raise TimingError('Invalid expression') + if len(stack) > 1: + raise TimingError('Invalid expression') + return stack + + modifiers = {'bar', 'bars', 'beat', 'beats'} + operators = {'+', '-'} + tokens = self.tokenise() + stack = [] + try: + for token in tokens: + if token in operators: + a = collapse(stack) + if len(a) != 1: + raise TimingError('Binary operation must have valid lhs') + op = BinOp(token, a=a[0]) + stack = [op] + elif token in modifiers: + if len(stack) < 1: + raise TimingError('Modifier without value', self._at) + if isinstance(stack[-1], ImmediateNumber): + stack[-1].add_modifier(token) + else: + raise TimingError('Modifier can only follow number', self._at) + elif re.match(r'\d+\.\d+|\d+', token): + if (len(stack) >= 2) and (stack[-1] == ':'): + if isinstance(stack[-2], ImmediateNumber): + stack[-2].add_part(token) + stack.pop() + else: + raise TimingError('HH:MM:SS must start with a number', self._at) + else: + stack.append(ImmediateNumber(token)) + else: + stack.append(token) + stack = collapse(stack) + except TimingError as e: + e.at = self._at + raise + if len(stack) != 1: + raise TimingError('Invalid expression', self._at) + self._expr = stack[0] + + def tokenise(self): + pattern = r'\d+\.\d+|\d+|\w+|[+\-:]' + #print(self._at) + tokens = re.findall(pattern, self._at) + tokens = [x for x in tokens if x != ''] + #print(tokens) + return tokens + + @property + def at(self): + if self._at_cached is None: + self._at_cached = self.evaluate() + return self._at_cached + + def evaluate(self, tree=None, context=None): + if tree is None: + tree = self._expr + if isinstance(tree, BinOp): + a = self.evaluate(tree.a, context) + b = self.evaluate(tree.b, a) + if a is None: + return None + if b is None: + return None + if tree.op == '+': + return a + b + elif tree.op == '-': + return a - b + else: + raise ValueError('op must be + or -') + elif isinstance(tree, ImmediateNumber): + if tree.type() == 'time': + return tree.value() + else: + (value, modifier) = tree.value() + return self._timings.evaluate(value, modifier, context) + else: + try: + return self._timings.lookup(tree) + except KeyError: + return None +