Initial code commit
This commit is contained in:
279
timings.py
Normal file
279
timings.py
Normal file
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user