294 lines
9.0 KiB
Python
294 lines
9.0 KiB
Python
# 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}"'
|
|
|
|
def namespaced_key(x):
|
|
if x is None:
|
|
return (1, x)
|
|
else:
|
|
return (0, x)
|
|
|
|
class Timings:
|
|
def __init__(self):
|
|
self._timings = []
|
|
self._markers = {}
|
|
self._marker_objects = {}
|
|
self._marker_buttons = []
|
|
self._marker_buttons_prestart = {}
|
|
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 is_button_prestart(self, name):
|
|
if name in self._marker_buttons_prestart:
|
|
return True
|
|
|
|
return False
|
|
|
|
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, prestart=False):
|
|
self._marker_objects[name] = at
|
|
if at is None:
|
|
self._marker_buttons.append(name)
|
|
if prestart:
|
|
self._marker_buttons_prestart[name] = True
|
|
self._resolved = False
|
|
|
|
def register_tempo(self, tempo, beats, at):
|
|
self._tempo_objects.append((tempo, beats, at))
|
|
self._resolved = False
|
|
|
|
def resolve_tempo(self):
|
|
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]
|
|
|
|
def resolve(self, final=False):
|
|
try:
|
|
self.resolve_tempo()
|
|
for name in self._marker_objects.keys():
|
|
if self._marker_objects[name] is not None:
|
|
self._markers[name] = self._marker_objects[name].evaluate()
|
|
self.resolve_tempo()
|
|
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
|
|
|