Mini Bike GPS Dyno

Every kind of sensor. To add a new one, write one class marked @sensor_type.

"""Sensors. Each type describes its own options, so the app can add and set them upwithout code changes. Every sensor runs in its own thread and sends readings toemit(channel, monotonic_time, value):  channel "gps"         value = dict from ubx.parse_nav_pvt (speed_mps, fix_ok, sats, ...)  channel "wheel_rpm"   value = rear wheel revolutions per minute  channel "engine_rpm"  value = engine revolutions per minute  channel "accel_g"     value = forward acceleration in g (from an accelerometer)"""import mathimport randomimport threadingimport timefrom . import ubxSENSOR_TYPES = {}def sensor_type(name, label, options, channels, help_text=""):    def register(cls):        SENSOR_TYPES[name] = {"label": label, "class": cls, "options": options,                              "channels": channels, "help": help_text}        cls.type_name = name        return cls    return registerdef describe_types():    """What the app's "Add sensor" form needs, without the Python classes."""    return {name: {k: v for k, v in info.items() if k != "class"}            for name, info in SENSOR_TYPES.items()}class Sensor:    def __init__(self, config, emit):        self.config = config        self.emit = emit        self.stop_event = threading.Event()        self.thread = None        self.state = "starting"        self.detail = ""        self.last_reading = 0.0    def opt(self, key):        for option in SENSOR_TYPES[self.type_name]["options"]:            if option["key"] == key:                return self.config.get(key, option["default"])        return self.config.get(key)    def start(self):        self.thread = threading.Thread(target=self._run_forever, daemon=True,                                       name="sensor-" + self.config.get("id", "?"))        self.thread.start()    def stop(self):        self.stop_event.set()        if self.thread:            self.thread.join(timeout=3)    def _run_forever(self):        while not self.stop_event.is_set():            try:                self.run()            except Exception as e:  # keep going: a loose wire shouldn't kill the dyno                self.state = "error"                self.detail = str(e)[:200]            self.stop_event.wait(2)    def run(self):        raise NotImplementedError    def status(self):        age = time.monotonic() - self.last_reading if self.last_reading else None        return {"id": self.config.get("id"), "type": self.type_name,                "name": self.config.get("name"), "state": self.state,                "detail": self.detail,                "seconds_since_reading": round(age, 1) if age is not None else None}# ---------------------------------------------------------------- GPS ---@sensor_type(    "ublox_gps", "GPS (u-blox, like the HGLRC M100)",    options=[        {"key": "port", "label": "Serial port", "kind": "text", "default": "/dev/serial0",         "help": "/dev/serial0 = the Pi's pins 8 (TX) and 10 (RX)"},        {"key": "rate_hz", "label": "Updates per second", "kind": "select", "default": 10,         "choices": [5, 10, 15]},    ],    channels=["gps"],    help_text="Wire GPS TX to Pi pin 10, GPS RX to pin 8, 5V to pin 4, GND to pin 6.")class UbloxGps(Sensor):    BAUDS = [115200, 9600, 38400, 57600, 230400]    def run(self):        import serial  # pyserial, part of Raspberry Pi OS        port = self.opt("port")        self.state = "searching"        self.detail = "looking for the GPS on " + port        baud = self._find_baud(serial, port)        if baud is None:            self.state = "no data"            self.detail = "nothing heard on %s (check the wires, TX goes to pin 10)" % port            return        if baud != 115200:            with serial.Serial(port, baud, timeout=0.2) as ser:                ser.write(ubx.valset({"CFG-UART1-BAUDRATE": 115200}))                ser.flush()                time.sleep(0.3)        with serial.Serial(port, 115200, timeout=0.2) as ser:            self._configure(ser)            self._read_loop(ser)    def _find_baud(self, serial, port):        for baud in self.BAUDS:            if self.stop_event.is_set():                return None            try:                with serial.Serial(port, baud, timeout=0.2) as ser:                    ser.reset_input_buffer()                    data = b""                    end = time.monotonic() + 1.3                    while time.monotonic() < end:                        data += ser.read(512)                    if ubx.SYNC in data or b"$G" in data:                        return baud            except OSError as e:                self.detail = "can't open %s: %s" % (port, e)                time.sleep(1)        return None    def _configure(self, ser):        rate_ms = int(round(1000 / int(self.opt("rate_hz"))))        full = {            "CFG-RATE-MEAS": rate_ms,            "CFG-RATE-NAV": 1,            "CFG-UART1OUTPROT-UBX": 1,            "CFG-UART1OUTPROT-NMEA": 0,            "CFG-MSGOUT-UBX_NAV_PVT_UART1": 1,            "CFG-NAVSPG-DYNMODEL": 4,            # GPS + Galileo only, so the M10 can keep up 10 fixes a second            "CFG-SIGNAL-GLO_ENA": 0,            "CFG-SIGNAL-BDS_ENA": 0,        }        basic = {k: v for k, v in full.items() if "SIGNAL" not in k}        for values in (full, basic):            ser.reset_input_buffer()            ser.write(ubx.valset(values))            ser.flush()            if self._wait_ack(ser):                self.detail = "configured: %d Hz, UBX" % int(self.opt("rate_hz"))                return True        self.detail = "GPS didn't confirm the settings (it may still work at its own rate)"        return False    def _wait_ack(self, ser):        reader = ubx.Reader()        end = time.monotonic() + 1.5        while time.monotonic() < end:            for cls, mid, payload in reader.feed(ser.read(256)):                if cls == 0x05 and len(payload) >= 2 and payload[0] == 0x06 and payload[1] == 0x8A:                    return mid == 0x01  # 0x01 = ACK, 0x00 = NAK        return False    def _read_loop(self, ser):        reader = ubx.Reader()        last_msg = time.monotonic()        count_start, count = time.monotonic(), 0        self.state = "waiting for fix"        while not self.stop_event.is_set():            data = ser.read(512)            now = time.monotonic()            for cls, mid, payload in reader.feed(data):                if cls == 0x01 and mid == 0x07:                    pvt = ubx.parse_nav_pvt(payload)                    if not pvt:                        continue                    last_msg = now                    count += 1                    self.last_reading = now                    self.emit("gps", now, pvt)                    if pvt["fix_ok"]:                        self.state = "ok"                        self.detail = "%s fix, %d satellites" % (ubx.FIX_NAMES.get(pvt["fix"], "?"), pvt["sats"])                    else:                        self.state = "waiting for fix"                        self.detail = "%d satellites so far (needs open sky)" % pvt["sats"]            if now - count_start >= 5:                self.rate = count / (now - count_start)                count_start, count = now, 0            if now - last_msg > 4:                self.state = "no data"                self.detail = "GPS went quiet, reconnecting"                return  # _run_forever reconnects and reconfigures    def status(self):        s = super().status()        s["rate_hz"] = round(getattr(self, "rate", 0.0), 1)        return s# ----------------------------------------------------- practice / demo ---@sensor_type(    "sim_gps", "Practice GPS (fake rides, for trying the app)",    options=[        {"key": "engine_hp", "label": "Pretend engine horsepower", "kind": "number", "default": 6.5},        {"key": "every_s", "label": "Seconds between fake rides", "kind": "number", "default": 40},    ],    channels=["gps"],    help_text="Makes up a ride every so often so you can see how runs look. "              "Turn it off before real riding.")class SimulatedGps(Sensor):    def run(self):        from .analysis import simulate_run        self.state = "ok"        while not self.stop_event.is_set():            self.detail = "parked"            self._stream(None, 3.0)            ride = simulate_run(engine_hp=float(self.opt("engine_hp")), seconds=9.0)            self.detail = "doing a fake ride"            self._stream(ride, None)            self.detail = "parked"            self._stream(None, max(5.0, float(self.opt("every_s"))))    def _stream(self, ride, seconds):        """Send 10 Hz fixes: the ride's speeds, or standing still for `seconds`."""        speeds = ride["speed_mps"] if ride else [0.0] * int(seconds * 10)        for v in speeds:            if self.stop_event.is_set():                return            now = time.monotonic()            self.last_reading = now            noisy = max(0.0, v + random.gauss(0, 0.04))            self.emit("gps", now, {"fix": 3, "fix_ok": True, "sats": 14, "speed_mps": noisy,                                   "speed_acc_mps": 0.15, "lat": 0.0, "lon": 0.0, "alt_m": 0.0,                                   "heading_deg": 90.0, "itow_ms": int(now * 1000),                                   "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())})            time.sleep(0.1)# --------------------------------------------------- RPM (wheel/engine) ---@sensor_type(    "pulse_rpm", "RPM pickup (hall sensor, magnet, or tach signal)",    options=[        {"key": "role", "label": "Measures", "kind": "select", "default": "wheel",         "choices": ["wheel", "engine"]},        {"key": "gpio", "label": "GPIO number (not pin number)", "kind": "number", "default": 17,         "help": "GPIO17 is physical pin 11. Power 3.3V sensors from pin 1, never 5V into a GPIO."},        {"key": "pulses_per_rev", "label": "Pulses per revolution", "kind": "number", "default": 1,         "help": "Number of magnets on the wheel, or tach pulses per engine turn"},        {"key": "pull", "label": "Pull resistor", "kind": "select", "default": "up",         "choices": ["up", "down", "none"]},        {"key": "edge", "label": "Count on", "kind": "select", "default": "falling",         "choices": ["falling", "rising"]},        {"key": "max_rpm", "label": "Highest believable RPM", "kind": "number", "default": 6000,         "help": "Pulses faster than this are treated as electrical noise"},    ],    channels=["wheel_rpm", "engine_rpm"],    help_text="One magnet on the wheel + a hall sensor (3.3V) is enough for wheel speed.")class PulseRpm(Sensor):    def run(self):        import lgpio  # part of Raspberry Pi OS        gpio = int(self.opt("gpio"))        ppr = max(1, int(self.opt("pulses_per_rev")))        channel = "engine_rpm" if self.opt("role") == "engine" else "wheel_rpm"        max_rpm = max(100.0, float(self.opt("max_rpm")))        min_gap_ns = int(60e9 / (max_rpm * ppr))        flags = {"up": lgpio.SET_PULL_UP, "down": lgpio.SET_PULL_DOWN}.get(self.opt("pull"), 0)        edge = lgpio.FALLING_EDGE if self.opt("edge") == "falling" else lgpio.RISING_EDGE        lock = threading.Lock()        pulses = []           # the GPIO library's timestamp (ns) of recent pulses, for exact spacing        last_arrival = [0]    # our own clock when the latest pulse arrived, for "how long ago"        handle = lgpio.gpiochip_open(0)        try:            lgpio.gpio_claim_alert(handle, gpio, edge, flags)            def on_pulse(chip, g, level, tick):                with lock:                    if pulses and tick - pulses[-1] < min_gap_ns:                        return  # too soon: a bounce or electrical noise                    pulses.append(tick)                    del pulses[:-64]                    last_arrival[0] = time.monotonic_ns()            callback = lgpio.callback(handle, gpio, edge, on_pulse)            self.state = "ok"            self.detail = "GPIO%d, %d pulse(s)/rev" % (gpio, ppr)            while not self.stop_event.wait(0.05):  # 20 readings a second                with lock:                    recent = list(pulses[-(ppr + 1):])                    since_last = time.monotonic_ns() - last_arrival[0]                rpm = 0.0                if len(recent) >= 2:                    # average spacing over the last full turn (or as much as we have)                    period_ns = (recent[-1] - recent[0]) / (len(recent) - 1) * ppr                    if period_ns > 0 and since_last < max(2 * period_ns, 1e8):                        rpm = 60e9 / period_ns                    elif since_last < 3e9:                        rpm = 60e9 / (since_last * ppr)  # slowing down, no new pulse yet                self.last_reading = time.monotonic()                self.emit(channel, self.last_reading, rpm)            callback.cancel()        finally:            lgpio.gpiochip_close(handle)# ------------------------------------------------------- accelerometer ---@sensor_type(    "mpu6050", "Accelerometer (GY-521 / MPU-6050, from your 37-sensor kit)",    options=[        {"key": "bus", "label": "I2C bus", "kind": "number", "default": 1},        {"key": "address", "label": "Address", "kind": "select", "default": "0x68",         "choices": ["0x68", "0x69"]},        {"key": "forward", "label": "Axis pointing forward", "kind": "select", "default": "x",         "choices": ["x", "-x", "y", "-y"]},        {"key": "rate_hz", "label": "Readings per second", "kind": "number", "default": 50},    ],    channels=["accel_g"],    help_text="VCC to pin 1 (3.3V), GND to pin 9, SDA to pin 3, SCL to pin 5. "              "Mount it flat; it zeroes itself while the bike stands still at power-on.")class Mpu6050(Sensor):    def run(self):        from smbus2 import SMBus        address = int(str(self.opt("address")), 16)        axis = self.opt("forward")        index = 0 if "x" in axis else 1        sign = -1.0 if axis.startswith("-") else 1.0        period = 1.0 / max(5.0, min(200.0, float(self.opt("rate_hz"))))        with SMBus(int(self.opt("bus"))) as bus:            bus.write_byte_data(address, 0x6B, 0x00)  # wake up            bus.write_byte_data(address, 0x1A, 0x03)  # 44 Hz low-pass filter            bus.write_byte_data(address, 0x1C, 0x08)  # +-4 g range            time.sleep(0.1)            def read_g():                raw = bus.read_i2c_block_data(address, 0x3B, 6)                value = (raw[index * 2] << 8) | raw[index * 2 + 1]                if value >= 0x8000:                    value -= 0x10000                return sign * value / 8192.0  # 8192 counts per g at +-4 g            self.state = "zeroing"            self.detail = "keep the bike still"            zero = sum(read_g() for _ in range(100)) / 100.0            self.state = "ok"            self.detail = "zeroed at %+.3f g" % zero            while not self.stop_event.wait(period):                self.last_reading = time.monotonic()                self.emit("accel_g", self.last_reading, read_g() - zero)class SensorManager:    """Starts the sensors listed in sensors.json and restarts them when the list changes."""    def __init__(self, emit):        self.emit = emit        self.sensors = []        self.lock = threading.Lock()    def start(self, configs):        with self.lock:            for sensor in self.sensors:                sensor.stop()            self.sensors = []            for config in configs:                info = SENSOR_TYPES.get(config.get("type"))                if not info or not config.get("enabled", True):                    continue                sensor = info["class"](config, self.emit)                sensor.start()                self.sensors.append(sensor)    def status(self):        with self.lock:            return [s.status() for s in self.sensors]

The horsepower maths.

"""Turns a recorded run into horsepower, acceleration and timing numbers.Physics: the force pushing the bike forward at the rear tire is    F = m_eff * a  +  m * g * (Crr + slope)  +  1/2 * rho * CdA * v^2(speeding up, + tire rolling resistance, + hill, + air drag). Power = F * v.Divide by 745.7 for horsepower at the wheel. Engine horsepower is the wheel figuredivided by the drivetrain efficiency (clutch/chain or torque converter).Speed comes from the GPS (Doppler, ~10 per second) or from a wheel RPM sensor ifone is fitted. Acceleration comes from an accelerometer if one is fitted,otherwise from how fast the smoothed speed changes."""import mathimport numpy as npG = 9.81RHO_AIR = 1.20      # kg/m^3, near sea level at room temperatureLB_TO_KG = 0.45359237MPS_TO_MPH = 2.236936W_PER_HP = 745.7M_TO_FT = 3.28084# ------------------------------------------------------------ helpers ---def savgol(values, window, deriv=0, dt=1.0):    """Savitzky-Golay filter (quadratic): smooths, or gives a smooth derivative."""    values = np.asarray(values, dtype=float)    n = len(values)    window = int(window) | 1    if n < 3:        return np.zeros(n) if deriv else values.copy()    if window > n:        window = n if n % 2 else n - 1    if window < 3:        return np.gradient(values, dt) if deriv else values.copy()    half = window // 2    x = np.arange(-half, half + 1, dtype=float)    basis = np.vander(x, 3, increasing=True)          # 1, x, x^2    coeffs = np.linalg.pinv(basis)[deriv]             # row 0 = value, row 1 = slope    padded = np.concatenate([np.full(half, values[0]), values, np.full(half, values[-1])])    out = np.convolve(padded, coeffs[::-1], mode="valid")    return out / dt if deriv else outdef resample(times, values, grid):    times = np.asarray(times, dtype=float)    values = np.asarray(values, dtype=float)    if len(times) < 2:        return np.full(len(grid), values[0] if len(values) else 0.0)    order = np.argsort(times)    return np.interp(grid, times[order], values[order])def crossing_time(t, v, level):    """First time v reaches `level`, interpolated between samples, or None."""    above = np.nonzero(v >= level)[0]    if not len(above):        return None    i = above[0]    if i == 0:        return float(t[0])    v0, v1 = v[i - 1], v[i]    frac = (level - v0) / (v1 - v0) if v1 != v0 else 0.0    return float(t[i - 1] + frac * (t[i] - t[i - 1]))def efficiency(settings):    if settings.get("drivetrain") == "torque_converter":        return float(settings["efficiency_torque_converter"])    return float(settings["efficiency_clutch"])def r(x, digits=2):    return None if x is None or not math.isfinite(x) else round(float(x), digits)# ----------------------------------------------------------- analysis ---def analyze(run, settings):    channels = run.get("channels", {})    warnings = []    mass = (float(settings["bike_weight_lb"]) + float(settings["rider_weight_lb"])) * LB_TO_KG    mass_eff = mass * (1 + float(settings["rotating_pct"]) / 100.0)    if not settings.get("weights_set"):        warnings.append("Using example weights - enter your real weights in Settings.")    # ---- speed (m/s) on an even time grid ----    wheel = channels.get("wheel_rpm")    use_wheel = settings.get("speed_source") == "wheel" or (        settings.get("speed_source") == "auto" and wheel and max(wheel.get("v") or [0]) > 0)    gps = channels.get("gps") or {}    if use_wheel and wheel and len(wheel.get("t", [])) >= 5:        source = "wheel sensor"        t_raw = np.asarray(wheel["t"], dtype=float)        circumference = math.pi * float(settings["tire_diameter_in"]) * 0.0254        v_raw = np.asarray(wheel["v"], dtype=float) / 60.0 * circumference        dt = 0.05    else:        source = "GPS"        ok = np.asarray(gps.get("ok", []), dtype=bool)        t_raw = np.asarray(gps.get("t", []), dtype=float)        v_raw = np.asarray(gps.get("speed_mps", []), dtype=float)        if len(ok) == len(t_raw) and len(t_raw):            if ok.mean() < 0.9:                warnings.append("GPS fix was weak for part of this run - park in open sky first.")            t_raw, v_raw = t_raw[ok], v_raw[ok]        dt = float(np.median(np.diff(t_raw))) if len(t_raw) > 2 else 0.1        dt = min(max(dt, 0.02), 0.5)        if dt > 0.15:            warnings.append("GPS sent only %.0f updates a second (10 is best)." % (1 / dt))        sats = gps.get("sats") or []        if sats and np.mean(sats) < 8:            warnings.append("Only %.0f satellites on average - speed is less exact." % np.mean(sats))    if len(t_raw) < 8:        return {"ok": False, "error": "Not enough speed readings in this run.", "warnings": warnings}    grid = np.arange(t_raw.min(), t_raw.max() + 1e-9, dt)    window = int(settings.get("smoothing", 7))    if source == "wheel sensor":        window = window * 2 + 1  # 20 readings/s instead of 10: same smoothing in seconds    v = np.clip(savgol(resample(t_raw, v_raw, grid), window), 0, None)    a = savgol(resample(t_raw, v_raw, grid), window, deriv=1, dt=dt)    accel_source = "speed change (" + source + ")"    accel = channels.get("accel_g")    if accel and len(accel.get("t", [])) >= 10:        a = savgol(resample(accel["t"], accel["v"], grid), window) * G        accel_source = "accelerometer"    # ---- launch and the "pull" (from launch to top speed) ----    t_launch = crossing_time(grid, v, 1.0 / MPS_TO_MPH)  # 1 mph    if t_launch is None:        return {"ok": False, "error": "The bike never got moving in this run.", "warnings": warnings}    i_launch = int(np.searchsorted(grid, t_launch))    i_top = i_launch + int(np.argmax(v[i_launch:]))    pull = np.arange(len(grid))    pull = (pull >= i_launch) & (pull <= i_top) & (v > 2.0 / MPS_TO_MPH)    # ---- forces, power ----    slope = float(settings["slope_pct"]) / 100.0    angle = math.atan(slope)    f_roll = mass * G * float(settings["crr"]) * math.cos(angle)    f_hill = mass * G * math.sin(angle)    f_air = 0.5 * RHO_AIR * float(settings["cda_m2"]) * v ** 2    force = mass_eff * a + f_roll + f_hill + f_air    hp_wheel = force * v / W_PER_HP    hp_wheel_s = savgol(hp_wheel, 5)    eff = efficiency(settings)    hp_engine_s = hp_wheel_s / eff    t = grid - t_launch    mph = v * MPS_TO_MPH    distance_ft = np.concatenate([[0.0], np.cumsum((v[1:] + v[:-1]) / 2 * np.diff(grid))]) * M_TO_FT    distance_ft -= np.interp(t_launch, grid, distance_ft)    summary = {"speed_source": source, "accel_source": accel_source,               "mass_lb": r(mass / LB_TO_KG, 0), "efficiency": eff}    if pull.sum() >= 3:        hp_pull = np.where(pull, hp_wheel_s, -np.inf)        i_peak = int(np.argmax(hp_pull))        summary.update({            "peak_wheel_hp": r(hp_wheel_s[i_peak]),            "peak_engine_hp": r(hp_engine_s[i_peak]),            "peak_hp_at_mph": r(mph[i_peak], 1),            "max_accel_g": r(np.max(a[pull]) / G, 2),        })    else:        warnings.append("The run was too short to measure power.")    summary.update({        "top_mph": r(mph[i_top], 1),        "time_to_top_s": r(grid[i_top] - t_launch, 2),        "zero_to_10_s": r(_since(crossing_time(grid, v, 10 / MPS_TO_MPH), t_launch)),        "zero_to_20_s": r(_since(crossing_time(grid, v, 20 / MPS_TO_MPH), t_launch)),        "zero_to_30_s": r(_since(crossing_time(grid, v, 30 / MPS_TO_MPH), t_launch)),        "sixty_ft_s": r(_since(crossing_time(grid, distance_ft, 60.0), 0.0) if distance_ft[-1] >= 60 else None),        "distance_ft": r(float(np.max(distance_ft)), 0),    })    # ---- engine RPM: the classic dyno chart (power and torque by RPM) ----    by_rpm = None    engine = channels.get("engine_rpm")    if engine and len(engine.get("t", [])) >= 10 and max(engine["v"]) > 500:        rpm = savgol(resample(engine["t"], engine["v"], grid), window)        ok_rpm = pull & (rpm > 1000)        if ok_rpm.sum() >= 3:            torque = hp_engine_s * 5252.0 / np.maximum(rpm, 1)            by_rpm = {"rpm": _list(rpm[ok_rpm], 0), "hp": _list(hp_engine_s[ok_rpm]),                      "torque_lbft": _list(torque[ok_rpm])}            i = int(np.argmax(np.where(ok_rpm, hp_engine_s, -np.inf)))            summary["peak_hp_at_rpm"] = r(rpm[i], 0)            summary["peak_torque_lbft"] = r(float(np.max(torque[ok_rpm])))    keep = (t >= -1.0)    series = {        "t": _list(t[keep]), "mph": _list(mph[keep], 1), "accel_g": _list(a[keep] / G, 3),        "hp_wheel": _list(np.where(v[keep] > 1 / MPS_TO_MPH, hp_wheel_s[keep], 0)),        "distance_ft": _list(distance_ft[keep], 1),    }    for name in ("wheel_rpm", "engine_rpm"):        ch = channels.get(name)        if ch and len(ch.get("t", [])) >= 2:            series[name] = _list(resample(ch["t"], ch["v"], grid)[keep], 0)    curve = {"mph": _list(mph[pull], 1), "hp_wheel": _list(hp_wheel_s[pull]),             "hp_engine": _list(hp_engine_s[pull])}    return {"ok": True, "summary": summary, "series": series, "curve_by_speed": curve,            "curve_by_rpm": by_rpm, "warnings": warnings}def _since(t, t0):    return None if t is None else t - t0def _list(arr, digits=2):    return [round(float(x), digits) for x in arr]# --------------------------------------------------------- simulation ---def simulate_run(engine_hp=6.5, seconds=9.0, settings=None, top_mph=36.0, dt_out=0.1):    """A believable mini bike pull, used by the practice GPS and the self-test.    The engine makes `engine_hp` at the crank; the clutch/chain passes on the    usual share; traction limits the launch; power fades near the governed top    speed. Then it brakes to a stop.    """    from .config import DEFAULT_SETTINGS    s = dict(DEFAULT_SETTINGS)    s.update(settings or {})    mass = (s["bike_weight_lb"] + s["rider_weight_lb"]) * LB_TO_KG    mass_eff = mass * (1 + s["rotating_pct"] / 100.0)    p_wheel = engine_hp * W_PER_HP * efficiency(s)    f_traction = 0.45 * mass * G    v_top = top_mph / MPS_TO_MPH    angle = math.atan(s["slope_pct"] / 100.0)    v, t, h = 0.0, 0.0, 0.002    out_t, out_v, next_out = [], [], 0.0    braking = False    while True:        if not braking and (t >= seconds or v >= v_top * 0.995):            braking = True        if braking:            drive = -0.45 * mass * G        else:            power = p_wheel * max(0.0, 1 - (v / v_top) ** 8)            drive = min(f_traction, power / max(v, 0.1))        resist = mass * G * (s["crr"] * math.cos(angle) + math.sin(angle)) + 0.5 * RHO_AIR * s["cda_m2"] * v * v        v = max(0.0, v + (drive - resist) / mass_eff * h)        t += h        if t >= next_out:            out_t.append(round(t, 3))            out_v.append(v)            next_out += dt_out        if braking and v <= 0.0:            break    return {"t": out_t, "speed_mps": out_v, "wheel_hp": p_wheel / W_PER_HP}

Starts and stops runs by itself.

"""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 for1.5 seconds. Runs that never reach min_top_mph are thrown away. Each run is savedas a JSON file in the runs folder."""import collectionsimport jsonimport mathimport osimport threadingimport timefrom . import configfrom .analysis import MPS_TO_MPHPREROLL_S = 2.0STOP_AFTER_S = 1.5MAX_RUN_S = 120.0CHANNELS = ("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)

The web server the phone app talks to.

"""The web server the phone app talks to (port 8080).  GET  /                      the app itself (web/index.html)  GET  /api/status            "is this the dyno?" + sensor health (the phone app looks for this)  GET  /api/live              speed, GPS, RPM right now  GET  /api/runs              all runs, newest first, with their headline numbers  GET  /api/runs/<id>         one run worked out with the current settings  GET  /api/runs/<id>.csv     the same as a spreadsheet file  POST /api/runs/<id>         {"name": "..."} to rename  DELETE /api/runs/<id>  POST /api/record/start|stop manual recording  GET/POST /api/settings  GET/POST /api/sensors       the sensor list (+ the kinds of sensor you can add)"""import csvimport ioimport jsonimport osimport reimport socketimport threadingfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom . import analysis, config, sensorsVERSION = "1.0"WEB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "web")RUN_ID = re.compile(r"^\d{8}-\d{6}$")class Dyno:    """Everything the server needs, shared between requests."""    def __init__(self, recorder, manager):        self.recorder = recorder        self.manager = manager        self.cache = {}  # run id -> (settings key, analysis)        self.cache_lock = threading.Lock()        recorder.on_saved.append(lambda run_id: self.cache.pop(run_id, None))    def load_run(self, run_id):        if not RUN_ID.match(run_id):            return None        try:            with open(os.path.join(config.RUNS_DIR, run_id + ".json")) as f:                return json.load(f)        except (OSError, ValueError):            return None    def analyzed(self, run_id):        settings = config.load_settings()        key = json.dumps(settings, sort_keys=True)        with self.cache_lock:            hit = self.cache.get(run_id)            if hit and hit[0] == key:                return hit[1], hit[2]        run = self.load_run(run_id)        if run is None:            return None, None        try:            result = analysis.analyze(run, settings)        except Exception as e:  # a strange run shouldn't break the whole list            result = {"ok": False, "error": "Couldn't work this run out: %s" % e, "warnings": []}        meta = {k: v for k, v in run.items() if k != "channels"}        meta["channels"] = sorted(run.get("channels", {}))        with self.cache_lock:            self.cache[run_id] = (key, meta, result)        return meta, result    def run_ids(self):        try:            names = os.listdir(config.RUNS_DIR)        except OSError:            return []        return sorted((n[:-5] for n in names if n.endswith(".json") and RUN_ID.match(n[:-5])),                      reverse=True)def make_handler(dyno):    class Handler(BaseHTTPRequestHandler):        server_version = "MiniBikeDyno/" + VERSION        def log_message(self, fmt, *args):            pass  # keep the service log quiet        # ---- helpers ----        def send_json(self, data, code=200):            body = json.dumps(data).encode()            self.send_response(code)            self.send_header("Content-Type", "application/json")            self.send_header("Cache-Control", "no-store")            self.send_header("Content-Length", str(len(body)))            self.end_headers()            self.wfile.write(body)        def send_file(self, path, content_type):            try:                with open(path, "rb") as f:                    body = f.read()            except OSError:                return self.send_json({"error": "not found"}, 404)            self.send_response(200)            self.send_header("Content-Type", content_type)            self.send_header("Cache-Control", "no-cache")            self.send_header("Content-Length", str(len(body)))            self.end_headers()            self.wfile.write(body)        def body_json(self):            length = int(self.headers.get("Content-Length") or 0)            if length <= 0 or length > 200000:                return {}            try:                return json.loads(self.rfile.read(length))            except ValueError:                return {}        # ---- routes ----        def do_GET(self):            path = self.path.split("?")[0]            if path in ("/", "/index.html"):                return self.send_file(os.path.join(WEB_DIR, "index.html"), "text/html; charset=utf-8")            if path == "/api/status":                return self.send_json({"app": "minibike-dyno", "version": VERSION,                                       "host": socket.gethostname(),                                       "runs": len(dyno.run_ids()),                                       "recording": dyno.recorder.run is not None,                                       "sensors": dyno.manager.status()})            if path == "/api/live":                return self.send_json(dyno.recorder.live())            if path == "/api/runs":                runs = []                for run_id in dyno.run_ids():                    meta, result = dyno.analyzed(run_id)                    if meta is None:                        continue                    item = dict(meta)                    item["ok"] = result.get("ok")                    item["summary"] = result.get("summary", {})                    runs.append(item)                return self.send_json({"runs": runs})            m = re.match(r"^/api/runs/(\d{8}-\d{6})(\.csv)?$", path)            if m:                meta, result = dyno.analyzed(m.group(1))                if meta is None:                    return self.send_json({"error": "no such run"}, 404)                if m.group(2):                    return self.send_csv(m.group(1), result)                return self.send_json({"run": meta, "result": result})            if path == "/api/settings":                return self.send_json({"settings": config.load_settings(),                                       "defaults": config.DEFAULT_SETTINGS})            if path == "/api/sensors":                return self.send_json({"sensors": config.load_sensors(),                                       "types": sensors.describe_types(),                                       "status": dyno.manager.status()})            return self.send_json({"error": "not found"}, 404)        def send_csv(self, run_id, result):            series = result.get("series") or {}            columns = [c for c in ("t", "mph", "accel_g", "hp_wheel", "distance_ft",                                   "wheel_rpm", "engine_rpm") if c in series]            out = io.StringIO()            writer = csv.writer(out)            writer.writerow(columns)            for row in zip(*(series[c] for c in columns)):                writer.writerow(row)            body = out.getvalue().encode()            self.send_response(200)            self.send_header("Content-Type", "text/csv")            self.send_header("Content-Disposition", 'attachment; filename="run-%s.csv"' % run_id)            self.send_header("Content-Length", str(len(body)))            self.end_headers()            self.wfile.write(body)        def do_POST(self):            path = self.path.split("?")[0]            data = self.body_json()            if path == "/api/settings":                settings = config.save_settings(data)                dyno.recorder.update_settings(settings)                return self.send_json({"settings": settings})            if path == "/api/sensors":                cleaned = clean_sensor_list(data.get("sensors"))                if cleaned is None:                    return self.send_json({"error": "bad sensor list"}, 400)                config.save_sensors(cleaned)                dyno.manager.start(cleaned)                return self.send_json({"sensors": cleaned, "status": dyno.manager.status()})            if path == "/api/record/start":                dyno.recorder.manual_start()                return self.send_json({"recording": True})            if path == "/api/record/stop":                return self.send_json({"recording": False, "saved": dyno.recorder.manual_stop()})            m = re.match(r"^/api/runs/(\d{8}-\d{6})$", path)            if m:                return self.rename(m.group(1), data.get("name", ""))            return self.send_json({"error": "not found"}, 404)        def rename(self, run_id, name):            run = dyno.load_run(run_id)            if run is None:                return self.send_json({"error": "no such run"}, 404)            run["name"] = str(name)[:60]            path = os.path.join(config.RUNS_DIR, run_id + ".json")            with open(path + ".tmp", "w") as f:                json.dump(run, f, separators=(",", ":"))            os.replace(path + ".tmp", path)            dyno.cache.pop(run_id, None)            return self.send_json({"ok": True})        def do_DELETE(self):            m = re.match(r"^/api/runs/(\d{8}-\d{6})$", self.path.split("?")[0])            if not m:                return self.send_json({"error": "not found"}, 404)            try:                os.remove(os.path.join(config.RUNS_DIR, m.group(1) + ".json"))            except OSError:                return self.send_json({"error": "no such run"}, 404)            dyno.cache.pop(m.group(1), None)            return self.send_json({"ok": True})    return Handlerdef clean_sensor_list(items):    """Keep only known sensor types and their known options."""    if not isinstance(items, list):        return None    types = sensors.SENSOR_TYPES    cleaned, seen = [], set()    for i, item in enumerate(items):        if not isinstance(item, dict) or item.get("type") not in types:            continue        info = types[item["type"]]        sensor_id = re.sub(r"[^a-z0-9_-]", "", str(item.get("id") or "").lower()) or "%s%d" % (item["type"], i)        while sensor_id in seen:            sensor_id += "x"        seen.add(sensor_id)        entry = {"id": sensor_id, "type": item["type"],                 "name": str(item.get("name") or info["label"])[:60],                 "enabled": bool(item.get("enabled", True))}        for option in info["options"]:            value = item.get(option["key"], option["default"])            if option["kind"] == "number":                try:                    value = float(value)                    value = int(value) if value.is_integer() else value                except (TypeError, ValueError):                    value = option["default"]            elif option["kind"] == "select" and value not in option["choices"]:                matches = [c for c in option["choices"] if str(c) == str(value)]                value = matches[0] if matches else option["default"]            else:                value = str(value)[:100]            entry[option["key"]] = value        cleaned.append(entry)    return cleaneddef serve(dyno, port=8080):    server = ThreadingHTTPServer(("0.0.0.0", port), make_handler(dyno))    server.daemon_threads = True    return server

"""Settings (weights, tire size, ...) and the sensor list, saved as JSON files on the Pi."""import jsonimport osimport threadingDATA_DIR = os.environ.get("DYNO_DATA", os.path.expanduser("~/minibike-dyno-data"))RUNS_DIR = os.path.join(DATA_DIR, "runs")SETTINGS_FILE = os.path.join(DATA_DIR, "settings.json")SENSORS_FILE = os.path.join(DATA_DIR, "sensors.json")# Everything the horsepower math needs. Units are the ones shown in the app.DEFAULT_SETTINGS = {    "bike_weight_lb": 120.0,      # mini bike with fuel    "rider_weight_lb": 150.0,     # rider with helmet and gear    "weights_set": False,         # the app nags until the real weights are entered    "tire_diameter_in": 13.0,     # rear tire, outside diameter (for wheel-speed sensors)    "drivetrain": "clutch",       # "clutch" (chain + centrifugal clutch) or "torque_converter"    "efficiency_clutch": 0.85,    # share of engine power that reaches the tire    "efficiency_torque_converter": 0.70,    "cda_m2": 0.55,               # air drag area of bike + upright rider    "crr": 0.020,                 # tire rolling resistance on pavement    "rotating_pct": 5.0,          # extra "weight" from spinning wheels, clutch, chain    "slope_pct": 0.0,             # driveway slope in the riding direction (+ = uphill)    "smoothing": 7,               # GPS samples used to smooth speed (odd number, 5-15)    "start_mph": 3.0,             # a run starts when you go faster than this    "stop_mph": 2.0,              # and ends after you slow below this    "min_top_mph": 5.0,           # shorter/slower "runs" are thrown away    "speed_source": "auto",       # "auto" (wheel sensor if there is one), "gps" or "wheel"}_lock = threading.Lock()def _ensure_dirs():    os.makedirs(RUNS_DIR, exist_ok=True)def _read_json(path, fallback):    try:        with open(path) as f:            return json.load(f)    except (OSError, ValueError):        return fallbackdef _write_json(path, data):    _ensure_dirs()    tmp = path + ".tmp"    with open(tmp, "w") as f:        json.dump(data, f, indent=2)    os.replace(tmp, path)def load_settings():    with _lock:        saved = _read_json(SETTINGS_FILE, {})    settings = dict(DEFAULT_SETTINGS)    settings.update({k: v for k, v in saved.items() if k in DEFAULT_SETTINGS})    return settingsdef save_settings(changes):    """Merge changes (from the app) into the saved settings, keeping types sane."""    settings = load_settings()    for key, value in changes.items():        if key not in DEFAULT_SETTINGS:            continue        default = DEFAULT_SETTINGS[key]        try:            if isinstance(default, bool):                value = bool(value)            elif isinstance(default, int):                value = int(value)            elif isinstance(default, float):                value = float(value)            else:                value = str(value)        except (TypeError, ValueError):            continue        settings[key] = value    if "bike_weight_lb" in changes or "rider_weight_lb" in changes:        settings["weights_set"] = True    if settings["smoothing"] % 2 == 0:        settings["smoothing"] += 1    settings["smoothing"] = max(3, min(21, settings["smoothing"]))    with _lock:        _write_json(SETTINGS_FILE, settings)    return settings# The GPS is the only sensor to start with. New sensors are added from the app's# Sensors page; each entry is {"id", "type", "name", "enabled", plus its own options}.DEFAULT_SENSORS = [    {"id": "gps", "type": "ublox_gps", "name": "GPS (HGLRC M100)", "enabled": True,     "port": "/dev/serial0", "rate_hz": 10},    {"id": "practice", "type": "sim_gps", "name": "Practice GPS (fake rides)", "enabled": False,     "engine_hp": 6.5, "every_s": 40},]def load_sensors():    with _lock:        sensors = _read_json(SENSORS_FILE, None)    if not isinstance(sensors, list):        sensors = [dict(s) for s in DEFAULT_SENSORS]    return sensorsdef save_sensors(sensors):    with _lock:        _write_json(SENSORS_FILE, sensors)

Speaks the GPS's own language (u-blox UBX) to get 10 readings a second.

"""u-blox UBX protocol: build config messages and read NAV-PVT (position, velocity, time).The HGLRC M100-5883 has a u-blox M10 GPS. We switch it to 115200 baud, 10 fixes asecond, and binary NAV-PVT messages only (NMEA text off), which carry ground speedstraight from the Doppler measurement (accurate to a few cm/s)."""import structSYNC = b"\xb5\x62"# Configuration keys (u-blox M10 interface description) and their value sizes.KEYS = {    "CFG-UART1-BAUDRATE": (0x40520001, "<I"),    "CFG-RATE-MEAS": (0x30210001, "<H"),          # ms between measurements    "CFG-RATE-NAV": (0x30210002, "<H"),           # measurements per solution    "CFG-UART1OUTPROT-UBX": (0x10740001, "<B"),    "CFG-UART1OUTPROT-NMEA": (0x10740002, "<B"),    "CFG-MSGOUT-UBX_NAV_PVT_UART1": (0x20910007, "<B"),    "CFG-NAVSPG-DYNMODEL": (0x20110021, "<B"),    # 4 = automotive    "CFG-SIGNAL-BDS_ENA": (0x10310022, "<B"),    "CFG-SIGNAL-GLO_ENA": (0x10310025, "<B"),}LAYER_RAM = 0x01LAYER_BBR = 0x02  # battery-backed RAM: survives a warm restart if the module keeps powerdef checksum(body):    a = b = 0    for byte in body:        a = (a + byte) & 0xFF        b = (b + a) & 0xFF    return bytes((a, b))def frame(msg_class, msg_id, payload=b""):    body = struct.pack("<BBH", msg_class, msg_id, len(payload)) + payload    return SYNC + body + checksum(body)def valset(values, layers=LAYER_RAM | LAYER_BBR):    """UBX-CFG-VALSET (0x06 0x8A): values is {key name: value}."""    payload = struct.pack("<BBH", 0, layers, 0)    for name, value in values.items():        key_id, fmt = KEYS[name]        payload += struct.pack("<I", key_id) + struct.pack(fmt, value)    return frame(0x06, 0x8A, payload)FIX_NAMES = {0: "no fix", 1: "dead reckoning", 2: "2D", 3: "3D", 4: "GPS+DR", 5: "time only"}def parse_nav_pvt(p):    """NAV-PVT payload (92 bytes) -> dict in plain units."""    if len(p) < 92:        return None    (itow, year, month, day, hour, minute, sec, valid, t_acc, nano, fix_type, flags,     flags2, num_sv, lon, lat, height, h_msl, h_acc, v_acc, vel_n, vel_e, vel_d,     g_speed, head_mot, s_acc, head_acc, p_dop) = struct.unpack_from(        "<IHBBBBBBIiBBBBiiiiIIiiiiiIIH", p, 0)    return {        "itow_ms": itow,        "utc": "%04d-%02d-%02dT%02d:%02d:%02dZ" % (year, month, day, hour, minute, sec),        "fix": fix_type,        "fix_ok": bool(flags & 0x01) and fix_type in (3, 4),        "sats": num_sv,        "lat": lat * 1e-7,        "lon": lon * 1e-7,        "alt_m": h_msl / 1000.0,        "speed_mps": g_speed / 1000.0,        "vel_down_mps": vel_d / 1000.0,        "heading_deg": head_mot * 1e-5,        "speed_acc_mps": s_acc / 1000.0,        "h_acc_m": h_acc / 1000.0,    }class Reader:    """Feed raw bytes in; get complete UBX messages out as (class, id, payload)."""    def __init__(self):        self.buf = bytearray()    def feed(self, data):        self.buf.extend(data)        out = []        while True:            start = self.buf.find(SYNC)            if start < 0:                # keep a trailing 0xB5 in case the 0x62 is in the next chunk                self.buf = self.buf[-1:] if self.buf.endswith(b"\xb5") else bytearray()                return out            if start:                del self.buf[:start]            if len(self.buf) < 6:                return out            length = struct.unpack_from("<H", self.buf, 4)[0]            if length > 2048:  # garbage that happened to look like a header                del self.buf[:2]                continue            total = 6 + length + 2            if len(self.buf) < total:                return out            body = bytes(self.buf[2:6 + length])            if checksum(body) == bytes(self.buf[6 + length:total]):                out.append((body[0], body[1], body[4:]))                del self.buf[:total]            else:                del self.buf[:2]

The phone app's screens: Live, Runs, Settings, Sensors.

<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover"><meta name="theme-color" content="#0f1216"><title>Mini Bike Dyno</title><style>  :root {    --bg: #0f1216; --card: #181d24; --card2: #20262f; --line: #2c3440; --text: #eef2f6;    --dim: #93a1b0; --accent: #ff8a1f; --accent2: #3fa9f5; --good: #34c77b; --warn: #f5c542;    --bad: #ef5350;  }  * { box-sizing: border-box; }  html, body { margin: 0; background: var(--bg); color: var(--text);    font: 16px/1.4 system-ui, -apple-system, "Segoe UI", Roboto, sans-serif; }  body { padding-bottom: calc(72px + env(safe-area-inset-bottom)); }  header { position: sticky; top: 0; z-index: 5; background: var(--bg); padding: 14px 16px 10px;    display: flex; align-items: center; gap: 10px; border-bottom: 1px solid var(--line); }  header h1 { font-size: 19px; margin: 0; flex: 1; letter-spacing: .3px; }  .dot { width: 10px; height: 10px; border-radius: 50%; background: var(--bad); }  .dot.on { background: var(--good); }  #conn { font-size: 13px; color: var(--dim); }  main { padding: 14px 16px; max-width: 760px; margin: 0 auto; }  nav { position: fixed; bottom: 0; left: 0; right: 0; display: flex; background: #12161c;    border-top: 1px solid var(--line); padding-bottom: env(safe-area-inset-bottom); z-index: 6; }  nav button { flex: 1; background: none; border: 0; color: var(--dim); padding: 10px 0 12px;    font: inherit; font-size: 13px; }  nav button .i { display: block; font-size: 21px; line-height: 1.2; }  nav button.on { color: var(--accent); }  section { display: none; } section.on { display: block; }  .card { background: var(--card); border: 1px solid var(--line); border-radius: 14px; padding: 14px;    margin-bottom: 12px; }  .speed { text-align: center; padding: 18px 10px 12px; }  .speed .n { font-size: 84px; font-weight: 700; line-height: 1; font-variant-numeric: tabular-nums; }  .speed .u { color: var(--dim); font-size: 16px; }  .state { text-align: center; margin-top: 10px; font-weight: 600; }  .state.rec { color: var(--bad); } .state.ok { color: var(--good); } .state.wait { color: var(--warn); }  .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(140px, 1fr)); gap: 10px; }  .tile { background: var(--card2); border-radius: 12px; padding: 10px 12px; }  .tile .k { color: var(--dim); font-size: 12px; text-transform: uppercase; letter-spacing: .4px; }  .tile .v { font-size: 24px; font-weight: 700; font-variant-numeric: tabular-nums; }  .tile .v small { font-size: 13px; color: var(--dim); font-weight: 500; }  .tile.big .v { color: var(--accent); font-size: 30px; }  button.btn { width: 100%; border: 0; border-radius: 12px; padding: 14px; font: inherit; font-weight: 700;    background: var(--accent); color: #1a0e00; margin-top: 10px; }  button.btn.sec { background: var(--card2); color: var(--text); border: 1px solid var(--line); }  button.btn.danger { background: #3a1c1f; color: #ff8a86; border: 1px solid #5a2a2e; }  button.small { width: auto; padding: 8px 12px; font-size: 14px; margin: 0; }  .row { display: flex; gap: 8px; align-items: center; }  .row > * { flex: 1; }  .banner { background: #2b2410; border: 1px solid #5c4a14; color: #ffe08a; border-radius: 12px;    padding: 10px 12px; margin-bottom: 12px; font-size: 14px; }  .banner.info { background: #10222f; border-color: #1d4561; color: #a9d8ff; }  .runs .run { display: block; width: 100%; text-align: left; font: inherit; color: inherit; cursor: pointer; }  .run .when { font-weight: 700; }  .run .name { color: var(--dim); font-size: 14px; }  .chips { display: flex; gap: 6px; flex-wrap: wrap; margin-top: 8px; }  .chip { background: var(--card2); border-radius: 999px; padding: 3px 10px; font-size: 13px; }  .chip.hp { background: #3a2410; color: var(--accent); font-weight: 700; }  h2 { font-size: 17px; margin: 18px 2px 10px; }  h3 { font-size: 15px; margin: 0 0 8px; color: var(--dim); font-weight: 600; }  canvas { width: 100%; height: 230px; display: block; }  label.f { display: block; margin-bottom: 12px; }  label.f span { display: block; color: var(--dim); font-size: 13px; margin-bottom: 4px; }  label.f em { display: block; color: var(--dim); font-size: 12px; font-style: normal; margin-top: 3px; }  input, select { width: 100%; background: var(--card2); color: var(--text); border: 1px solid var(--line);    border-radius: 10px; padding: 11px 12px; font: inherit; }  details summary { cursor: pointer; color: var(--accent2); margin: 6px 0 12px; }  .sensor .top { display: flex; align-items: center; gap: 10px; }  .sensor .top b { flex: 1; }  .sensor .st { font-size: 14px; margin-top: 6px; }  .st.ok { color: var(--good); } .st.bad { color: var(--bad); } .st.wait { color: var(--warn); }  .toggle { width: 46px; height: 28px; border-radius: 999px; background: var(--line); position: relative;    border: 0; flex: none !important; }  .toggle::after { content: ""; position: absolute; top: 3px; left: 3px; width: 22px; height: 22px;    border-radius: 50%; background: #fff; transition: left .15s; }  .toggle.on { background: var(--good); } .toggle.on::after { left: 21px; }  .muted { color: var(--dim); font-size: 14px; }  .warnings li { color: var(--warn); font-size: 14px; margin-bottom: 4px; }  .back { background: none; border: 0; color: var(--accent2); font: inherit; padding: 0 0 10px; }  .legend { display: flex; gap: 14px; flex-wrap: wrap; font-size: 13px; color: var(--dim); margin-top: 6px; }  .legend i { display: inline-block; width: 14px; height: 3px; vertical-align: middle; margin-right: 5px; }  .empty { text-align: center; color: var(--dim); padding: 30px 10px; }</style></head><body><header>  <span class="dot" id="dot"></span>  <h1>Mini Bike Dyno</h1>  <span id="conn">connecting...</span></header><main>  <!-- LIVE -->  <section id="live" class="on">    <div id="liveBanners"></div>    <div class="card speed">      <div class="n" id="spd">0</div>      <div class="u">mph</div>      <div class="state wait" id="state">Starting...</div>    </div>    <div class="grid" id="liveTiles"></div>    <button class="btn sec" id="recBtn">Start recording by hand</button>    <p class="muted" style="margin:8px 2px 0">You don't need this button: a run records by itself when      you go faster than <span id="startMph">3</span> mph, and saves when you slow down.</p>    <h2>Latest run</h2>    <div id="latest"><div class="empty">No runs yet.</div></div>  </section>  <!-- RUNS -->  <section id="runs">    <div id="runList" class="runs"></div>    <div id="runDetail" style="display:none"></div>  </section>  <!-- SETTINGS -->  <section id="settings">    <div class="banner info">These are used to work out horsepower. Changing them re-calculates every run.</div>    <div class="card" id="settingsForm"></div>    <button class="btn" id="saveSettings">Save settings</button>    <p class="muted" id="settingsMsg"></p>  </section>  <!-- SENSORS -->  <section id="sensors">    <div id="sensorList"></div>    <button class="btn sec" id="addSensorBtn">+ Add a sensor</button>    <div class="card" id="addSensor" style="display:none; margin-top:12px"></div>    <p class="muted" style="margin-top:14px">New sensors start working as soon as you save. Wheel speed      from an RPM sensor is used instead of GPS speed when it's there (Settings &rarr; Advanced &rarr;      Speed from).</p>  </section></main><nav>  <button data-tab="live" class="on"><span class="i">&#9201;</span>Live</button>  <button data-tab="runs"><span class="i">&#128200;</span>Runs</button>  <button data-tab="settings"><span class="i">&#9881;</span>Settings</button>  <button data-tab="sensors"><span class="i">&#128268;</span>Sensors</button></nav><script>"use strict";const $ = (id) => document.getElementById(id);const api = async (path, opts) => {  const r = await fetch(path, opts);  if (!r.ok && r.status !== 404 && r.status !== 400) throw new Error("HTTP " + r.status);  return r.json();};const post = (path, data, method) => api(path, { method: method || "POST",  headers: { "Content-Type": "application/json" }, body: JSON.stringify(data || {}) });const fmt = (x, d = 1, unit = "") => (x === null || x === undefined || Number.isNaN(x)) ? "&ndash;"  : Number(x).toFixed(d) + (unit ? "<small> " + unit + "</small>" : "");const esc = (s) => String(s ?? "").replace(/[&<>"']/g, (c) => ({ "&": "&amp;", "<": "&lt;", ">": "&gt;", '"': "&quot;", "'": "&#39;" }[c]));let tab = "live", settings = null, sensorData = null, runsCache = [], liveTimer = null, connected = false;// ---------------- tabs ----------------document.querySelectorAll("nav button").forEach((b) => b.onclick = () => showTab(b.dataset.tab));function showTab(name) {  tab = name;  document.querySelectorAll("nav button").forEach((b) => b.classList.toggle("on", b.dataset.tab === name));  document.querySelectorAll("section").forEach((s) => s.classList.toggle("on", s.id === name));  if (name === "runs") { $("runDetail").style.display = "none"; $("runList").style.display = ""; loadRuns(); }  if (name === "settings") loadSettings();  if (name === "sensors") loadSensors();  if (name === "live") { loadLatest(); }  window.scrollTo(0, 0);}function setConnected(ok) {  connected = ok;  $("dot").classList.toggle("on", ok);  $("conn").textContent = ok ? "connected" : "can't reach the Pi";}// ---------------- live ----------------async function pollLive() {  try {    const live = await api("/api/live");    setConnected(true);    renderLive(live);  } catch (e) { setConnected(false); }  liveTimer = setTimeout(pollLive, tab === "live" ? 400 : 2000);}let lastRunSeen = null;function renderLive(l) {  $("spd").textContent = Math.round(l.speed_mph || 0);  const st = $("state");  const gps = l.gps;  if (l.recording) {    st.className = "state rec";    st.textContent = (l.manual ? "Recording (by hand) " : "Recording run ") + (l.run_seconds || 0).toFixed(0) + " s";  } else if (l.speed_from === "wheel" || (gps && gps.fix_ok)) {    st.className = "state ok"; st.textContent = "Ready - ride whenever you like";  } else if (gps) {    st.className = "state wait"; st.textContent = "GPS looking for satellites (needs open sky)";  } else {    st.className = "state wait"; st.textContent = "No speed signal yet - see Sensors";  }  const tiles = [];  if (gps) {    tiles.push(tile("GPS", gps.fix_ok ? "3D fix" : "no fix", gps.sats + " satellites"));    tiles.push(tile("GPS updates", fmt(gps.rate_hz, 0, "per sec"), "&plusmn;" + (gps.speed_acc_mph ?? 0).toFixed(2) + " mph"));  }  if (l.wheel_rpm !== undefined) tiles.push(tile("Wheel", fmt(l.wheel_rpm, 0, "rpm")));  if (l.engine_rpm !== undefined) tiles.push(tile("Engine", fmt(l.engine_rpm, 0, "rpm")));  if (l.accel_g !== undefined) tiles.push(tile("Acceleration", fmt(l.accel_g, 2, "g")));  if (l.recording) tiles.push(tile("Top this run", fmt(l.run_top_mph, 1, "mph")));  $("liveTiles").innerHTML = tiles.join("");  $("recBtn").textContent = l.recording ? "Stop and save this run" : "Start recording by hand";  $("recBtn").className = l.recording ? "btn danger" : "btn sec";  $("recBtn").dataset.rec = l.recording ? "1" : "";  if (l.last_run && l.last_run !== lastRunSeen) { lastRunSeen = l.last_run; loadLatest(); }}function tile(k, v, sub) {  return `<div class="tile"><div class="k">${k}</div><div class="v">${v}</div>${sub ? `<div class="muted">${sub}</div>` : ""}</div>`;}$("recBtn").onclick = async () => {  if ($("recBtn").dataset.rec) await post("/api/record/stop"); else await post("/api/record/start");};async function loadLatest() {  try {    const data = await api("/api/runs");    runsCache = data.runs || [];    $("latest").innerHTML = runsCache.length ? runCard(runsCache[0]) : '<div class="empty">No runs yet. Ride!</div>';    bindRunCards($("latest"));  } catch (e) {}}async function loadBanners() {  try {    const [s, st] = await Promise.all([api("/api/settings"), api("/api/status")]);    settings = s.settings;    $("startMph").textContent = settings.start_mph;    const b = [];    if (!settings.weights_set) b.push(`<div class="banner">Enter your bike's and your weight in <b>Settings</b> so the horsepower is right.</div>`);    for (const sensor of st.sensors || []) {      if (sensor.type === "sim_gps") b.push(`<div class="banner info">Practice mode is on: the runs are pretend. Turn off "${esc(sensor.name)}" in Sensors before riding.</div>`);      else if (sensor.state === "no data" || sensor.state === "error") b.push(`<div class="banner">${esc(sensor.name)}: ${esc(sensor.detail || sensor.state)}</div>`);    }    $("liveBanners").innerHTML = b.join("");  } catch (e) {}}// ---------------- runs ----------------function runTitle(r) {  const d = new Date(r.started.replace(" ", "T"));  return d.toLocaleDateString(undefined, { weekday: "short", month: "short", day: "numeric" }) + " " +    d.toLocaleTimeString(undefined, { hour: "numeric", minute: "2-digit", second: "2-digit" });}function runCard(r) {  const s = r.summary || {};  return `<button class="card run" data-id="${r.id}">    <div class="when">${esc(runTitle(r))}${r.manual ? " (by hand)" : ""}</div>    ${r.name ? `<div class="name">${esc(r.name)}</div>` : ""}    <div class="chips">      ${r.ok ? `<span class="chip hp">${fmt(s.peak_engine_hp, 1)} hp</span>` : `<span class="chip">no power reading</span>`}      ${s.zero_to_20_s != null ? `<span class="chip">0-20 mph ${fmt(s.zero_to_20_s, 2)} s</span>` : ""}      ${s.top_mph != null ? `<span class="chip">top ${fmt(s.top_mph, 1)} mph</span>` : ""}    </div></button>`;}function bindRunCards(el) {  el.querySelectorAll(".run").forEach((c) => c.onclick = () => { if (tab !== "runs") showTab("runs"); openRun(c.dataset.id); });}async function loadRuns() {  $("runList").innerHTML = '<div class="empty">Loading...</div>';  try {    const data = await api("/api/runs");    runsCache = data.runs || [];    $("runList").innerHTML = runsCache.length      ? `<p class="muted" style="margin:0 2px 10px">${runsCache.length} run${runsCache.length > 1 ? "s" : ""}. Tip: do one run each way on the driveway and compare them - the average cancels out slope and wind.</p>` + runsCache.map(runCard).join("")      : '<div class="empty">No runs yet.<br>Clip the Pi to the bike and ride - runs save by themselves.</div>';    bindRunCards($("runList"));  } catch (e) { $("runList").innerHTML = '<div class="empty">Can\'t reach the Pi.</div>'; }}let current = null, compare = null;async function openRun(id) {  $("runList").style.display = "none";  const box = $("runDetail");  box.style.display = "";  box.innerHTML = '<div class="empty">Working it out...</div>';  current = await api("/api/runs/" + id);  compare = null;  renderRun();  window.scrollTo(0, 0);}function renderRun() {  const { run, result } = current;  const s = result.summary || {};  const others = runsCache.filter((r) => r.id !== run.id && r.ok);  const box = $("runDetail");  box.innerHTML = `    <button class="back" id="backBtn">&lsaquo; All runs</button>    <div class="card">      <div class="row"><div><b>${esc(runTitle(run))}</b>${run.name ? `<div class="muted">${esc(run.name)}</div>` : ""}</div>      <button class="btn sec small" id="renameBtn" style="flex:none">Rename</button></div>    </div>    ${result.ok ? "" : `<div class="banner">${esc(result.error)}</div>`}    ${(result.warnings || []).length ? `<ul class="warnings card">${result.warnings.map((w) => `<li>${esc(w)}</li>`).join("")}</ul>` : ""}    <div class="grid">      <div class="tile big"><div class="k">Engine hp (estimate)</div><div class="v">${fmt(s.peak_engine_hp, 2)}</div><div class="muted">at ${fmt(s.peak_hp_at_mph, 0)} mph${s.peak_hp_at_rpm ? ", " + s.peak_hp_at_rpm + " rpm" : ""}</div></div>      <div class="tile"><div class="k">Wheel hp</div><div class="v">${fmt(s.peak_wheel_hp, 2)}</div><div class="muted">${Math.round((s.efficiency || 0) * 100)}% gets through</div></div>      <div class="tile"><div class="k">Top speed</div><div class="v">${fmt(s.top_mph, 1, "mph")}</div></div>      <div class="tile"><div class="k">0-10 mph</div><div class="v">${fmt(s.zero_to_10_s, 2, "s")}</div></div>      <div class="tile"><div class="k">0-20 mph</div><div class="v">${fmt(s.zero_to_20_s, 2, "s")}</div></div>      <div class="tile"><div class="k">0-30 mph</div><div class="v">${fmt(s.zero_to_30_s, 2, "s")}</div></div>      <div class="tile"><div class="k">60 ft</div><div class="v">${fmt(s.sixty_ft_s, 2, "s")}</div></div>      <div class="tile"><div class="k">Max accel</div><div class="v">${fmt(s.max_accel_g, 2, "g")}</div></div>      ${s.peak_torque_lbft ? `<div class="tile"><div class="k">Peak torque</div><div class="v">${fmt(s.peak_torque_lbft, 1, "lb-ft")}</div></div>` : ""}      <div class="tile"><div class="k">Distance</div><div class="v">${fmt(s.distance_ft, 0, "ft")}</div></div>    </div>    ${others.length ? `<div class="card" style="margin-top:12px"><label class="f" style="margin:0"><span>Compare with another run</span>      <select id="cmpSel"><option value="">(none)</option>${others.map((r) => `<option value="${r.id}" ${compare && compare.run.id === r.id ? "selected" : ""}>${esc(runTitle(r))} - ${fmt(r.summary.peak_engine_hp, 1)} hp</option>`).join("")}</select></label></div>` : ""}    <div class="card" style="margin-top:12px"><h3>Horsepower by speed</h3><canvas id="cHp"></canvas><div class="legend" id="lHp"></div></div>    ${result.curve_by_rpm ? `<div class="card"><h3>Engine power and torque by RPM</h3><canvas id="cRpm"></canvas><div class="legend" id="lRpm"></div></div>` : ""}    <div class="card"><h3>Speed</h3><canvas id="cSpd"></canvas><div class="legend" id="lSpd"></div></div>    <div class="card"><h3>Acceleration</h3><canvas id="cAcc"></canvas></div>    <p class="muted">Speed from ${esc(s.speed_source || "?")}, acceleration from ${esc(s.accel_source || "?")}.      Weight used: ${fmt(s.mass_lb, 0)} lb (bike + rider).</p>    <div class="row"><button class="btn sec" id="csvBtn">Download as spreadsheet</button>      <button class="btn danger" id="delBtn">Delete run</button></div>`;  $("backBtn").onclick = () => { $("runDetail").style.display = "none"; $("runList").style.display = ""; loadRuns(); };  $("renameBtn").onclick = async () => {    const name = prompt("Name this run (e.g. 'new carb, uphill')", run.name || "");    if (name === null) return;    await post("/api/runs/" + run.id, { name });    current.run.name = name; renderRun();  };  $("delBtn").onclick = async () => {    if (!confirm("Delete this run for good?")) return;    await api("/api/runs/" + run.id, { method: "DELETE" });    $("backBtn").onclick();  };  $("csvBtn").onclick = () => { window.location.href = "/api/runs/" + run.id + ".csv"; };  if ($("cmpSel")) $("cmpSel").onchange = async (e) => {    compare = e.target.value ? await api("/api/runs/" + e.target.value) : null;    renderRun();  };  drawRunCharts();}const COL = { a: "#ff8a1f", b: "#3fa9f5", c: "#34c77b", d: "#c58af9" };function drawRunCharts() {  const { result } = current;  if (!result.ok) return;  const c = result.curve_by_speed, cmp = compare && compare.result.ok ? compare.result : null;  const hp = [{ x: c.mph, y: c.hp_engine, color: COL.a, label: "Engine hp (estimate)" },              { x: c.mph, y: c.hp_wheel, color: COL.a, dash: [5, 4], label: "Wheel hp" }];  if (cmp) hp.push({ x: cmp.curve_by_speed.mph, y: cmp.curve_by_speed.hp_engine, color: COL.b, label: "Compared run" });  chart("cHp", "lHp", hp, "mph", "hp");  const sp = [{ x: result.series.t, y: result.series.mph, color: COL.c, label: "This run" }];  if (cmp) sp.push({ x: cmp.series.t, y: cmp.series.mph, color: COL.b, label: "Compared run" });  chart("cSpd", "lSpd", sp, "seconds", "mph");  chart("cAcc", null, [{ x: result.series.t, y: result.series.accel_g, color: COL.d }], "seconds", "g");  if (result.curve_by_rpm) {    const r = result.curve_by_rpm;    chart("cRpm", "lRpm", [{ x: r.rpm, y: r.hp, color: COL.a, label: "hp" },                           { x: r.rpm, y: r.torque_lbft, color: COL.b, label: "lb-ft" }], "rpm", "");  }}// a small line chart: series = [{x:[], y:[], color, dash, label}]function chart(canvasId, legendId, series, xLabel, yLabel) {  const cv = $(canvasId); if (!cv) return;  const dpr = window.devicePixelRatio || 1, W = cv.clientWidth, H = cv.clientHeight;  cv.width = W * dpr; cv.height = H * dpr;  const g = cv.getContext("2d"); g.scale(dpr, dpr);  const xs = series.flatMap((s) => s.x), ys = series.flatMap((s) => s.y).filter((v) => v != null);  if (!xs.length || !ys.length) { g.fillStyle = "#93a1b0"; g.fillText("No data", 10, 20); return; }  let x0 = Math.min(...xs), x1 = Math.max(...xs), y0 = Math.min(0, ...ys), y1 = Math.max(...ys);  if (x1 === x0) x1 = x0 + 1; if (y1 === y0) y1 = y0 + 1;  const tx = ticks(x0, x1, 5), ty = ticks(y0, y1, 4);  x0 = Math.min(x0, tx[0]); x1 = Math.max(x1, tx[tx.length - 1]);  y0 = Math.min(y0, ty[0]); y1 = Math.max(y1, ty[ty.length - 1]);  const L = 44, R = 10, T = 10, B = 30;  const px = (x) => L + (x - x0) / (x1 - x0) * (W - L - R);  const py = (y) => H - B - (y - y0) / (y1 - y0) * (H - T - B);  g.font = "11px system-ui, sans-serif"; g.strokeStyle = "#2c3440"; g.fillStyle = "#93a1b0"; g.lineWidth = 1;  ty.forEach((v) => { g.beginPath(); g.moveTo(L, py(v)); g.lineTo(W - R, py(v)); g.stroke();    g.textAlign = "right"; g.fillText(+v.toFixed(2), L - 6, py(v) + 4); });  tx.forEach((v) => { g.textAlign = "center"; g.fillText(+v.toFixed(2), px(v), H - B + 16); });  g.textAlign = "right"; g.fillText(xLabel, W - R, H - 2);  g.textAlign = "left"; g.fillText(yLabel, 2, 10);  series.forEach((s) => {    g.strokeStyle = s.color; g.lineWidth = 2.2; g.setLineDash(s.dash || []); g.beginPath();    let started = false;    s.x.forEach((x, i) => { const y = s.y[i]; if (y == null) return;      if (!started) { g.moveTo(px(x), py(y)); started = true; } else g.lineTo(px(x), py(y)); });    g.stroke(); g.setLineDash([]);  });  if (legendId) $(legendId).innerHTML = series.filter((s) => s.label)    .map((s) => `<span><i style="${s.dash ? `border-top:3px dashed ${s.color};height:0` : `background:${s.color}`}"></i>${esc(s.label)}</span>`).join("");}function ticks(a, b, n) {  const step0 = (b - a) / n, mag = Math.pow(10, Math.floor(Math.log10(step0)));  const step = [1, 2, 2.5, 5, 10].map((m) => m * mag).find((s) => s >= step0) || step0;  const out = []; for (let v = Math.floor(a / step) * step; v <= b + step * 0.001; v += step) out.push(+v.toFixed(6));  if (out[out.length - 1] < b) out.push(out[out.length - 1] + step);  return out;}// ---------------- settings ----------------const FIELDS = [  ["bike_weight_lb", "Mini bike weight (lb)", "number", "With fuel. A bathroom scale under each wheel works."],  ["rider_weight_lb", "Rider weight (lb)", "number", "With helmet and gear."],  ["drivetrain", "Drivetrain", "select", "How the engine drives the wheel.",    [["clutch", "Centrifugal clutch + chain"], ["torque_converter", "Torque converter"]]],  ["tire_diameter_in", "Rear tire height (in)", "number", "Ground to top of tire. Used with wheel RPM sensors."],];const ADVANCED = [  ["slope_pct", "Driveway slope (%)", "number", "Uphill in the riding direction = positive. Easier: run both ways and compare."],  ["efficiency_clutch", "Clutch + chain efficiency", "number", "Share of engine power reaching the tire (0.85 = 85%)."],  ["efficiency_torque_converter", "Torque converter efficiency", "number", ""],  ["cda_m2", "Air drag area CdA (m²)", "number", "About 0.5 for an upright rider."],  ["crr", "Tire rolling resistance", "number", "About 0.02 on pavement, 0.04 on dirt/grass."],  ["rotating_pct", "Spinning parts (%)", "number", "Extra effective weight from wheels, clutch, chain."],  ["smoothing", "Smoothing (GPS readings)", "number", "Higher = smoother curves, less detail."],  ["start_mph", "Run starts above (mph)", "number", ""],  ["stop_mph", "Run ends below (mph)", "number", ""],  ["min_top_mph", "Ignore runs slower than (mph)", "number", ""],  ["speed_source", "Speed from", "select", "Auto uses a wheel RPM sensor when you have one.",    [["auto", "Automatic"], ["gps", "GPS only"], ["wheel", "Wheel sensor only"]]],];function field([key, label, kind, help, choices], value) {  const input = kind === "select"    ? `<select data-k="${key}">${choices.map(([v, t]) => `<option value="${v}" ${String(value) === v ? "selected" : ""}>${t}</option>`).join("")}</select>`    : `<input data-k="${key}" type="number" step="any" inputmode="decimal" value="${esc(value)}">`;  return `<label class="f"><span>${label}</span>${input}${help ? `<em>${help}</em>` : ""}</label>`;}async function loadSettings() {  const data = await api("/api/settings");  settings = data.settings;  $("settingsForm").innerHTML = FIELDS.map((f) => field(f, settings[f[0]])).join("") +    `<details><summary>Advanced</summary>${ADVANCED.map((f) => field(f, settings[f[0]])).join("")}</details>`;  $("settingsMsg").textContent = "";}$("saveSettings").onclick = async () => {  const changes = {};  document.querySelectorAll("#settingsForm [data-k]").forEach((el) => {    const k = el.dataset.k, v = el.value;    if (String(settings[k]) !== v) changes[k] = el.tagName === "SELECT" ? v : Number(v);  });  const data = await post("/api/settings", changes);  settings = data.settings;  $("settingsMsg").textContent = "Saved. All runs now use these numbers.";  loadBanners();};// ---------------- sensors ----------------async function loadSensors() {  sensorData = await api("/api/sensors");  renderSensors();}function statusFor(id) { return (sensorData.status || []).find((s) => s.id === id); }function renderSensors() {  const { sensors, types } = sensorData;  $("sensorList").innerHTML = sensors.length ? sensors.map((s, i) => {    const st = statusFor(s.id), t = types[s.type] || { label: s.type };    const cls = !s.enabled ? "" : !st ? "wait" : st.state === "ok" ? "ok" : (st.state === "error" || st.state === "no data") ? "bad" : "wait";    const text = !s.enabled ? "Off" : st ? `${st.state}${st.detail ? " - " + st.detail : ""}` : "starting...";    return `<div class="card sensor"><div class="top"><b>${esc(s.name)}</b>        <button class="toggle ${s.enabled ? "on" : ""}" data-i="${i}" aria-label="on/off"></button></div>      <div class="muted">${esc(t.label)}</div><div class="st ${cls}">${esc(text)}</div>      <div class="row" style="margin-top:10px"><button class="btn sec small" data-edit="${i}">Change</button>        <button class="btn danger small" data-del="${i}">Remove</button></div></div>`;  }).join("") : '<div class="empty">No sensors yet.</div>';  document.querySelectorAll("#sensorList .toggle").forEach((b) => b.onclick = () => {    sensors[b.dataset.i].enabled = !sensors[b.dataset.i].enabled; saveSensors();  });  document.querySelectorAll("#sensorList [data-del]").forEach((b) => b.onclick = () => {    if (!confirm("Remove " + sensors[b.dataset.del].name + "?")) return;    sensors.splice(b.dataset.del, 1); saveSensors();  });  document.querySelectorAll("#sensorList [data-edit]").forEach((b) => b.onclick = () => sensorForm(+b.dataset.edit));}async function saveSensors() {  const data = await post("/api/sensors", { sensors: sensorData.sensors });  if (data.sensors) sensorData.sensors = data.sensors;  sensorData.status = data.status || [];  renderSensors(); loadBanners();  setTimeout(async () => { sensorData = await api("/api/sensors"); renderSensors(); }, 2500);}$("addSensorBtn").onclick = () => sensorForm(-1);function sensorForm(index) {  const types = sensorData.types;  const existing = index >= 0 ? sensorData.sensors[index] : null;  const box = $("addSensor");  box.style.display = "";  const typeName = existing ? existing.type : Object.keys(types)[0];  const render = (type) => {    const t = types[type];    const values = existing && existing.type === type ? existing : {};    box.innerHTML = `<h3>${existing ? "Change sensor" : "Add a sensor"}</h3>      ${existing ? "" : `<label class="f"><span>Kind of sensor</span><select id="sfType">${Object.entries(types).map(([k, v]) => `<option value="${k}" ${k === type ? "selected" : ""}>${esc(v.label)}</option>`).join("")}</select></label>`}      ${t.help ? `<div class="banner info">${esc(t.help)}</div>` : ""}      <label class="f"><span>Name</span><input id="sfName" value="${esc(values.name || t.label)}"></label>      ${t.options.map((o) => {        const v = values[o.key] ?? o.default;        const input = o.kind === "select"          ? `<select data-o="${o.key}">${o.choices.map((c) => `<option ${String(c) === String(v) ? "selected" : ""}>${esc(c)}</option>`).join("")}</select>`          : `<input data-o="${o.key}" ${o.kind === "number" ? 'type="number" step="any" inputmode="decimal"' : ""} value="${esc(v)}">`;        return `<label class="f"><span>${esc(o.label)}</span>${input}${o.help ? `<em>${esc(o.help)}</em>` : ""}</label>`;      }).join("")}      <div class="row"><button class="btn sec" id="sfCancel">Cancel</button><button class="btn" id="sfSave">${existing ? "Save" : "Add"}</button></div>`;    if ($("sfType")) $("sfType").onchange = (e) => render(e.target.value);    $("sfCancel").onclick = () => { box.style.display = "none"; };    $("sfSave").onclick = () => {      const entry = existing ? { ...existing } : { id: type + Date.now().toString(36).slice(-4), type, enabled: true };      entry.name = $("sfName").value;      box.querySelectorAll("[data-o]").forEach((el) => entry[el.dataset.o] = el.value);      if (existing) sensorData.sensors[index] = entry; else sensorData.sensors.push(entry);      box.style.display = "none";      saveSensors();    };  };  render(typeName);  box.scrollIntoView({ behavior: "smooth" });}// ---------------- start ----------------pollLive();loadBanners();loadLatest();setInterval(loadBanners, 8000);</script></body></html>

13 self-tests, including a pretend ride.

"""Self-tests: run with  python3 -m unittest discover -s tests  (from the pi folder).The key test fakes a ride with a known engine horsepower (same physics as theanalysis, plus GPS-like noise) and checks the analysis finds that horsepower again."""import jsonimport osimport randomimport structimport sysimport tempfileimport unittestsys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))os.environ["DYNO_DATA"] = tempfile.mkdtemp(prefix="dyno-test-")from dyno import analysis, config, recorder, server, ubx  # noqa: E402MPH = analysis.MPS_TO_MPHdef fake_run(engine_hp, noise=0.05, rate_hz=10, settings=None, seed=1):    random.seed(seed)    sim = analysis.simulate_run(engine_hp=engine_hp, seconds=9.0, settings=settings,                                dt_out=1.0 / rate_hz)    speeds = [max(0.0, v + random.gauss(0, noise)) for v in sim["speed_mps"]]    pre = [0.0] * (2 * rate_hz)  # standing still before the launch    n = len(pre) + len(speeds)    t = [round(i / rate_hz, 3) for i in range(n)]    return {"channels": {"gps": {"t": t, "speed_mps": pre + speeds, "ok": [True] * n,                                 "sats": [15] * n}}}, simclass TestHorsepower(unittest.TestCase):    def settings(self, **changes):        s = dict(config.DEFAULT_SETTINGS)        s["weights_set"] = True        s.update(changes)        return s    def test_finds_the_horsepower_it_was_given(self):        for hp in (3.0, 6.5, 10.0):            run, sim = fake_run(hp)            result = analysis.analyze(run, self.settings())            self.assertTrue(result["ok"], result)            found = result["summary"]["peak_engine_hp"]            print("  pretend %.1f hp engine -> measured %.2f hp (wheel %.2f), 0-20 mph %s s, top %s mph"                  % (hp, found, result["summary"]["peak_wheel_hp"],                     result["summary"]["zero_to_20_s"], result["summary"]["top_mph"]))            self.assertAlmostEqual(found, hp, delta=hp * 0.07)    def test_torque_converter_setting(self):        s = self.settings(drivetrain="torque_converter")        run, _ = fake_run(6.5, settings=s)        found = analysis.analyze(run, s)["summary"]["peak_engine_hp"]        self.assertAlmostEqual(found, 6.5, delta=0.5)    def test_heavier_rider_means_more_power_for_the_same_speeds(self):        run, _ = fake_run(6.5)        light = analysis.analyze(run, self.settings(rider_weight_lb=150))["summary"]["peak_engine_hp"]        heavy = analysis.analyze(run, self.settings(rider_weight_lb=200))["summary"]["peak_engine_hp"]        self.assertGreater(heavy, light * 1.1)    def test_slow_gps_still_works(self):        run, _ = fake_run(6.5, rate_hz=5)        result = analysis.analyze(run, self.settings())        self.assertAlmostEqual(result["summary"]["peak_engine_hp"], 6.5, delta=0.6)        self.assertTrue(any("updates a second" in w for w in result["warnings"]))    def test_wheel_sensor_is_used_when_present(self):        run, sim = fake_run(6.5)        s = self.settings()        circ = 3.14159265 * s["tire_diameter_in"] * 0.0254        gps = run["channels"]["gps"]        t20 = [i / 20 for i in range(len(gps["t"]) * 2)]        rpm = [max(0.0, analysis.resample(gps["t"], gps["speed_mps"], [tt])[0]) * 60 / circ for tt in t20]        run["channels"]["wheel_rpm"] = {"t": t20, "v": rpm}        result = analysis.analyze(run, s)        self.assertEqual(result["summary"]["speed_source"], "wheel sensor")        self.assertAlmostEqual(result["summary"]["peak_engine_hp"], 6.5, delta=0.6)    def test_no_movement(self):        n = 50        run = {"channels": {"gps": {"t": [i / 10 for i in range(n)], "speed_mps": [0.0] * n,                                    "ok": [True] * n}}}        self.assertFalse(analysis.analyze(run, self.settings())["ok"])class TestUbx(unittest.TestCase):    def test_frame_checksum_and_reader(self):        payload = bytes(92)        msg = ubx.frame(0x01, 0x07, payload)        reader = ubx.Reader()        got = reader.feed(b"junk$GPGGA,\r\n" + msg[:10]) + reader.feed(msg[10:] + b"\xb5")        self.assertEqual(got, [(0x01, 0x07, payload)])    def test_bad_checksum_is_skipped(self):        msg = bytearray(ubx.frame(0x01, 0x07, bytes(92)))        msg[-1] ^= 0xFF        self.assertEqual(ubx.Reader().feed(bytes(msg)), [])    def test_nav_pvt_decoding(self):        p = bytearray(92)        struct.pack_into("<I", p, 0, 123456)        struct.pack_into("<HBBBBB", p, 4, 2026, 10, 18, 20, 30, 15)        p[20], p[21], p[23] = 3, 0x01, 14            # 3D fix, fix OK, 14 satellites        struct.pack_into("<ii", p, 24, -841234567, 401234567)  # lon, lat        struct.pack_into("<i", p, 60, 12345)         # 12.345 m/s        struct.pack_into("<I", p, 68, 150)           # speed accuracy 0.15 m/s        pvt = ubx.parse_nav_pvt(bytes(p))        self.assertTrue(pvt["fix_ok"])        self.assertEqual(pvt["sats"], 14)        self.assertAlmostEqual(pvt["speed_mps"], 12.345)        self.assertAlmostEqual(pvt["lat"], 40.1234567)        self.assertAlmostEqual(pvt["speed_acc_mps"], 0.15)        self.assertEqual(pvt["utc"], "2026-10-18T20:30:15Z")    def test_valset_layout(self):        msg = ubx.valset({"CFG-RATE-MEAS": 100})        self.assertEqual(msg[2:4], b"\x06\x8a")        self.assertEqual(struct.unpack_from("<H", msg, 4)[0], 4 + 4 + 2)        self.assertEqual(struct.unpack_from("<IH", msg, 10), (0x30210001, 100))class TestRecorder(unittest.TestCase):    def test_records_a_run_by_itself_and_ignores_rolling_around(self):        rec = recorder.Recorder()        rec.settings["weights_set"] = True        before = set(os.listdir(config.RUNS_DIR)) if os.path.isdir(config.RUNS_DIR) else set()        def gps(t, v):            rec.emit("gps", t, {"speed_mps": v, "fix_ok": True, "fix": 3, "sats": 15,                                "speed_acc_mps": 0.1})        t = 1000.0        for v in [0.0] * 30 + [1.5] * 20 + [0.0] * 30:  # walking pace: not a run            gps(t, v)            t += 0.1        sim = analysis.simulate_run(engine_hp=6.5)        for v in [0.0] * 30 + sim["speed_mps"] + [0.0] * 30:            gps(t, v)            t += 0.1        after = set(os.listdir(config.RUNS_DIR)) - before        self.assertEqual(len(after), 1, after)        with open(os.path.join(config.RUNS_DIR, after.pop())) as f:            run = json.load(f)        self.assertGreater(run["top_mph_live"], 25)        self.assertGreaterEqual(run["channels"]["gps"]["t"][0], 0)        result = analysis.analyze(run, rec.settings)        self.assertTrue(result["ok"])        self.assertAlmostEqual(result["summary"]["peak_engine_hp"], 6.5, delta=0.6)    def test_unplugged_wheel_sensor_does_not_block_gps_runs(self):        rec = recorder.Recorder()        before = set(os.listdir(config.RUNS_DIR)) if os.path.isdir(config.RUNS_DIR) else set()        t = 5000.0        sim = analysis.simulate_run(engine_hp=6.5)        for v in [0.0] * 30 + sim["speed_mps"] + [0.0] * 30:            rec.emit("wheel_rpm", t, 0.0)  # sensor fitted but not seeing the magnet            rec.emit("gps", t, {"speed_mps": v, "fix_ok": True, "fix": 3, "sats": 15})            t += 0.1        self.assertEqual(len(set(os.listdir(config.RUNS_DIR)) - before), 1)class TestSensorList(unittest.TestCase):    def test_cleaning(self):        cleaned = server.clean_sensor_list([            {"type": "pulse_rpm", "name": "Rear wheel", "gpio": "27", "pulses_per_rev": 2,             "role": "wheel", "edge": "rising", "evil": "x"},            {"type": "not_a_sensor"},            {"type": "mpu6050", "address": "0x69", "forward": "sideways"},        ])        self.assertEqual(len(cleaned), 2)        self.assertEqual(cleaned[0]["gpio"], 27)        self.assertNotIn("evil", cleaned[0])        self.assertEqual(cleaned[1]["forward"], "x")  # unknown choice -> default        self.assertEqual(cleaned[1]["address"], "0x69")if __name__ == "__main__":    unittest.main(verbosity=2)