544 lines
22 KiB
Python
544 lines
22 KiB
Python
"""Typed analysis expressions and extensions for the run query language."""
|
||
|
||
import ast
|
||
import math
|
||
import re
|
||
import statistics
|
||
|
||
from .stats_dsl import UNITS, QueryError, parse_dsl
|
||
|
||
FUNCTIONS = {
|
||
"latest",
|
||
"sum",
|
||
"mean",
|
||
"weighted_mean",
|
||
"median",
|
||
"min",
|
||
"max",
|
||
"stddev",
|
||
"count",
|
||
"p10",
|
||
"p25",
|
||
"p75",
|
||
"p90",
|
||
"p95",
|
||
"p99",
|
||
}
|
||
GROUPS = {"day", "week", "month", "year", "weekday", "hour", "run", "all"}
|
||
EXTRA_METRICS = {
|
||
"pace_variability": (
|
||
"Speed variability",
|
||
"%",
|
||
"Duration-weighted speed coefficient of variation across usable moving recording intervals. Lower means steadier recorded speed.",
|
||
),
|
||
"heart_rate_drift": (
|
||
"Speed/HR decoupling",
|
||
"%",
|
||
"Percent reduction in speed-to-HR efficiency from the first to second moving-time half. Positive means lower efficiency later. Requires four moving minutes and 80% HR coverage in each half; includes warm-up and terrain changes.",
|
||
),
|
||
"sensor_coverage": (
|
||
"Moving HR coverage",
|
||
"%",
|
||
"Percent of usable moving recording time with positive recorded heart rate. Gaps over ten seconds and pauses are excluded.",
|
||
),
|
||
"gap": (
|
||
"Grade-adjusted pace",
|
||
"/km",
|
||
"Modelled grade-adjusted pace from persisted recording analysis using Minetti running cost. Missing terrain stays missing.",
|
||
),
|
||
"count": ("Runs", "runs", "Number of runs."),
|
||
"speed": ("Speed", "km/h", "Recorded distance divided by timer duration."),
|
||
"gap_since_previous_run": (
|
||
"Gap since previous run",
|
||
"days",
|
||
"Elapsed start-to-start time since your previous run, including runs outside this query. This is not time since the previous run ended.",
|
||
),
|
||
"distance_change": (
|
||
"Distance change",
|
||
"km",
|
||
"Distance minus the previous run's distance, before filtering.",
|
||
),
|
||
"pace_change": (
|
||
"Pace change",
|
||
"sec/km",
|
||
"Pace minus the previous run's pace. Negative means faster.",
|
||
),
|
||
"heart_rate_change": (
|
||
"Heart rate change",
|
||
"bpm",
|
||
"Average heart rate minus the previous run's average, before filtering.",
|
||
),
|
||
"run_day_streak": (
|
||
"Consecutive running days",
|
||
"days",
|
||
"Consecutive local calendar days with a run, through this run's local date, before filters.",
|
||
),
|
||
"elevation_per_km": (
|
||
"Ascent per kilometre",
|
||
"m/km",
|
||
"Recorded ascent divided by recorded distance.",
|
||
),
|
||
"max_heart_rate": ("Maximum heart rate", "bpm", "Recorded maximum heart rate."),
|
||
"vertical_oscillation": (
|
||
"Vertical oscillation",
|
||
"cm",
|
||
"Recorded sample average, weighted by run timer duration.",
|
||
),
|
||
"calories": ("Calories", "kcal", "Recorded session calories. Missing values are excluded."),
|
||
"training_effect": (
|
||
"Aerobic training effect",
|
||
"score",
|
||
"Recorded Garmin aerobic training effect; an unweighted mean across runs.",
|
||
),
|
||
}
|
||
EXTRA_UNITS = {
|
||
"pace_variability": {"%": 1},
|
||
"heart_rate_drift": {"%": 1},
|
||
"sensor_coverage": {"%": 1},
|
||
"gap": {"/km": 1},
|
||
"count": {"runs": 1},
|
||
"speed": {"km/h": 1, "m/s": 3.6},
|
||
"gap_since_previous_run": {"days": 1, "h": 1 / 24, "min": 1 / 1440},
|
||
"distance_change": {"km": 1, "m": 0.001},
|
||
"pace_change": {"sec/km": 1},
|
||
"heart_rate_change": {"bpm": 1},
|
||
"run_day_streak": {"days": 1},
|
||
"elevation_per_km": {"m/km": 1},
|
||
"max_heart_rate": {"bpm": 1},
|
||
"vertical_oscillation": {"cm": 1},
|
||
"calories": {"kcal": 1},
|
||
"training_effect": {"score": 1},
|
||
}
|
||
|
||
|
||
from .training_metrics import METRICS as TRAINING_METRICS
|
||
from .training_metrics import UNITS as TRAINING_UNITS
|
||
|
||
EXTRA_METRICS.update(TRAINING_METRICS)
|
||
EXTRA_UNITS.update(TRAINING_UNITS)
|
||
|
||
|
||
def valid(value):
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
|
||
|
||
|
||
def expression_tree(source, catalog, line=1):
|
||
if len(source) > 200:
|
||
raise QueryError("Keep each calculation under 200 characters.", line)
|
||
try:
|
||
tree = ast.parse(source, mode="eval")
|
||
except SyntaxError:
|
||
raise QueryError("Use metric names, numbers, parentheses and + - * /.", line) from None
|
||
if len(list(ast.walk(tree))) > 64:
|
||
raise QueryError("Simplify this calculation.", line)
|
||
|
||
def unit(node):
|
||
if isinstance(node, ast.Expression):
|
||
return unit(node.body)
|
||
if isinstance(node, ast.Name) and node.id in catalog:
|
||
# Pace values inside arithmetic are seconds per km, not a formatted string.
|
||
return (
|
||
"sec/km"
|
||
if catalog[node.id][1] == "/km"
|
||
else "h"
|
||
if node.id == "duration"
|
||
else catalog[node.id][1]
|
||
)
|
||
if isinstance(node, ast.Constant) and valid(node.value) and abs(node.value) <= 1000000:
|
||
return "ratio"
|
||
if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)):
|
||
return unit(node.operand)
|
||
if isinstance(node, ast.BinOp) and isinstance(
|
||
node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)
|
||
):
|
||
left, right = unit(node.left), unit(node.right)
|
||
if isinstance(node.op, (ast.Add, ast.Sub)):
|
||
if left != right:
|
||
raise QueryError(
|
||
f"Cannot add/subtract {left} and {right}. Use matching units.", line
|
||
)
|
||
return left
|
||
if isinstance(node.op, ast.Div) and left == right:
|
||
return "ratio"
|
||
if right == "ratio":
|
||
return left
|
||
if left == "ratio" and isinstance(node.op, ast.Mult):
|
||
return right
|
||
if isinstance(node.op, ast.Div) and (left, right) in {("km", "h"), ("m", "km")}:
|
||
return f"{left}/{right}"
|
||
return f"({left}{'*' if isinstance(node.op, ast.Mult) else '/'}{right})"
|
||
raise QueryError(
|
||
"Calculations allow existing metrics, numbers and + - * /. Calls, attributes and indexing are unavailable.",
|
||
line,
|
||
)
|
||
|
||
return tree, unit(tree)
|
||
|
||
|
||
def evaluate(tree, values):
|
||
def visit(node):
|
||
if isinstance(node, ast.Expression):
|
||
return visit(node.body)
|
||
if isinstance(node, ast.Name):
|
||
return values.get(node.id)
|
||
if isinstance(node, ast.Constant):
|
||
return node.value
|
||
if isinstance(node, ast.UnaryOp):
|
||
value = visit(node.operand)
|
||
return None if not valid(value) else -value if isinstance(node.op, ast.USub) else value
|
||
left, right = visit(node.left), visit(node.right)
|
||
if not valid(left) or not valid(right):
|
||
return None
|
||
if isinstance(node.op, ast.Add):
|
||
result = left + right
|
||
elif isinstance(node.op, ast.Sub):
|
||
result = left - right
|
||
elif isinstance(node.op, ast.Mult):
|
||
result = left * right
|
||
elif right == 0:
|
||
return None
|
||
else:
|
||
result = left / right
|
||
return result if valid(result) else None
|
||
|
||
return visit(tree)
|
||
|
||
|
||
def parse_analysis(text, metrics, defaults):
|
||
if not isinstance(text, str) or len(text) > 4000:
|
||
raise QueryError("Keep the query under 4,000 characters.")
|
||
from .stats_expressions import calculation, lower
|
||
from .stats_features import extract, validate
|
||
|
||
text, expressions = lower(text)
|
||
text, features = extract(text)
|
||
catalog, units = dict(metrics), {**UNITS, **EXTRA_UNITS}
|
||
lines = text.splitlines()
|
||
derived, aggregates, seen, locations = [], {}, set(), {}
|
||
options = {
|
||
"all_time": False,
|
||
"bin": None,
|
||
"take": None,
|
||
"limit": None,
|
||
"rank": None,
|
||
"trend": None,
|
||
"plot": None,
|
||
"chart": "auto",
|
||
"having": [],
|
||
}
|
||
group = None
|
||
or_lines = any(re.match(r"\s*or\s", line) for line in lines)
|
||
filter_counts = []
|
||
aliases = {}
|
||
|
||
def once(key, line):
|
||
if key in seen:
|
||
raise QueryError(f"Repeated {key} clause.", line)
|
||
seen.add(key)
|
||
locations[key] = line
|
||
|
||
def aggregate(source, line):
|
||
source = source.strip()
|
||
match = re.fullmatch(r"(\w+)\s*\(\s*(\w*)\s*\)", source)
|
||
if not match:
|
||
if source not in catalog:
|
||
raise QueryError(f"Unknown metric '{source}'.", line)
|
||
return source
|
||
function, field = match.groups()
|
||
if function not in FUNCTIONS or (not field and function != "count"):
|
||
raise QueryError(
|
||
"Use latest, sum, mean, weighted_mean, median, min, max, stddev, count or p10/p25/p75/p90/p95/p99.",
|
||
line,
|
||
)
|
||
if field and field not in catalog:
|
||
raise QueryError(f"Unknown metric '{field}'.", line)
|
||
key = f"{function}({field})"
|
||
aggregates[key] = {"function": function, "field": field}
|
||
return key
|
||
|
||
for index, source in enumerate(lines):
|
||
line, number = source.strip(), index + 1
|
||
for name, expression in expressions.get(number, []):
|
||
spec, conversions = calculation(name, expression, catalog, units, defaults, number)
|
||
if len(derived) >= 32:
|
||
raise QueryError("Use at most 32 metric expressions.", number)
|
||
catalog[name] = (expression, spec["unit"], f"Per-input expression: {expression}.")
|
||
units[name] = conversions
|
||
derived.append(spec)
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if match := re.fullmatch(r"let (\w+) = (.+)", line):
|
||
name, expression = match.groups()
|
||
if (
|
||
not re.fullmatch(r"[a-z][a-z0-9_]{0,39}", name)
|
||
or name.startswith("analysis_")
|
||
or name in catalog
|
||
or sum(not s.get("inline") for s in derived) >= 8
|
||
):
|
||
raise QueryError(
|
||
"Choose a new lowercase metric name; at most eight calculations.", number
|
||
)
|
||
_, unit = expression_tree(expression, catalog, number)
|
||
catalog[name] = (
|
||
name.replace("_", " ").capitalize(),
|
||
unit,
|
||
f"Calculated per run: {expression}. Missing inputs or division by zero give missing data.",
|
||
)
|
||
units[name] = {"s": 1, "min": 60} if unit == "time" else {unit: 1}
|
||
spec = {"name": name, "expression": expression, "unit": unit}
|
||
derived.append(spec)
|
||
lines[index] = ""
|
||
elif line == "during all time":
|
||
once("all-time period", number)
|
||
options["all_time"] = True
|
||
lines[index] = "during last 12 weeks"
|
||
elif line.startswith("summarize "):
|
||
keys = [aggregate(part, number) for part in line[10:].split(",")]
|
||
if len(keys) != len(set(keys)):
|
||
raise QueryError("Choose different metrics or summary expressions.", number)
|
||
replacements = []
|
||
for key in keys:
|
||
if key in aggregates:
|
||
alias = f"analysis_agg_{len(aliases)}"
|
||
aliases[alias] = key
|
||
catalog[alias] = (
|
||
key,
|
||
"runs"
|
||
if aggregates[key]["function"] == "count"
|
||
else catalog[aggregates[key]["field"]][1],
|
||
"Aggregate",
|
||
)
|
||
units[alias] = {catalog[alias][1]: 1}
|
||
replacements.append(alias)
|
||
else:
|
||
replacements.append(key)
|
||
lines[index] = "summarize " + ", ".join(replacements)
|
||
elif match := re.fullmatch(r"by (\w+)(?: bins (\S+) (\S+))?", line):
|
||
once("by", number)
|
||
name, width, unit = match.groups()
|
||
if width:
|
||
if name not in catalog or name not in units or unit not in units[name]:
|
||
raise QueryError("Choose a numeric metric and its unit for bins.", number)
|
||
try:
|
||
size = float(width) * units[name][unit]
|
||
except ValueError:
|
||
raise QueryError("Use a positive numeric bin width.", number) from None
|
||
if not valid(size) or size <= 0 or size > 1000000:
|
||
raise QueryError("Use a positive finite bin width.", number)
|
||
options["bin"] = {"field": name, "width": size}
|
||
group = "bin"
|
||
else:
|
||
if name not in GROUPS:
|
||
raise QueryError(
|
||
"Group by day, week, month, year, weekday, hour, run, all, or metric bins.",
|
||
number,
|
||
)
|
||
group = name
|
||
lines[index] = "by week" if group not in {"day", "week", "month"} else line
|
||
elif match := re.fullmatch(r"(take|limit) (\d+)", line):
|
||
name, raw = match.groups()
|
||
once(name, number)
|
||
value = int(raw)
|
||
if not 1 <= value <= 1000:
|
||
raise QueryError("Use a limit from 1 to 1,000.", number)
|
||
options[name] = value
|
||
lines[index] = ""
|
||
elif match := re.fullmatch(r"rank by (.+) (asc|desc)", line):
|
||
once("rank", number)
|
||
options["rank"] = [aggregate(match[1], number), match[2]]
|
||
lines[index] = ""
|
||
elif match := re.fullmatch(r"having (.+?) (>=|<=|>|<|=|!=) (-?\S+)(?: (\S+))?", line):
|
||
key, op, raw, unit = match.groups()
|
||
key = aggregate(key, number)
|
||
function = aggregates.get(key, {}).get("function")
|
||
field = aggregates.get(key, {}).get("field", key)
|
||
accepted = (
|
||
{"runs": 1}
|
||
if function == "count"
|
||
else {"sec/km": 1}
|
||
if catalog[field][1] == "/km" and function in {"sum", "stddev"}
|
||
else units[field]
|
||
)
|
||
unit = unit or next(iter(accepted))
|
||
if unit not in accepted:
|
||
raise QueryError(f"Use {' or '.join(accepted)} for {key}.", number)
|
||
try:
|
||
if unit == "/km" and ":" in raw:
|
||
minutes, seconds = raw.split(":")
|
||
if not re.fullmatch(r"\d+:\d{2}", raw) or int(seconds) >= 60:
|
||
raise ValueError
|
||
value = int(minutes) * 60 + int(seconds)
|
||
else:
|
||
value = float(raw) * accepted[unit]
|
||
except ValueError:
|
||
raise QueryError("Use a numeric threshold for having.", number) from None
|
||
if not valid(value) or abs(value) > 1000000:
|
||
raise QueryError("Use a finite threshold.", number)
|
||
if len(options["having"]) >= 10:
|
||
raise QueryError("Use at most ten having conditions.", number)
|
||
options["having"].append({"field": key, "op": op, "value": value})
|
||
lines[index] = ""
|
||
elif match := re.fullmatch(
|
||
r"trend (cumulative|change|percent_change|rolling (\d+) buckets?)", line
|
||
):
|
||
once("trend", number)
|
||
options["trend"] = match[1]
|
||
if match[2] and not 1 <= int(match[2]) <= 52:
|
||
raise QueryError("Use 1–52 rolling buckets.", number)
|
||
lines[index] = ""
|
||
elif match := re.fullmatch(r"plot (\w+) vs (\w+)", line):
|
||
once("plot", number)
|
||
if any(name not in catalog for name in match.groups()):
|
||
raise QueryError("Choose two numeric run metrics for the scatter plot.", number)
|
||
options["plot"] = list(match.groups())
|
||
lines[index] = ""
|
||
elif match := re.fullmatch(r"chart (auto|line|bar|none)", line):
|
||
once("chart", number)
|
||
options["chart"] = match[1]
|
||
lines[index] = ""
|
||
elif re.match(r"(?:where|and|or) ", line):
|
||
if or_lines:
|
||
# Keep AND higher precedence than OR and prevent SQL bounds from swallowing alternatives.
|
||
rewritten = re.sub(
|
||
r"^(where|and|or) (distance|heart_rate) ", r"\1 analysis_raw_\2 ", line
|
||
)
|
||
for field in ("distance", "heart_rate"):
|
||
alias = "analysis_raw_" + field
|
||
catalog[alias] = catalog[field]
|
||
units[alias] = units[field]
|
||
lines[index] = re.sub(r"^or ", "and ", rewritten)
|
||
filter_counts.append(
|
||
(
|
||
line.startswith("or "),
|
||
2 if re.match(r"^(where|and|or) \w+ between ", line) else 1,
|
||
)
|
||
)
|
||
q = parse_dsl("\n".join(lines), catalog, defaults, units=units)
|
||
q.update(options)
|
||
q.update(features)
|
||
q["derived"], q["aggregates"] = derived, aggregates
|
||
if group:
|
||
q["group"] = group
|
||
for key in ("metric", "secondary"):
|
||
q[key] = aliases.get(q[key], q[key])
|
||
q["extra_metrics"] = [aliases.get(key, key) for key in q["extra_metrics"]]
|
||
selected = [
|
||
q["metric"],
|
||
*([] if q["secondary"] == "none" else [q["secondary"]]),
|
||
*q["extra_metrics"],
|
||
]
|
||
if len(set(selected)) != len(selected):
|
||
raise QueryError("Choose different summary expressions.")
|
||
if options["rank"] and options["rank"][0] not in selected:
|
||
raise QueryError("Include the ranking expression in summarize.", locations["rank"])
|
||
if options["trend"] and q["group"] not in {"day", "week", "month", "year"}:
|
||
raise QueryError("Trends require grouping by day, week, month or year.", locations["trend"])
|
||
if options["trend"] and options["rank"]:
|
||
raise QueryError(
|
||
"Ranked groups and time trends use different orderings. Use one per query.",
|
||
locations["trend"],
|
||
)
|
||
if options["plot"] and options["chart"] == "none":
|
||
raise QueryError("A scatter plot conflicts with chart none.", locations["chart"])
|
||
if options["all_time"] and (q["end"] or q["start"]):
|
||
raise QueryError("All time cannot also specify an end date.", locations["all-time period"])
|
||
if options["all_time"]:
|
||
if any(line.strip() == "compare previous period" for line in text.splitlines()):
|
||
raise QueryError(
|
||
"All time has no preceding period. Use compare none.", locations["all-time period"]
|
||
)
|
||
q["compare"] = False
|
||
q["filter_groups"] = []
|
||
if or_lines:
|
||
position, groups = 0, [[]]
|
||
for new_group, count in filter_counts:
|
||
if new_group:
|
||
if not groups[-1]:
|
||
raise QueryError("Start an OR filter with where or and.")
|
||
groups.append([])
|
||
batch = q["filters"][position : position + count]
|
||
for condition in batch:
|
||
condition["field"] = condition["field"].removeprefix("analysis_raw_")
|
||
groups[-1].extend(batch)
|
||
position += count
|
||
q["filter_groups"], q["filters"] = groups, []
|
||
if q["reference"]:
|
||
ref = q["reference"]
|
||
checked = parse_dsl(
|
||
f"runs\nwhere {ref['field']} = {ref['raw']} {ref['unit']}\nsummarize {ref['field']}",
|
||
catalog,
|
||
defaults,
|
||
units,
|
||
)
|
||
ref["value"] = checked["filters"][0]["value"]
|
||
if ref["field"] not in selected:
|
||
raise QueryError("Include the reference metric in summarize.")
|
||
for cohort in q["cohorts"]:
|
||
match = re.fullmatch(r"(\w+) (.+)", cohort["source"])
|
||
if not match or match[1] not in catalog:
|
||
raise QueryError("Cohorts need a numeric metric condition.")
|
||
field, condition = match.groups()
|
||
cohort_catalog = {**catalog, "cohort_input": catalog[field]}
|
||
checked = parse_dsl(
|
||
f"runs\nwhere cohort_input {condition}\nsummarize {field}",
|
||
cohort_catalog,
|
||
defaults,
|
||
{**units, "cohort_input": units[field]},
|
||
)
|
||
cohort["conditions"] = [{**item, "field": field} for item in checked["filters"]]
|
||
if not cohort["conditions"]:
|
||
raise QueryError("Supply a cohort comparison with a value and unit.")
|
||
validate(q, catalog)
|
||
return q
|
||
|
||
|
||
def describe(key, catalog, aggregates):
|
||
if key not in aggregates:
|
||
return catalog[key]
|
||
spec = aggregates[key]
|
||
function, field = spec["function"], spec["field"]
|
||
label = (
|
||
"Runs"
|
||
if not field
|
||
else f"{function.replace('_', ' ').capitalize()} · {catalog[field][0].lower()}"
|
||
)
|
||
unit = "runs" if function == "count" else catalog[field][1]
|
||
if field and catalog[field][1] == "/km" and function in {"stddev", "sum"}:
|
||
unit = "sec/km"
|
||
definition = (
|
||
(
|
||
"Count of rows"
|
||
if not field
|
||
else f"{function} of available per-run {catalog[field][0].lower()} values"
|
||
)
|
||
+ ". Percentiles use linear interpolation; stddev is population standard deviation. Missing values are excluded."
|
||
)
|
||
if function == "latest":
|
||
definition = "Value on the latest input by local date/time, including a missing value on that input. Sorting does not change which input is latest."
|
||
return label, unit, definition
|
||
|
||
|
||
def statistic(values, function):
|
||
values = [value for value in values if valid(value)]
|
||
if function == "count":
|
||
return len(values)
|
||
if not values:
|
||
return None
|
||
if function == "sum":
|
||
return sum(values)
|
||
if function == "mean":
|
||
return statistics.fmean(values)
|
||
if function == "median":
|
||
return statistics.median(values)
|
||
if function == "min":
|
||
return min(values)
|
||
if function == "max":
|
||
return max(values)
|
||
if function == "stddev":
|
||
return statistics.pstdev(values)
|
||
percentile = int(function[1:]) / 100
|
||
ordered = sorted(values)
|
||
index = (len(ordered) - 1) * percentile
|
||
lower = math.floor(index)
|
||
return ordered[lower] + (ordered[math.ceil(index)] - ordered[lower]) * (index - lower)
|