189 lines
8.0 KiB
Python
189 lines
8.0 KiB
Python
"""Small, explicit running query language with source-located diagnostics."""
|
||
|
||
import difflib
|
||
import json
|
||
import math
|
||
import re
|
||
from datetime import date
|
||
|
||
NUMBER = r"(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?"
|
||
UNITS = {
|
||
"distance": {"km": 1, "m": 0.001},
|
||
"duration": {"h": 1, "min": 1 / 60, "s": 1 / 3600},
|
||
"pace": {"/km": 1},
|
||
"heart_rate": {"bpm": 1},
|
||
"cadence": {"spm": 1},
|
||
"power": {"W": 1},
|
||
"elevation": {"m": 1},
|
||
"ground_contact": {"ms": 1},
|
||
"stride_length": {"m": 1},
|
||
}
|
||
|
||
|
||
class QueryError(ValueError):
|
||
def __init__(self, message, line=1):
|
||
super().__init__(message)
|
||
self.line = line
|
||
|
||
|
||
def parse_dsl(text, metrics, defaults):
|
||
if not isinstance(text, str) or len(text) > 4000:
|
||
raise QueryError("Keep the query under 4,000 characters.")
|
||
q = dict(defaults)
|
||
q["filters"] = []
|
||
q["extra_metrics"] = []
|
||
seen = set()
|
||
locations = {}
|
||
|
||
def claim(key, line):
|
||
if key in seen:
|
||
raise QueryError(f"Repeated {key} clause. Keep one per query.", line)
|
||
seen.add(key)
|
||
locations[key] = line
|
||
|
||
def metric(key, line):
|
||
if key not in metrics:
|
||
suggestion = difflib.get_close_matches(key, metrics, n=1)
|
||
hint = (
|
||
f" Did you mean '{suggestion[0]}'?"
|
||
if suggestion
|
||
else " Open the metric reference for supported names."
|
||
)
|
||
raise QueryError(f"Unknown metric '{key}'.{hint}", line)
|
||
|
||
def amount(key, raw, unit, line):
|
||
if unit not in UNITS[key]:
|
||
raise QueryError(f"Use {' or '.join(UNITS[key])} for {key}.", line)
|
||
if key == "pace" and ":" in raw:
|
||
minutes, seconds = raw.split(":")
|
||
if not minutes.isdecimal() or not seconds.isdecimal() or int(seconds) >= 60:
|
||
raise QueryError("Write pace like 5:30 /km (seconds from 00 to 59).", line)
|
||
value = int(minutes) * 60 + int(seconds)
|
||
else:
|
||
try:
|
||
value = float(raw)
|
||
except ValueError:
|
||
raise QueryError("Use a numeric value with its unit.", line) from None
|
||
if not math.isfinite(value) or not 0 <= value <= 1000000:
|
||
raise QueryError("Filter values must be finite and between 0 and 1,000,000.", line)
|
||
return value * UNITS[key][unit]
|
||
|
||
def condition(key, op, value, line):
|
||
legacy = {
|
||
("distance", ">="): "minimum",
|
||
("distance", "<="): "maximum",
|
||
("heart_rate", ">="): "hr_min",
|
||
("heart_rate", "<="): "hr_max",
|
||
}.get((key, op))
|
||
if legacy:
|
||
if value > 1000:
|
||
raise QueryError("Distance and heart rate bounds must be at most 1,000.", line)
|
||
old = q[legacy]
|
||
q[legacy] = value if old is None else max(old, value) if op == ">=" else min(old, value)
|
||
locations[legacy] = line
|
||
else:
|
||
q["filters"].append({"field": key, "op": op, "value": value})
|
||
|
||
for line_number, source in enumerate(text.splitlines(), 1):
|
||
line = source.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if line == "runs":
|
||
claim("runs", line_number)
|
||
elif match := re.fullmatch(r"during last (\d+) (days?|weeks?|months?)", line):
|
||
claim("during", line_number)
|
||
count, unit = int(match[1]), match[2].rstrip("s")
|
||
limit = {"day": 732, "week": 104, "month": 24}[unit]
|
||
if not 1 <= count <= limit:
|
||
raise QueryError(f"Use 1–{limit} {unit}s.", line_number)
|
||
if unit == "week":
|
||
q["weeks"] = count
|
||
else:
|
||
q["days" if unit == "day" else "months"] = count
|
||
elif match := re.fullmatch(r"during (\d{4}-\d{2}-\d{2}) to (\d{4}-\d{2}-\d{2})", line):
|
||
claim("during", line_number)
|
||
try:
|
||
q["start"], q["end"] = date.fromisoformat(match[1]), date.fromisoformat(match[2])
|
||
if not 0 <= (q["end"] - q["start"]).days < 732:
|
||
raise ValueError
|
||
except ValueError:
|
||
raise QueryError(
|
||
"Use valid dates in order, covering at most 732 days.", line_number
|
||
) from None
|
||
elif match := re.fullmatch(r"ending (\d{4}-\d{2}-\d{2})", line):
|
||
claim("ending", line_number)
|
||
if q["start"]:
|
||
raise QueryError(
|
||
"An explicit date range already supplies its end date.", line_number
|
||
)
|
||
try:
|
||
q["end"] = date.fromisoformat(match[1])
|
||
except ValueError:
|
||
raise QueryError("Use a valid end date, YYYY-MM-DD.", line_number) from None
|
||
elif line.startswith("summarize "):
|
||
claim("summarize", line_number)
|
||
keys = [key.strip() for key in line[10:].split(",")]
|
||
if not 1 <= len(keys) <= 6 or len(set(keys)) != len(keys):
|
||
raise QueryError("Choose 1–6 different metrics, separated by commas.", line_number)
|
||
for key in keys:
|
||
metric(key, line_number)
|
||
q["metric"] = keys[0]
|
||
q["secondary"] = keys[1] if len(keys) > 1 else "none"
|
||
q["extra_metrics"] = keys[2:]
|
||
elif match := re.fullmatch(r"by (day|week|month)", line):
|
||
claim("by", line_number)
|
||
q["group"] = match[1]
|
||
elif match := re.fullmatch(r"compare (previous period|none)", line):
|
||
claim("compare", line_number)
|
||
q["compare"] = match[1] == "previous period"
|
||
elif match := re.fullmatch(r"sort by (\w+) (asc|desc)", line):
|
||
claim("sort", line_number)
|
||
if match[1] != "date":
|
||
metric(match[1], line_number)
|
||
q["sort"] = [match[1], match[2]]
|
||
elif match := re.fullmatch(r'(?:where|and) title contains (".*")', line):
|
||
try:
|
||
title = json.loads(match[1])
|
||
if not isinstance(title, str) or not 1 <= len(title) <= 160:
|
||
raise ValueError
|
||
except ValueError:
|
||
raise QueryError(
|
||
'Use a title fragment in double quotes, e.g. "River".', line_number
|
||
) from None
|
||
q["filters"].append({"field": "title", "op": "contains", "value": title})
|
||
elif match := re.fullmatch(r"(?:where|and) (has|missing) (\w+)", line):
|
||
metric(match[2], line_number)
|
||
q["filters"].append({"field": match[2], "op": match[1]})
|
||
elif match := re.fullmatch(r"(?:where|and) (\w+) between (\S+) and (\S+) (\S+)", line):
|
||
key, low, high, unit = match.groups()
|
||
metric(key, line_number)
|
||
lower, upper = amount(key, low, unit, line_number), amount(key, high, unit, line_number)
|
||
if lower > upper:
|
||
raise QueryError("The lower bound must not exceed the upper bound.", line_number)
|
||
condition(key, ">=", lower, line_number)
|
||
condition(key, "<=", upper, line_number)
|
||
elif match := re.fullmatch(r"(?:where|and) (\w+) (>=|<=|>|<|=|!=) (\S+) (\S+)", line):
|
||
key, op, raw, unit = match.groups()
|
||
metric(key, line_number)
|
||
value = amount(key, raw, unit, line_number)
|
||
condition(key, op, value, line_number)
|
||
else:
|
||
raise QueryError(
|
||
"Unrecognised clause. Use autocomplete (Ctrl+Space) or a starting query.",
|
||
line_number,
|
||
)
|
||
if len(q["filters"]) > 20:
|
||
raise QueryError("Use at most 20 filter conditions.", line_number)
|
||
if not {"runs", "summarize"} <= seen:
|
||
raise QueryError("Include 'runs' and a 'summarize' clause.")
|
||
if "ending" in seen and q["start"]:
|
||
raise QueryError(
|
||
"Use either an explicit date range or an ending clause.", locations["during"]
|
||
)
|
||
for low, high in (("minimum", "maximum"), ("hr_min", "hr_max")):
|
||
if q[low] is not None and q[high] is not None and q[low] > q[high]:
|
||
raise QueryError(
|
||
"A filter's minimum cannot exceed its maximum.", locations.get(high, 1)
|
||
)
|
||
return q
|