import asyncio import json import logging from datetime import datetime, timezone logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("livetiming") GPBIKES_HOST = "127.0.0.1" GPBIKES_PORT = 54320 GPBIKES_PASSWORD = "" WS_HOST = "0.0.0.0" WS_PORT = 8765 KEEPALIVE_INTERVAL = 10 RECORDS_PATH = "/home/gpbikes/livetiming/records.json" KNOWN_TYPES = { "EVENT", "SESSION", "WEATHER", "ENTRY", "ENTRYREMOVE", "CLASSIFICATION", "BESTLAP", "LASTLAP", "LAP", "SPLIT", "SPEED", "PENALTY", "PENALTYCLEAR", "PENALTYCHANGE", "TRACKDATA", "TRACKSEGMENT", "TRACKPOSITION", "CONTACT", "END", } state = { "event": {}, "session": {}, "weather": {}, "entries": {}, "classification": [], "_raw_classification": [], "track_record": None, "leaderboard": [], "connected": False, } ws_clients = set() # houdt per (track|category, race_number) het laatst geziene rondenummer bij # (in-memory; reset bij herstart van deze service - een kleine, geaccepteerde # onnauwkeurigheid voor een hobby-server) last_lap_seen = {} def load_records(): try: with open(RECORDS_PATH) as f: content = f.read().strip() if not content: return {} return json.loads(content) except Exception: return {} def save_records(): try: with open(RECORDS_PATH, "w") as f: json.dump(records, f, indent=2) except Exception as e: log.error(f"Kon records.json niet opslaan: {e}") records = load_records() def fmt_ms(value): try: ms = int(value) except (TypeError, ValueError): return value or "--" if ms <= 0: return "--" minutes = ms // 60000 seconds = (ms % 60000) / 1000 return f"{minutes}:{seconds:06.3f}" def fmt_gap(value): try: ms = int(value) except (TypeError, ValueError): return value or "--" if ms == 0: return "--" seconds = ms / 1000 return f"+{seconds:.3f}" def record_key(): ev = state["event"] track = ev.get("track") or "unknown" category = ev.get("categories") or "Open" return f"{track}|{category}" def build_rows(): class_by_num = {r.get("race_number"): r for r in state["_raw_classification"]} rows = [] seen = set() for race_number, entry in state["entries"].items(): row = dict(class_by_num.get(race_number, {})) row["race_number"] = race_number row["name"] = entry.get("name") or f"#{race_number}" row.setdefault("gap", "--") row.setdefault("best_lap", "--") row.setdefault("bike_status", row.get("bike_status") or "--") rows.append(row) seen.add(race_number) for r in state["_raw_classification"]: if r.get("race_number") not in seen: rows.append(r) for row in rows: row["best_lap"] = fmt_ms(row.get("best_lap")) gap_raw = row.get("gap") if gap_raw and gap_raw not in ("--", "L"): row["gap"] = fmt_gap(gap_raw) return rows def update_records(key, bucket, rows): changed = False now = datetime.now(timezone.utc).isoformat(timespec="seconds") for row in rows: lap_raw = row.get("best_lap") if not lap_raw or lap_raw == "--": continue try: lap_ms = int(lap_raw) except ValueError: continue if lap_ms <= 0: continue race_number = row.get("race_number") name = row.get("name", f"#{race_number}") rider_rec = bucket["riders"].get(race_number) if rider_rec is None: bucket["riders"][race_number] = { "name": name, "lap_time_ms": lap_ms, "lap_time_str": fmt_ms(lap_ms), "date": now, "laps_ridden": 0, } changed = True elif lap_ms < rider_rec["lap_time_ms"]: rider_rec["lap_time_ms"] = lap_ms rider_rec["lap_time_str"] = fmt_ms(lap_ms) rider_rec["date"] = now rider_rec["name"] = name changed = True if bucket["best"] is None or lap_ms < bucket["best"]["lap_time_ms"]: bucket["best"] = { "name": name, "lap_time_ms": lap_ms, "lap_time_str": fmt_ms(lap_ms), "date": now, } changed = True return changed def update_lap_counts(key, bucket, rows): changed = False for row in rows: race_number = row.get("race_number") name = row.get("name", f"#{race_number}") lap_raw = row.get("lap_number") try: lap_num = int(lap_raw) except (TypeError, ValueError): continue seen_key = (key, race_number) prev = last_lap_seen.get(seen_key, 0) if lap_num > prev: delta = lap_num - prev rider_rec = bucket["riders"].get(race_number) if rider_rec is None: rider_rec = { "name": name, "lap_time_ms": None, "lap_time_str": "--", "date": None, "laps_ridden": 0, } bucket["riders"][race_number] = rider_rec rider_rec["laps_ridden"] = rider_rec.get("laps_ridden", 0) + delta rider_rec["name"] = name changed = True last_lap_seen[seen_key] = lap_num return changed def refresh_display(): rows = build_rows() state["classification"] = rows key = record_key() bucket = records.setdefault(key, {"best": None, "riders": {}}) changed = update_records(key, bucket, rows) changed = update_lap_counts(key, bucket, rows) or changed if changed: save_records() state["track_record"] = bucket.get("best") riders = [r for r in bucket.get("riders", {}).values() if r.get("lap_time_ms") is not None] riders.sort(key=lambda r: r["lap_time_ms"]) state["leaderboard"] = riders[:20] def broadcast(): refresh_display() if not ws_clients: return payload = json.dumps(state) dead = set() for ws in ws_clients: try: asyncio.create_task(ws.send(payload)) except Exception: dead.add(ws) ws_clients.difference_update(dead) class GPBikesProtocol(asyncio.DatagramProtocol): def __init__(self): self.transport = None def connection_made(self, transport): self.transport = transport log.info("UDP socket klaar, versturen CONNECT") self.send_lines(["CONNECT", GPBIKES_PASSWORD]) def send_lines(self, lines): data = ("\n".join(lines) + "\n").encode("utf-8") self.transport.sendto(data) def datagram_received(self, data, addr): try: text = data.decode("utf-8", errors="replace") except Exception: return lines = text.split("\n") log.debug(f"RAW: {lines}") if not lines: return head = lines[0] if head == "OK": log.info("Verbonden met GP Bikes live timing") state["connected"] = True self.send_lines(["START", "0", "0"]) broadcast() return if head == "FULL": log.error("Er is al een andere live timing client verbonden (FULL)") return if head == "WRONGPASSWORD": log.error("Verkeerd live timing wachtwoord") return if head == "ALIVE": return if head == "MSG": if len(lines) < 3: return msg_id = lines[1] rest = lines[2:] self.dispatch_blocks(rest) self.send_lines(["ACK", msg_id]) broadcast() return if head == "DATA": rest = lines[1:] self.dispatch_blocks(rest) broadcast() return def dispatch_blocks(self, tokens): it = iter(tokens) pending = None def nxt(default=""): nonlocal pending if pending is not None: val = pending pending = None return val return next(it, default) while True: tok = pending if pending is not None else next(it, None) pending = None if tok is None: break if tok in KNOWN_TYPES: self.handle_message(tok, it, nxt) def handle_message(self, msg_type, it, nxt): if msg_type == "EVENT": state["event"] = { "type": nxt(), "name": nxt(), "track": nxt(), "track_layout": nxt(), "_reserved": nxt(), "track_length": nxt(), "categories": nxt(), } nxt() elif msg_type == "SESSION": state["session"].update({ "session": nxt(), "state": nxt(), "length": nxt(), }) elif msg_type == "WEATHER": state["weather"] = { "conditions": nxt(), "air_temp": nxt(), "ground_temp": nxt(), } elif msg_type == "ENTRY": race_number = nxt() entry = { "name": nxt(), "bike_name": nxt(), "bike_short_name": nxt(), "categories": nxt(), "guid": nxt(), "extra": nxt(), } if race_number: state["entries"][race_number] = entry elif msg_type == "ENTRYREMOVE": race_number = nxt() state["entries"].pop(race_number, None) state["_raw_classification"] = [ c for c in state["_raw_classification"] if c.get("race_number") != race_number ] elif msg_type == "CLASSIFICATION": session = nxt() status = nxt() session_timer = nxt() session_length = nxt() session_lap = nxt() number_of_laps = nxt() classification_type = nxt() rows = [] if classification_type in ("1", "2"): nxt() number_of_entries = nxt() try: count = int(number_of_entries) except ValueError: count = 0 for _ in range(count): race_number = nxt() if race_number == "": break row = {"race_number": race_number} if classification_type == "1": row["best_lap"] = nxt() row["lap_number"] = nxt() row["total_laps"] = nxt() row["gap"] = nxt() row["speed_trap"] = nxt() else: row["race_time_or_status"] = nxt() row["lap_number"] = nxt() row["gap"] = nxt() row["bike_status"] = nxt() entry = state["entries"].get(race_number, {}) row["name"] = entry.get("name", f"#{race_number}") rows.append(row) state["_raw_classification"] = rows state["session"].update({ "session": session, "state": status, "session_timer": session_timer, "session_length": session_length, "session_lap": session_lap, "number_of_laps": number_of_laps, }) elif msg_type == "BESTLAP": race_number = nxt() nxt() lap_time = nxt() for row in state["_raw_classification"]: if row.get("race_number") == race_number: row["best_lap"] = lap_time break async def keepalive_loop(protocol): while True: await asyncio.sleep(KEEPALIVE_INTERVAL) if protocol.transport is not None: protocol.send_lines(["KEEPALIVE"]) async def ws_handler(websocket): ws_clients.add(websocket) try: refresh_display() await websocket.send(json.dumps(state)) async for _ in websocket: pass finally: ws_clients.discard(websocket) async def main(): import websockets loop = asyncio.get_running_loop() transport, protocol = await loop.create_datagram_endpoint( GPBikesProtocol, remote_addr=(GPBIKES_HOST, GPBIKES_PORT), ) asyncio.create_task(keepalive_loop(protocol)) async with websockets.serve(ws_handler, WS_HOST, WS_PORT): log.info(f"Websocket server luistert op ws://{WS_HOST}:{WS_PORT}") await asyncio.Future() if __name__ == "__main__": asyncio.run(main())