"""Keeps the latest readings for the live screen and records runs automatically. A run starts when the speed goes above start_mph (the 2 seconds before are kept, so the launch is included) and ends after the speed stays below stop_mph for 1.5 seconds. Runs that never reach min_top_mph are thrown away. Each run is saved as a JSON file in the runs folder. """ import collections import json import math import os import threading import time from . import config from .analysis import MPS_TO_MPH PREROLL_S = 2.0 STOP_AFTER_S = 1.5 MAX_RUN_S = 120.0 CHANNELS = ("gps", "wheel_rpm", "engine_rpm", "accel_g") class Recorder: def __init__(self): self.lock = threading.Lock() self.settings = config.load_settings() self.buffers = {ch: collections.deque(maxlen=400) for ch in CHANNELS} self.latest = {} # channel -> (time, value) self.speed_mps = 0.0 self.speed_from = "none" self.gps_speed = (0.0, 0.0) # (time, m/s) of the latest GPS speed self.wheel_speed = (0.0, 0.0) # (time, m/s) of the latest wheel-sensor speed self.run = None # the run being recorded, or None self.manual = False self.slow_since = None self.last_saved = None self.on_saved = [] # callbacks(run_id) self.rates = {ch: collections.deque(maxlen=60) for ch in CHANNELS} def update_settings(self, settings): with self.lock: self.settings = dict(settings) # ---- readings from the sensors ---- def emit(self, channel, t, value): if channel not in CHANNELS: return with self.lock: self.buffers[channel].append((t, value)) self.latest[channel] = (t, value) self.rates[channel].append(t) if self.run is not None: self._append(self.run, channel, t, value) speed = self._speed_from_reading(channel, value) if speed is not None: self.speed_mps = speed self._detect(t, speed) def _speed_from_reading(self, channel, value): """Speed for starting/stopping runs. In "auto" it uses whichever of GPS and wheel sensor shows more speed, so an unplugged sensor stuck at 0 (or a GPS without a fix) can't stop a run from recording.""" if channel == "gps": self.gps_speed = (time.monotonic(), value["speed_mps"] if value.get("fix_ok") else 0.0) elif channel == "wheel_rpm": circumference = math.pi * float(self.settings["tire_diameter_in"]) * 0.0254 self.wheel_speed = (time.monotonic(), value / 60.0 * circumference) else: return None source = self.settings.get("speed_source", "auto") now = time.monotonic() candidates = [] if source in ("auto", "gps") and now - self.gps_speed[0] < 1.0: candidates.append((self.gps_speed[1], "gps")) if source in ("auto", "wheel") and now - self.wheel_speed[0] < 1.0: candidates.append((self.wheel_speed[1], "wheel")) if not candidates: return None speed, self.speed_from = max(candidates) return speed def _detect(self, t, speed): mph = speed * MPS_TO_MPH if self.run is None: if mph >= float(self.settings["start_mph"]): self._start(t, manual=False) return self.run["top_mph"] = max(self.run["top_mph"], mph) if self.manual: if t - self.run["t0"] > MAX_RUN_S * 2.5: self._finish() return if mph < float(self.settings["stop_mph"]): if self.slow_since is None: self.slow_since = t elif t - self.slow_since >= STOP_AFTER_S: self._finish() else: self.slow_since = None if self.run is not None and t - self.run["t0"] > MAX_RUN_S: self._finish() # ---- runs ---- def _start(self, t, manual): t0 = t - (0 if manual else PREROLL_S) self.run = {"t0": t0, "wall_start": time.time() - (t - t0), "top_mph": 0.0, "channels": {}, "manual": manual} self.manual = manual self.slow_since = None for channel, buffer in self.buffers.items(): for bt, value in buffer: if bt >= t0: self._append(self.run, channel, bt, value) def _append(self, run, channel, t, value): ch = run["channels"].setdefault(channel, {"t": []}) ch["t"].append(round(t - run["t0"], 3)) if channel == "gps": for key, src in (("speed_mps", "speed_mps"), ("sacc", "speed_acc_mps"), ("sats", "sats"), ("lat", "lat"), ("lon", "lon"), ("alt", "alt_m"), ("heading", "heading_deg")): ch.setdefault(key, []).append(value.get(src)) ch.setdefault("ok", []).append(bool(value.get("fix_ok"))) ch.setdefault("utc", value.get("utc")) else: ch.setdefault("v", []).append(round(float(value), 4)) def _finish(self): run, self.run, self.manual = self.run, None, False duration = max((max(c["t"]) for c in run["channels"].values() if c["t"]), default=0) if not run["manual"] and (run["top_mph"] < float(self.settings["min_top_mph"]) or duration < 2): return # just rolling around, not a run if not any(c["t"] for c in run["channels"].values()): return os.makedirs(config.RUNS_DIR, exist_ok=True) stamp = run["wall_start"] while True: # never overwrite a run that started in the same second run_id = time.strftime("%Y%m%d-%H%M%S", time.localtime(stamp)) if not os.path.exists(os.path.join(config.RUNS_DIR, run_id + ".json")): break stamp += 1 record = { "id": run_id, "name": "", "started": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(run["wall_start"])), "duration_s": round(duration, 2), "manual": run["manual"], "top_mph_live": round(run["top_mph"], 1), "recorded_with": {k: self.settings[k] for k in ("bike_weight_lb", "rider_weight_lb", "tire_diameter_in", "drivetrain")}, "channels": run["channels"], } path = os.path.join(config.RUNS_DIR, run_id + ".json") with open(path + ".tmp", "w") as f: json.dump(record, f, separators=(",", ":")) os.replace(path + ".tmp", path) self.last_saved = run_id for callback in list(self.on_saved): try: callback(run_id) except Exception: pass def manual_start(self): with self.lock: if self.run is None: self._start(time.monotonic(), manual=True) return True def manual_stop(self): with self.lock: if self.run is not None: self._finish() return self.last_saved # ---- for the live screen ---- def live(self): with self.lock: now = time.monotonic() out = {"speed_mph": round(self.speed_mps * MPS_TO_MPH, 1), "speed_from": self.speed_from, "recording": self.run is not None, "manual": self.manual, "last_run": self.last_saved} if self.run is not None: out["run_seconds"] = round(now - self.run["t0"], 1) out["run_top_mph"] = round(self.run["top_mph"], 1) gps = self.latest.get("gps") if gps and now - gps[0] < 3: g = gps[1] out["gps"] = {"fix": g.get("fix"), "fix_ok": g.get("fix_ok"), "sats": g.get("sats"), "speed_acc_mph": round((g.get("speed_acc_mps") or 0) * MPS_TO_MPH, 2), "rate_hz": self._rate("gps", now)} for ch in ("wheel_rpm", "engine_rpm", "accel_g"): reading = self.latest.get(ch) if reading and now - reading[0] < 3: out[ch] = round(reading[1], 3 if ch == "accel_g" else 0) return out def _rate(self, channel, now): times = [t for t in self.rates[channel] if now - t <= 3] return round(len(times) / 3.0, 1)