111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
"""Parameterized metric expressions, lowered to private calculation names."""
|
||
|
||
import re
|
||
|
||
from .stats_dsl import NUMBER, QueryError, parse_dsl
|
||
|
||
FUNCTIONS = {
|
||
'best_time': {'example': 'best_time(5 km)', 'description': 'Fastest recorded elapsed effort at a distance'},
|
||
'days_since_run': {'example': 'days_since_run(distance >= 10 km)', 'description': 'Local days since a run matching a condition; empty parentheses match any run'},
|
||
'rolling_sum': {'example': 'rolling_sum(distance, 28 days)', 'description': 'Trailing sum before filters'},
|
||
'rolling_mean': {'example': 'rolling_mean(heart_rate, 14 days)', 'description': 'Trailing mean before filters'},
|
||
'rolling_count': {'example': 'rolling_count(distance, 7 days)', 'description': 'Count usable inputs in a trailing window'},
|
||
}
|
||
CALL = re.compile(r'\b(' + '|'.join(FUNCTIONS) + r')\s*\(')
|
||
|
||
|
||
def lower(text):
|
||
declarations, names, lines = {}, {}, []
|
||
for number, line in enumerate(text.splitlines(), 1):
|
||
output, index = [], 0
|
||
while index < len(line):
|
||
if line[index] == '"':
|
||
end = index + 1
|
||
while end < len(line):
|
||
if line[end] == '\\':
|
||
end += 2
|
||
elif line[end] == '"':
|
||
end += 1
|
||
break
|
||
else:
|
||
end += 1
|
||
output.append(line[index:end])
|
||
index = end
|
||
continue
|
||
match = CALL.match(line, index) if not line.lstrip().startswith('#') else None
|
||
if not match:
|
||
output.append(line[index])
|
||
index += 1
|
||
continue
|
||
end, depth, quoted, escaped = match.end(), 1, False, False
|
||
while end < len(line):
|
||
char = line[end]
|
||
if escaped:
|
||
escaped = False
|
||
elif char == '\\' and quoted:
|
||
escaped = True
|
||
elif char == '"':
|
||
quoted = not quoted
|
||
elif not quoted:
|
||
depth += (char == '(') - (char == ')')
|
||
if depth == 0:
|
||
break
|
||
end += 1
|
||
if depth:
|
||
raise QueryError('Close the metric expression with ).', number)
|
||
# Preserve whitespace in quoted title conditions.
|
||
parts = re.split(r'("(?:[^"\\]|\\.)*")', line[index:end + 1])
|
||
for part in range(0, len(parts), 2):
|
||
parts[part] = re.sub(r'\s+', ' ', parts[part])
|
||
expression = re.sub(r'^(\w+)\s+\(', r'\1(', ''.join(parts).strip())
|
||
if expression not in names:
|
||
names[expression] = f'analysis_expr_{len(names)}'
|
||
declarations.setdefault(number, []).append((names[expression], expression))
|
||
output.append(names[expression])
|
||
index = end + 1
|
||
lines.append(''.join(output))
|
||
return '\n'.join(lines), declarations
|
||
|
||
|
||
def calculation(name, expression, catalog, units, defaults, line):
|
||
if match := re.fullmatch(r'best_time\(\s*(' + NUMBER + r')\s+(km|m)\s*\)', expression):
|
||
distance = float(match[1]) * (1000 if match[2] == 'km' else 1)
|
||
if distance not in {1000, 2000, 5000, 10000}:
|
||
raise QueryError('Recorded best times support 1, 2, 5 or 10 km.', line)
|
||
config, unit, conversions = {'best_distance': distance}, 'time', {'s': 1, 'min': 60}
|
||
elif match := re.fullmatch(r'days_since_run\(\s*(.*?)\s*\)', expression):
|
||
condition = match[1]
|
||
if not condition:
|
||
conditions = []
|
||
else:
|
||
try:
|
||
checked = parse_dsl('runs\nwhere ' + condition + '\nsummarize count',
|
||
catalog, {**defaults, 'minimum': None}, units)
|
||
except QueryError as error:
|
||
raise QueryError(str(error), line) from None
|
||
conditions = list(checked['filters'])
|
||
for field, key, op in [('distance', 'minimum', '>='), ('distance', 'maximum', '<='),
|
||
('heart_rate', 'hr_min', '>='), ('heart_rate', 'hr_max', '<=')]:
|
||
if checked[key] is not None:
|
||
conditions.append({'field': field, 'op': op, 'value': checked[key]})
|
||
config, unit, conversions = {'since': conditions}, 'days', {'days': 1, 'h': 1 / 24}
|
||
elif match := re.fullmatch(r'rolling_(sum|mean|count)\(\s*(\w+)\s*,\s*(\d+) days\s*\)', expression):
|
||
function, field, days = match.groups()
|
||
if field not in catalog or not 1 <= int(days) <= 365:
|
||
raise QueryError('Rolling functions need an earlier metric and 1–365 days.', line)
|
||
unit = 'runs' if function == 'count' else catalog[field][1]
|
||
conversions = {'runs': 1} if function == 'count' else units[field]
|
||
config = {'window': {'function': function, 'field': field, 'days': int(days)}}
|
||
else:
|
||
raise QueryError('Use best_time(5 km), days_since_run(distance >= 10 km), or rolling_sum(distance, 28 days).', line)
|
||
return {'name': name, 'expression': expression, 'unit': unit, 'inline': True, **config}, conversions
|
||
|
||
|
||
def render(text, specs):
|
||
expressions = {s['name']: s['expression'] for s in specs if s.get('inline')}
|
||
# Private names are substituted only outside quoted labels and title fragments.
|
||
parts = re.split(r'("(?:[^"\\]|\\.)*")', text)
|
||
for index in range(0, len(parts), 2):
|
||
parts[index] = re.sub(r'\banalysis_expr_\d+\b', lambda m: expressions.get(m[0], m[0]), parts[index])
|
||
return ''.join(parts)
|