"""Sensors. Each type describes its own options, so the app can add and set them up without code changes. Every sensor runs in its own thread and sends readings to emit(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 math import random import threading import time from . import ubx SENSOR_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 register def 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]