"""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/ one run worked out with the current settings GET /api/runs/.csv the same as a spreadsheet file POST /api/runs/ {"name": "..."} to rename DELETE /api/runs/ 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 csv import io import json import os import re import socket import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from . import analysis, config, sensors VERSION = "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 Handler def 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 cleaned def serve(dyno, port=8080): server = ThreadingHTTPServer(("0.0.0.0", port), make_handler(dyno)) server.daemon_threads = True return server