#!/usr/bin/env python3 """roam.py — RoamQuest file-contract parser (Stage 1, s078). Turns fitness EXPORT FILES into the RoamQuest measurement shape. Runs on the PLAYER's machine (tq.py-style); the box only ever sees the derived output. PRIVACY CONTRACT (load-bearing — see notes/design-roamquest.md): * NEVER emitted: coordinates, lat/lon bounding boxes, track names/descriptions, device identifiers, weight, height, raw heart-rate samples, sleep, calories. * Emitted: per-DAY buckets (steps · distance · move/active minutes · heart minutes · moving minutes) and per-SESSION records (type · start · end · duration · distance · SMOOTHED elevation gain · has_route). That is the whole surface. Sources (auto-detected by path): stepsy .csv "YYYY-MM-DD,steps", no header, one row/day opentracks .kmz|.kml OpenTracks KML 2.3 MultiTrack export takeout takeout-*.zip | dir Google Fit Takeout: Daily activity metrics/ + All sessions/ ONLY (All data/ + .tcx ignored) Usage: roam.py parse ... [-o out.json] [--handle NAME] parse one or more inputs --handle = the hero's board handle → output.player.handle (the ONLY identity field; never a real name) roam.py merge ... [-o out.json] merge parsed outputs (max per metric/day) roam.py summary human one-screen digest roam.py measure print the MEASUREMENTS your hero pushes to the board (eleven numbers; one hero per person since s078 — the --hero flag is a lab leftover) roam.py sheet print the HERO SHEET (the full-page extras: ~40 whitelisted derived numbers) measure + sheet = EVERYTHING that ever leaves your machine roam.py register --board URL --handle H --passphrase P claim your handle in the RoamQuest Hall (first-claim-wins) roam.py push --board URL --handle H (--passphrase P | --device-token T) [--dry-run] [--avatar 🥾] push your hero's measurements + sheet (the server derives the standing); --dry-run prints the exact request body and sends NOTHING roam.py serve [--page roam-logbook.html] [--port 8794] [--board URL --handle H (--passphrase P | --device-token T)] THE LOGBOOK, PRIVATE: a tiny HTTP server on 127.0.0.1 that serves the logbook page (a file YOU downloaded next to this script — this script never downloads anything) and your feed, behind a browser-enforced wall (CSP: the page can talk ONLY to this 127.0.0.1 server). Nothing leaves your box by opening it. With --board credentials the page gains a "share" button: it hands THE SHARE SNAPSHOT (below) to this server, which forwards it to the board — the page itself never talks to the board. roam.py share --board URL --handle H (--passphrase P | --device-token T) [snapshot.json | --list | --unlist | --revoke] snapshot.json → push/refresh your share snapshot (mints a share key the first time; prints the unlisted URL). --list / --unlist → show or hide it on the Hall (LISTED = the QUIET public edition; see below). --revoke → DELETE the snapshot AND the key on the server: revoke = gone. THE SHARE SNAPSHOT (the ONLY thing `share` sends — SHARE_SECTIONS below is the contract): the logbook at MONTH resolution — monthly and yearly totals, the level ladder, weekday/hour outing counts, streak and gap lengths with their dates, records (each with its single date), outing histograms, cadence/stride/moving share, the tales' inputs. NEVER a day-by-day record, NEVER an outing as a row, NEVER a timestamp (validate_snapshot() refuses them). A LISTED (public) logbook is served QUIETER by the board: no hour-of-day anywhere, streaks and gaps as lengths only, no dates on the longest outings, no busiest day, no diary notices, first/last at month resolution. Output shape ("roam-measurements/1"): {"schema": "roam-measurements/1", "generated": iso, "sources": [..], "player": {"handle": str}?, "days": {"YYYY-MM-DD": {"steps": int, "distance_m": float, "move_min": float, "heart_min": float, "active_min": float, "sessions": int, "moving_min": float}}, # s079: minutes the tracker saw you walking/running/riding # (Takeout activity segments; 0 where a source has none) "sessions": [{"id": str, "source": str, "type": str, "start": iso, "end": iso, "duration_min": float, "distance_m": float, "gain_m": float|null, "has_route": bool, "steps": int|null, "heart_min": float|null, "active_min": float|null}], "coverage": {"": {"first": date, "last": date, "days": n, "zero_days": n}}} THE TRUST MAP — every network request this script can make, exhaustively. Both go to the ONE --board URL you pass on the command line (there is no built-in server address); both flow through the single helper board_post() (grep urlopen — it is the only call site). Nothing here runs unless you type the command: POST ?action=register `roam.py register` sends pack + handle + passphrase POST ?action=submit `roam.py push` sends pack + handle + passphrase (or device token) + the snapshot: the ELEVEN measurements, the sheet (SHEET_KEYS — ~40 derived numbers + 3 dates), an avatar, an item name. The keeper DISCARDS the rhythm on arrival: peak hour + the weekday spread are never stored, and the dates are kept month-only. `push --dry-run` prints that body verbatim and sends NOTHING. POST ?action=share `roam.py share X`, sends pack + handle + credential + or the page's share THE SHARE SNAPSHOT (SHARE_SECTIONS, button via `serve` ≤ 128 KB, validated first) POST ?action=share_list `share --list/ sends pack + handle + credential + --unlist` listed 0/1 POST ?action=share_revoke `share --revoke` sends pack + handle + credential parse / merge / summary / measure / sheet make NO network request at all. `serve` LISTENS on 127.0.0.1 only (loopback Host names only; DNS-rebinding guarded) and makes the `share` request above ONLY when the page's share button is pressed and ONLY if you started it with --board credentials. This script never updates itself, never reads a sensor, never runs in the background, never needs sudo, and never sends the feed (FORBIDDEN_PUSH guards it: days, coverage, timestamps, coordinates and the player block can't ride). Day keys are the LOCAL calendar date of the player's export (Stepsy/Takeout give local dates; OpenTracks timestamps carry their own offset — we use it). Stepsy note: users may PAUSE counting (Flick does — only "going for a walk" counts), so a Stepsy zero-day means "not counted", not "did not move". The game layer reads coverage.zero_days to tell a pauser from an always-on counter. """ import sys, os, re, io, csv, json, math, zipfile, hashlib, statistics, datetime as dt import xml.etree.ElementTree as ET SCHEMA = "roam-measurements/1" KML_NS = {"k": "http://www.opengis.net/kml/2.2", "k23": "http://www.opengis.net/kml/2.3"} ACTIVITY_MAP = { # normalise vendor labels → the pack's activity vocabulary "walking": "walk", "walk": "walk", "hiking": "hike", "running": "run", "run": "run", "biking": "ride", "cycling": "ride", "road biking": "ride", "mountain biking": "ride", "e-biking": "ride", "paced": "walk", "paced walking": "walk", "trail running": "run", } def norm_activity(s): s = (s or "").strip().lower() if s in ACTIVITY_MAP: return ACTIVITY_MAP[s] head = re.split(r"[.\s_-]", s)[0] # "walking.paced" / "running_treadmill" → the verb return ACTIVITY_MAP.get(head, s or "unknown") def empty(): return {"schema": SCHEMA, "generated": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"), "sources": [], "days": {}, "sessions": [], "coverage": {}} def day_rec(days, key): return days.setdefault(key, {"steps": 0, "distance_m": 0.0, "move_min": 0.0, "heart_min": 0.0, "active_min": 0.0, "sessions": 0, "moving_min": 0.0}) def sess_id(source, typ, start): return hashlib.sha256(f"{source}|{typ}|{start}".encode()).hexdigest()[:16] def fnum(v, default=0.0): try: return float(v) except (TypeError, ValueError): return default # ---------------------------------------------------------------- geometry def haversine(a, b): R = 6371000.0 p1, p2 = math.radians(a[1]), math.radians(b[1]) dp, dl = p2 - p1, math.radians(b[0] - a[0]) h = math.sin(dp/2)**2 + math.cos(p1)*math.cos(p2)*math.sin(dl/2)**2 return 2*R*math.asin(math.sqrt(h)) GAIN_WINDOW = 15 # samples each side for the running median (~7 s/sample on phones) GAIN_HYST = 3.0 # metres — a climb counts only once it exceeds this since the last trough def smoothed_gain(alts): """Elevation gain robust to GPS altitude noise: running median then hysteresis. Raw GPS gain overstates ~7x on flat walks (s078 measurement: 730 m raw vs a 62 m altitude range); median-15 + 3 m hysteresis brings it to the plausible band.""" if len(alts) < 3: return 0.0 k = GAIN_WINDOW med = [statistics.median(alts[max(0, i-k):i+k+1]) for i in range(len(alts))] gain, ref = 0.0, med[0] # dead-band filter: the reference only moves once for a in med[1:]: # the trace leaves a +-GAIN_HYST band around it if a > ref + GAIN_HYST: gain += a - ref; ref = a elif a < ref - GAIN_HYST: ref = a return round(gain, 1) # ---------------------------------------------------------------- stepsy def parse_stepsy(path, out): src = "stepsy"; cov = {"first": None, "last": None, "days": 0, "zero_days": 0} with open(path, newline="") as f: for row in csv.reader(f): if len(row) < 2 or not re.fullmatch(r"\d{4}-\d\d-\d\d", row[0].strip()): continue d, steps = row[0].strip(), int(fnum(row[1])) rec = day_rec(out["days"], d); rec["steps"] = max(rec["steps"], steps) cov["days"] += 1; cov["zero_days"] += (steps == 0) cov["first"] = min(cov["first"] or d, d); cov["last"] = max(cov["last"] or d, d) out["sources"].append(src); out["coverage"][src] = cov # ---------------------------------------------------------------- opentracks def _kml_text(path): if path.lower().endswith(".kmz"): z = zipfile.ZipFile(path) name = next(n for n in z.namelist() if n.lower().endswith(".kml")) return z.read(name).decode("utf-8", "replace") return open(path, encoding="utf-8", errors="replace").read() def parse_opentracks(path, out): src = "opentracks" text = _kml_text(path) # namespace-agnostic: strip the default-ns prefix issue by regex on local names typ_m = re.search(r'', text) typ = norm_activity(typ_m.group(1) if typ_m else "") whens = [dt.datetime.fromisoformat(w.strip()) for w in re.findall(r"([^<]+)", text)] coords = [] for c in re.findall(r"([^<]*)", text): p = c.split() if len(p) >= 2: coords.append((float(p[0]), float(p[1]), float(p[2]) if len(p) > 2 else 0.0)) if not whens: return start, end = min(whens), max(whens) dist = sum(haversine(a, b) for a, b in zip(coords, coords[1:])) gain = smoothed_gain([c[2] for c in coords]) if len(coords) >= 3 else None rec = {"id": sess_id(src, typ, start.isoformat()), "source": src, "type": typ, "start": start.isoformat(timespec="seconds"), "end": end.isoformat(timespec="seconds"), "duration_min": round((end - start).total_seconds() / 60, 1), "distance_m": round(dist, 1), "gain_m": gain, "has_route": len(coords) >= 2, "steps": None, "heart_min": None, "active_min": None} out["sessions"].append(rec) d = start.date().isoformat() # LOCAL date from the export's own offset day = day_rec(out["days"], d); day["sessions"] += 1 day["distance_m"] = round(day["distance_m"] + dist, 1) day["active_min"] = round(day["active_min"] + rec["duration_min"], 1) cov = out["coverage"].setdefault(src, {"first": d, "last": d, "days": 0, "zero_days": 0, "sessions": 0}) cov["first"] = min(cov["first"], d); cov["last"] = max(cov["last"], d); cov["sessions"] += 1 cov["days"] = len({s["start"][:10] for s in out["sessions"] if s["source"] == src}) if src not in out["sources"]: out["sources"].append(src) # ---------------------------------------------------------------- google takeout DAILY_RE = re.compile(r"Daily activity metrics/(\d{4}-\d\d-\d\d)\.csv$") SESS_RE = re.compile(r"All sessions/[^/]+\.json$") TAKEOUT_COLS = { # header-keyed (12 header variants observed across one 6-year export) — never positional "steps": "Step count", "distance_m": "Distance (m)", "move_min": "Move Minutes count", "heart_min": "Heart Minutes", } MOVING_COLS = ("Walking duration (ms)", "Running duration (ms)", "Cycling duration (ms)") # s079 "time on the move" def _takeout_members(path): """yield (relname, bytes) for the two folders we consume; works on a zip or an extracted dir.""" if os.path.isdir(path): for root, _, files in os.walk(path): for fn in files: rel = os.path.join(root, fn).replace("\\", "/") if DAILY_RE.search(rel) or SESS_RE.search(rel): yield rel, open(os.path.join(root, fn), "rb").read() else: z = zipfile.ZipFile(path) for n in z.namelist(): if DAILY_RE.search(n) or SESS_RE.search(n): yield n, z.read(n) def parse_takeout(path, out): src = "takeout"; cov = {"first": None, "last": None, "days": 0, "zero_days": 0, "sessions": 0} for rel, data in _takeout_members(path): m = DAILY_RE.search(rel) if m: d = m.group(1); rec = day_rec(out["days"], d); tot = {k: 0.0 for k in TAKEOUT_COLS}; moving = 0.0 for row in csv.DictReader(io.StringIO(data.decode("utf-8", "replace"))): for k, col in TAKEOUT_COLS.items(): tot[k] += fnum(row.get(col, "")) moving += sum(fnum(row.get(c, "")) for c in MOVING_COLS) / 60000.0 rec["steps"] = max(rec["steps"], int(tot["steps"])) for k in ("distance_m", "move_min", "heart_min"): rec[k] = round(max(rec[k], tot[k]), 1) rec["moving_min"] = round(max(rec.get("moving_min", 0.0), moving), 1) cov["days"] += 1; cov["zero_days"] += (tot["steps"] == 0) cov["first"] = min(cov["first"] or d, d); cov["last"] = max(cov["last"] or d, d) continue j = json.loads(data.decode("utf-8", "replace")) typ = norm_activity(j.get("fitnessActivity")) start = dt.datetime.fromisoformat(j["startTime"].replace("Z", "+00:00")) end = dt.datetime.fromisoformat(j["endTime"].replace("Z", "+00:00")) agg = {} for a in j.get("aggregate", []): agg[a.get("metricName", "")] = fnum(a.get("floatValue", a.get("intValue"))) dur = fnum(str(j.get("duration", "")).rstrip("s"), (end - start).total_seconds()) / 60 rec = {"id": sess_id(src, typ, start.isoformat()), "source": src, "type": typ, "start": start.isoformat(timespec="seconds"), "end": end.isoformat(timespec="seconds"), "duration_min": round(dur, 1), "distance_m": round(agg.get("com.google.distance.delta", 0.0), 1), "gain_m": None, "has_route": False, "steps": int(agg["com.google.step_count.delta"]) if "com.google.step_count.delta" in agg else None, "heart_min": agg.get("com.google.heart_minutes.summary"), "active_min": agg.get("com.google.active_minutes")} out["sessions"].append(rec); cov["sessions"] += 1 day_rec(out["days"], start.date().isoformat())["sessions"] += 1 # UTC date: Takeout sessions carry Z out["sources"].append(src); out["coverage"][src] = cov # ---------------------------------------------------------------- detect / merge / summary def detect(path): b = os.path.basename(path).lower() if os.path.isdir(path): if os.path.isdir(os.path.join(path, "Takeout")) or os.path.isdir(os.path.join(path, "Fit")): return "takeout" return "dir" if b.startswith("takeout") and b.endswith(".zip"): return "takeout" if b.endswith((".kmz", ".kml")): return "opentracks" if b.startswith("stepsy") and b.endswith(".csv"): return "stepsy" if b.endswith(".csv"): with open(path) as f: first = f.readline() if re.match(r"\d{4}-\d\d-\d\d,\d+", first): return "stepsy" return None def parse_paths(paths): out = empty() for p in paths: kind = detect(p) if kind == "dir": parse_paths_into(out, sorted(os.path.join(p, f) for f in os.listdir(p))); continue parse_one(kind, p, out) out["sessions"].sort(key=lambda s: s["start"]) return out def parse_paths_into(out, paths): for p in paths: k = detect(p) if k and k != "dir": parse_one(k, p, out) def parse_one(kind, p, out): if kind == "stepsy": parse_stepsy(p, out) elif kind == "opentracks": parse_opentracks(p, out) elif kind == "takeout": parse_takeout(p, out) else: print(f"skip (unknown format): {p}", file=sys.stderr) def merge(docs): out = empty(); seen = set() for d in docs: for s in d.get("sources", []): if s not in out["sources"]: out["sources"].append(s) for k, v in d.get("days", {}).items(): rec = day_rec(out["days"], k) for m in ("steps", "distance_m", "move_min", "heart_min", "active_min", "moving_min"): rec[m] = max(rec.get(m, 0), v.get(m, 0)) # max, not sum: sources overlap (phone + watch) rec["sessions"] = max(rec["sessions"], v.get("sessions", 0)) for s in d.get("sessions", []): if s["id"] in seen: continue seen.add(s["id"]); out["sessions"].append(s) for s, c in d.get("coverage", {}).items(): out["coverage"][s] = c if d.get("player"): out["player"] = d["player"] out["sessions"].sort(key=lambda s: s["start"]) return out def summary(doc): days = doc["days"]; sess = doc["sessions"] lines = [f"roam-measurements · sources={','.join(doc['sources'])} · days={len(days)} · sessions={len(sess)}"] for s, c in doc["coverage"].items(): lines.append(f" {s:10s} {c.get('first')} → {c.get('last')} days={c.get('days')} zero_days={c.get('zero_days')} sessions={c.get('sessions','-')}") steps = sorted(v["steps"] for v in days.values()) if steps: lines.append(f" steps/day: median={steps[len(steps)//2]} p90={steps[int(len(steps)*.9)]} max={steps[-1]} zero={steps.count(0)}") if sess: by = {} for s in sess: by.setdefault(s["type"], []).append(s) for t, ss in sorted(by.items()): km = sum(s["distance_m"] for s in ss)/1000; mins = sum(s["duration_min"] for s in ss) gains = [s["gain_m"] for s in ss if s["gain_m"] is not None] lines.append(f" {t:8s} n={len(ss):5d} {km:8.1f} km {mins:8.0f} min route={sum(s['has_route'] for s in ss)} gain_m(sum)={sum(gains):.0f}") return "\n".join(lines) # ---------------------------------------------------------------- the party + the board (Stage 2) FAMILIES = {"ambient": None, "trail": ["walk", "hike", "run"], "road": ["ride"], "arena": ["row", "lift", "swim", "indoor", "workout"]} def fam_of(t): return next((f for f, ts in FAMILIES.items() if ts and t in ts), "trail") def hero_view(doc, fam): """A feed view — lockstep with roam-pack.js viewFor(). "all" = THE hero (option C, s078): every day bucket + every session. ambient = the day buckets alone; a family = its labelled sessions, day rows rebuilt from them (steps ~1,300/km when the tracker gives none). Families are sheet-only now (no per-family heroes).""" if fam == "all": days, _ = hero_view(doc, "ambient") for s in doc.get("sessions", []): # a session's minutes/HR/distance count on its day even when the day bucket lacks them d = days.setdefault(str(s["start"])[:10], {"steps": 0, "distance_m": 0, "move_min": 0, "heart_min": 0, "active_min": 0, "sessions": 0}) d["active_min"] += s.get("duration_min") or 0; d["sessions"] += 1 d["heart_min"] = max(d["heart_min"], s.get("heart_min") or 0) d["distance_m"] = max(d["distance_m"], s.get("distance_m") or 0) if d["steps"] == 0: d["steps"] = s["steps"] if s.get("steps") is not None else round((s.get("distance_m") or 0) * 1.3) return days, list(doc.get("sessions", [])) if fam == "ambient": return {k: {"steps": d.get("steps", 0), "distance_m": d.get("distance_m", 0), "move_min": d.get("move_min", 0), "heart_min": d.get("heart_min", 0), "active_min": 0, "sessions": 0, "moving_min": d.get("moving_min", 0)} for k, d in doc["days"].items()}, [] sess = [s for s in doc.get("sessions", []) if fam_of(s["type"]) == fam]; days = {} for s in sess: d = days.setdefault(str(s["start"])[:10], {"steps": 0, "distance_m": 0, "move_min": 0, "heart_min": 0, "active_min": 0, "sessions": 0}) d["steps"] += s["steps"] if s.get("steps") is not None else round((s.get("distance_m") or 0) * 1.3) d["distance_m"] += s.get("distance_m") or 0; d["heart_min"] += s.get("heart_min") or 0 d["active_min"] += s.get("duration_min") or 0; d["sessions"] += 1 return days, sess def day_effort(d): return max(d.get("heart_min", 0), 0.5 * max(d.get("active_min", 0), d.get("move_min", 0)), d.get("steps", 0) / 200) def heroes_of(doc): """Option C (s078): ONE hero per person. (The party shape is retired; families live in the sheet.)""" return ["all"] SHEET_KEYS = ("steps_total","km_total","gain_m_total","effort_min_total","effort_hr_share","outings","days_counted","span_days", "first_day","last_day","peak_hour","streak_days","best_day","best_day_steps","longest_min","farthest_km","most_gain_m", "mix_walk","mix_run","mix_ride","mix_hike","mix_other","km_walk","km_run","km_ride","km_hike", "fam_everyday_days","fam_trail_outings","fam_trail_km","fam_trail_longest","fam_road_outings","fam_road_km", "fam_road_longest","fam_arena_outings","fam_arena_min","wd_0","wd_1","wd_2","wd_3","wd_4","wd_5","wd_6") def sheet(doc): """THE HERO SHEET — the 'full hero page' extras (option (i)): ~40 whitelisted DERIVED numbers + 3 dates. This is the complete list of what the sheet can ever carry. No routes, no samples. The server whitelist is SMALLER (s083 a′): the keeper discards peak_hour + wd_0..wd_6 at ingest and stores the three dates month-only.""" days, sess = hero_view(doc, "all"); keys = sorted(days) counted = [k for k in keys if days[k]["steps"] > 0 or days[k]["sessions"] > 0] span = max(7, (dt.date.fromisoformat(keys[-1]) - dt.date.fromisoformat(keys[0])).days + 1) if len(keys) > 1 else 7 km = lambda ss: round(sum((x.get("distance_m") or 0) for x in ss) / 1000, 1) by = lambda t: [x for x in sess if x["type"] == t] fam = lambda f: [x for x in sess if fam_of(x["type"]) == f] routed = [x for x in sess if x.get("gain_m") is not None and (x.get("distance_m") or 0) > 0] # s080: lockstep with measure() hours = sorted(int(str(x["start"])[11:13]) for x in sess) wd = [0] * 7 for x in sess: wd[dt.date.fromisoformat(str(x["start"])[:10]).weekday()] += 1 streak = best = 0; prev = None for k in counted: d = dt.date.fromisoformat(k); streak = streak + 1 if prev and (d - prev).days == 1 else 1; best = max(best, streak); prev = d bd = max(keys, key=lambda k: days[k]["steps"]) if keys else None eff = [day_effort(d) for d in days.values()]; hr = sum(min(d.get("heart_min", 0), day_effort(d)) for d in days.values()) out = {"steps_total": sum(d["steps"] for d in days.values()), "km_total": max(km(sess), round(sum(d["distance_m"] for d in days.values()) / 1000, 1)), "gain_m_total": round(sum(x["gain_m"] for x in routed), 1), "effort_min_total": round(sum(eff), 1), "effort_hr_share": round(hr / sum(eff), 3) if sum(eff) else 0.0, "outings": len(sess), "days_counted": len(counted), "span_days": span, "first_day": keys[0] if keys else None, "last_day": keys[-1] if keys else None, "peak_hour": hours[len(hours) // 2] if hours else None, "streak_days": best, "best_day": bd, "best_day_steps": days[bd]["steps"] if bd else 0, "longest_min": round(max([x.get("duration_min") or 0 for x in sess], default=0), 1), "farthest_km": round(max([(x.get("distance_m") or 0) for x in sess], default=0) / 1000, 2), "most_gain_m": round(max([x["gain_m"] for x in routed], default=0), 1), "mix_walk": len(by("walk")), "mix_run": len(by("run")), "mix_ride": len(by("ride")), "mix_hike": len(by("hike")), "mix_other": len([x for x in sess if x["type"] not in ("walk", "run", "ride", "hike")]), "km_walk": km(by("walk")), "km_run": km(by("run")), "km_ride": km(by("ride")), "km_hike": km(by("hike")), "fam_everyday_days": len(counted), "fam_trail_outings": len(fam("trail")), "fam_trail_km": km(fam("trail")), "fam_trail_longest": round(max([x.get("duration_min") or 0 for x in fam("trail")], default=0), 1), "fam_road_outings": len(fam("road")), "fam_road_km": km(fam("road")), "fam_road_longest": round(max([x.get("duration_min") or 0 for x in fam("road")], default=0), 1), "fam_arena_outings": len(fam("arena")), "fam_arena_min": round(sum((x.get("duration_min") or 0) for x in fam("arena")), 1)} for i in range(7): out[f"wd_{i}"] = wd[i] assert set(out) <= set(SHEET_KEYS) return out def measure(doc, fam): """The board payload for one hero: ELEVEN derived numbers + the family + the handle. Nothing else ever leaves.""" days, sess = hero_view(doc, fam); keys = sorted(days) active = sum(1 for k in keys if days[k]["steps"] > 0 or days[k]["sessions"] > 0) or 1 span = max(7, (dt.date.fromisoformat(keys[-1]) - dt.date.fromisoformat(keys[0])).days + 1) if len(keys) > 1 else 7 km = sum((s.get("distance_m") or 0) for s in sess) / 1000; daykm = sum(d["distance_m"] for d in days.values()) / 1000 routed = [s for s in sess if s.get("gain_m") is not None and (s.get("distance_m") or 0) > 0] hours = sorted(int(str(s["start"])[11:13]) for s in sess) return {"family": fam, "steps_total": sum(d["steps"] for d in days.values()), "effort_min_total": round(sum(day_effort(d) for d in days.values()), 3), "km_total": round(max(km, daykm), 3), "gain_m_total": round(sum(s["gain_m"] for s in routed), 1), "gain_km": round(sum(s["distance_m"] / 1000 for s in routed), 3), "longest_min": max([s.get("duration_min") or 0 for s in sess], default=0), "sessions": len(sess), "active_days": active, "span_days": span, "peak_hour": hours[len(hours) // 2] if hours else None} FORBIDDEN_PUSH = {"days", "coverage", "start", "end", "coord", "latitude", "longitude", "player"} # a payload never carries the feed itself def board_post(url, action, body, timeout=20): import urllib.request data = json.dumps(body).encode() req = urllib.request.Request(url + "?action=" + action, data=data, headers={"Content-Type": "application/json", "User-Agent": "roam.py/1"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def push(doc, board, handle, passphrase=None, token=None, heroes=None, dry=False, avatar="🥾"): heroes = heroes or heroes_of(doc); out = [] for fam in heroes: m = measure(doc, fam) snap = {"measurements": m, "sheet": sheet(doc), "avatar": avatar, "item": "Boots of Questionable Tread"} body = {"pack": "roam", "handle": handle, "snapshot": snap} if token: body["device_token"] = token else: body["passphrase"] = passphrase assert not (set(snap) & FORBIDDEN_PUSH) and not (set(m) & FORBIDDEN_PUSH) and isinstance(m["sessions"], int), "privacy guard: the feed itself must never be in a push" if dry: shown = dict(body); shown["passphrase" if not token else "device_token"] = "***" print("POST", board + "?action=submit"); print(json.dumps(shown, indent=1)); continue r = board_post(board, "submit", body); out.append((fam, r)) print(f"pushed → renown {r.get('renown')} · rank #{r.get('rank')}" if r.get("ok") else f"push → {r}") return out # ---------------------------------------------------------------- the share snapshot (round 10, s079) # FORMULA-LOCKSTEP with the logbook page's SHARE_SECTIONS (mockups/roam-stats.html → /get/roam-logbook.html) and board.php. SHARE_SECTIONS = ("handle","level","xp","stats","m","steps","effort","km","gain","gainKm","hrShare","first","last","spanDays","sessN","countedN","daysN","mix", "months","years","ladder","effortBins","hh","hours","dows","dowSteps","dowN","peak","streaks","droughts", "bestDay","bestDaySteps","bestDayKm","bestWeek","bestMonth","farthest","longest","climb","fastest","farthestBy","busiest","oStreak","onThisDay","dayOne","firstOuting","odo","earliest","latest", "kindAgg","sessStepsSum","outingMin","sessAvg","oh","top8","qagg","feetMin","feetDays","trackedFeet","motion") SHARE_META = ("schema", "generated") SHARE_SCHEMA = "roam-logbook/1" SHARE_MAX_BYTES = 128 * 1024 SHARE_NEVER = ("days", "keys", "counted", "sess", "sessions", "feetOf", "raw") # the per-day / per-outing arrays — never in a snapshot _ISO_TS = re.compile(r"\d{4}-\d\d-\d\dT\d\d") _DATE_KEY = re.compile(r"^\d{4}-\d\d-\d\d$") SNAP_FORBIDDEN = {"lat", "lon", "latitude", "longitude", "coord", "coords", "coordinates", "description", "name", "weight", "height", "bpm", "sleep", "device", "calories", "kcal"} # EXACT key names (the feed-side FORBIDDEN is substring-based and would trip on "latest"/"longest") def validate_snapshot(obj): """The belt under the page's whitelist: returns a list of problems (empty = OK). A snapshot is a dict with schema roam-logbook/1, keys ⊆ SHARE_SECTIONS ∪ SHARE_META, no per-day/per-outing arrays, no timestamps (dates YYYY-MM-DD are fine; `generated` is the one ISO timestamp), no forbidden words, ≤ SHARE_MAX_BYTES.""" bad = [] if not isinstance(obj, dict): return ["snapshot must be a JSON object"] if obj.get("schema") != SHARE_SCHEMA: bad.append("schema must be %s" % SHARE_SCHEMA) extra = sorted(k for k in obj if k not in SHARE_SECTIONS and k not in SHARE_META) if extra: bad.append("keys not in SHARE_SECTIONS: %s" % extra) for k in obj: if k in SHARE_NEVER: bad.append("per-day/per-outing key %r at top level" % k) def walk(o, path): # STRUCTURAL rule, anywhere: no dict keyed by dates (a per-day map), no list of records carrying ids, no non-finite numbers if isinstance(o, dict): if any(_DATE_KEY.match(str(k)) for k in o): bad.append("a date-keyed map at %s (per-day data)" % (path or "/")) for k, v in o.items(): if k in SNAP_FORBIDDEN: bad.append("forbidden key %r at %s" % (k, path or "/")) walk(v, path + "." + k) elif isinstance(o, list): if any(isinstance(v, dict) and "id" in v for v in o): bad.append("a list of records with ids at %s (per-outing data)" % (path or "/")) for v in o: walk(v, path + "[]") elif isinstance(o, float) and (math.isnan(o) or math.isinf(o)): bad.append("non-finite number at %s" % path) walk(obj, "") text = json.dumps(obj, separators=(",", ":")) if len(text.encode()) > SHARE_MAX_BYTES: bad.append("snapshot is %d bytes (cap %d)" % (len(text.encode()), SHARE_MAX_BYTES)) stamps = _ISO_TS.findall(json.dumps({k: v for k, v in obj.items() if k != "generated"})) if stamps: bad.append("timestamps in the snapshot (%d) — dates only" % len(stamps)) return bad def share_body(handle, snapshot, passphrase=None, token=None, **extra): body = {"pack": "roam", "handle": handle}; body.update(extra) if snapshot is not None: body["snapshot"] = snapshot if token: body["device_token"] = token else: body["passphrase"] = passphrase return body def share(board, handle, passphrase=None, token=None, snapshot=None, listed=None, revoke=False): if revoke: return board_post(board, "share_revoke", share_body(handle, None, passphrase, token)) if listed is not None: return board_post(board, "share_list", share_body(handle, None, passphrase, token, listed=1 if listed else 0)) bad = validate_snapshot(snapshot) if bad: raise ValueError("snapshot refused: " + "; ".join(bad)) return board_post(board, "share", share_body(handle, snapshot, passphrase, token)) # ---------------------------------------------------------------- serve: the private logbook on 127.0.0.1 CSP_HTML = ("default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; " "font-src 'self' data:; connect-src 'self'; object-src 'none'; base-uri 'none'; form-action 'self'; frame-ancestors 'self'") PAGE_HINT = "curl -fsSL https://questful.online/get/roam-logbook.html -o roam-logbook.html (put it next to roam.py; this script never downloads)" def make_server(feed_path, page_path, port, creds=None): """creds = {"board","handle","passphrase"|"token"} or None (no share button).""" from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def _send(self, code, body, ctype="application/json"): if isinstance(body, str): body = body.encode() self.send_response(code); self.send_header("Content-Type", ctype); self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store"); self.send_header("X-Content-Type-Options", "nosniff") if ctype.startswith("text/html"): self.send_header("Content-Security-Policy", CSP_HTML); self.send_header("X-DNS-Prefetch-Control", "off") self.end_headers(); self.wfile.write(body) def _host_ok(self): host = (self.headers.get("Host") or "").lower() host = host[1:host.find("]")] if host.startswith("[") and "]" in host else host.split(":")[0] if host and host not in ("127.0.0.1", "localhost", "::1"): self._send(403, '{"error":"forbidden host"}'); return False return True def do_GET(self): if not self._host_ok(): return path = self.path.split("?", 1)[0] if path in ("/", "/index.html", "/roam-logbook.html"): if not os.path.exists(page_path): return self._send(404, "the logbook page is not here yet — " + PAGE_HINT, "text/plain; charset=utf-8") return self._send(200, open(page_path, "rb").read(), "text/html; charset=utf-8") if path == "/tq/roam.json": try: return self._send(200, open(feed_path, "rb").read()) except OSError: return self._send(404, '{"error":"no feed"}') if path == "/serve-info": return self._send(200, json.dumps({"roam": True, "share": bool(creds), "handle": (creds or {}).get("handle")})) return self._send(404, '{"error":"not found"}') def do_POST(self): if not self._host_ok(): return path = self.path.split("?", 1)[0] if path != "/share": return self._send(404, '{"error":"not found"}') n = int(self.headers.get("Content-Length") or 0) if n > SHARE_MAX_BYTES + 4096: return self._send(413, '{"error":"too large"}') try: body = json.loads(self.rfile.read(n).decode("utf-8")) except Exception: return self._send(400, '{"error":"bad json"}') snap = body.get("snapshot") if isinstance(body, dict) else None bad = validate_snapshot(snap) if bad: return self._send(422, json.dumps({"error": "snapshot refused", "problems": bad})) if not creds: return self._send(409, json.dumps({"error": "no board credentials — restart serve with --board URL --handle H and --passphrase or --device-token"})) try: r = share(creds["board"], creds["handle"], creds.get("passphrase"), creds.get("token"), snapshot=snap) except Exception as e: return self._send(502, json.dumps({"error": "board: %s" % e})) return self._send(200, json.dumps(r)) return ThreadingHTTPServer(("127.0.0.1", port), H) def serve(feed_path, page_path, port, creds): srv = make_server(feed_path, page_path, port, creds) print("RoamQuest logbook, private, on your box: http://127.0.0.1:%d/" % srv.server_address[1]) print(" feed: %s · page: %s%s" % (feed_path, page_path, "" if os.path.exists(page_path) else " (MISSING — " + PAGE_HINT + ")")) print(" share button: %s" % ("ON — the page can hand this server a snapshot to forward to %s" % creds["board"] if creds else "off (no --board credentials)")) try: srv.serve_forever() except KeyboardInterrupt: pass FORBIDDEN = ("lat", "lon", "coord", "description", "name", "weight", "height", "bpm", "sleep", "device", "calor") def privacy_check(doc): """Assert no forbidden key appears anywhere in the output. Cheap belt for the wall.""" bad = set() def walk(o): if isinstance(o, dict): for k, v in o.items(): if any(f in k.lower() for f in FORBIDDEN): bad.add(k) walk(v) elif isinstance(o, list): for v in o: walk(v) walk(doc); return sorted(bad) def main(argv): if len(argv) < 2 or argv[1] in ("-h", "--help"): print(__doc__); return 0 cmd, rest = argv[1], argv[2:] outp = None; handle = None if cmd == "parse" and "--handle" in rest: i = rest.index("--handle"); handle = rest[i+1]; rest = rest[:i] + rest[i+2:] if "-o" in rest: i = rest.index("-o"); outp = rest[i+1]; rest = rest[:i] + rest[i+2:] def opt(flag, default=None): nonlocal rest if flag in rest: i = rest.index(flag); v = rest[i+1]; rest = rest[:i] + rest[i+2:]; return v return default if cmd == "measure": hero = opt("--hero", "all"); print(json.dumps(measure(json.load(open(rest[0])), hero), indent=1)); return 0 if cmd == "sheet": print(json.dumps(sheet(json.load(open(rest[0]))), indent=1)); return 0 if cmd == "register": board, h, pw = opt("--board"), opt("--handle"), opt("--passphrase") if not (board and h and pw): print("register needs --board --handle --passphrase", file=sys.stderr); return 2 print(json.dumps(board_post(board, "register", {"pack": "roam", "handle": h, "passphrase": pw}))); return 0 if cmd == "push": board, h = opt("--board"), opt("--handle"); pw, tok = opt("--passphrase"), opt("--device-token") hero = opt("--hero"); av = opt("--avatar", "🥾"); dry = "--dry-run" in rest if dry: rest.remove("--dry-run") if not (board and h and (pw or tok)): print("push needs --board --handle and --passphrase or --device-token", file=sys.stderr); return 2 push(json.load(open(rest[0])), board, h, pw, tok, hero.split(",") if hero else None, dry, av); return 0 if cmd == "serve": board, h = opt("--board"), opt("--handle"); pw, tok = opt("--passphrase"), opt("--device-token") page = opt("--page", os.path.join(os.path.dirname(os.path.abspath(__file__)), "roam-logbook.html")); port = int(opt("--port", "8794")) if not rest: print("serve needs ", file=sys.stderr); return 2 creds = {"board": board, "handle": h, "passphrase": pw, "token": tok} if (board and h and (pw or tok)) else None serve(rest[0], page, port, creds); return 0 if cmd == "share": board, h = opt("--board"), opt("--handle"); pw, tok = opt("--passphrase"), opt("--device-token") if not (board and h and (pw or tok)): print("share needs --board --handle and --passphrase or --device-token", file=sys.stderr); return 2 if "--revoke" in rest: print(json.dumps(share(board, h, pw, tok, revoke=True))); return 0 if "--list" in rest: print(json.dumps(share(board, h, pw, tok, listed=True))); return 0 if "--unlist" in rest: print(json.dumps(share(board, h, pw, tok, listed=False))); return 0 if not rest: print("share needs a snapshot.json, or --list / --unlist / --revoke", file=sys.stderr); return 2 try: print(json.dumps(share(board, h, pw, tok, snapshot=json.load(open(rest[0]))))) except ValueError as e: print(str(e), file=sys.stderr); return 3 return 0 if cmd == "parse": doc = parse_paths(rest) elif cmd == "merge": doc = merge([json.load(open(p)) for p in rest]) elif cmd == "summary": print(summary(json.load(open(rest[0])))); return 0 else: print(__doc__); return 2 if handle: doc["player"] = {"handle": handle} bad = privacy_check(doc) if bad: print(f"PRIVACY CHECK FAILED: forbidden keys {bad}", file=sys.stderr); return 3 text = json.dumps(doc, indent=1, sort_keys=True) if outp: open(outp, "w").write(text + "\n"); print(summary(doc)) else: print(text) return 0 if __name__ == "__main__": sys.exit(main(sys.argv))