Replace fixed DSL fields with parameterized metric expressions

app/stats_expressions.py

@@ -0,0 +1,110 @@
+"""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)