fpv-laptimer-client (Python)

Python-Client für die API des FPV-Laptimers (/api/v1, ausführlich in der Doku). Befehle laufen über HTTP und brauchen nur die Standardbibliothek; für Events (WebSocket) braucht es zusätzlich das Paket websockets. Python ab 3.10.

Installieren

pip install fpv-laptimer-client              # nur Befehle (HTTP)
pip install "fpv-laptimer-client[events]"    # dazu Events über den WebSocket

Aus dem Repo statt von PyPI: pip install "./clients/python[events]".

Oder ohne Installation: laptimer_client/__init__.py als laptimer_client.py neben das eigene Script kopieren. Befehle funktionieren dann sofort (auch auf dem Pi in Hook-Scripts); für Events zusätzlich pip install websockets (ab Version 13).

Einen API-Schlüssel legst du auf der Seite API & Hooks → Schlüssel an.

Kurz

from laptimer_client import Laptimer, LaptimerError

lt = Laptimer("http://pi:5800", key="lt_…")
lt.assign_channels(["NITRO", "ACE", "BOB"], create=True)   # Kanal 0, 1, 2
lt.start_session(race=True, laps=3, countdown_s=5)
…
lt.confirm()                                                # Ergebnis bestätigen

• Jede Methode gibt das JSON des Servers als dict/list zurück (Felder wie in Datenformate).
• Argumente mit None werden nicht mitgeschickt, d. h. „bleibt, wie es ist“ bzw. „Standard“. Ein ausdrückliches null (z. B. „kein Pilot auf diesem Kanal“) ist NOBODY: lt.set_channel(2, NOBODY).
• Fehler werfen LaptimerError: e.message ist die (deutsche) Meldung des Servers zum direkten Anzeigen, e.status der HTTP-Status (400 falsche Argumente, 401 Schlüssel falsch, 404 unbekannter Befehl; über den WebSocket dieselben Zahlen; None, wenn der Server nicht erreichbar ist).
• Laptimer() ohne Angaben nimmt LAPTIMER_API_URL (Standard http://localhost:5800) und LAPTIMER_API_KEY aus der Umgebung. Die Adresse darf auch pi:5800 oder …/api/v1 sein.

Beispiele

Rennen mit 3 Runden starten, jede Runde ausgeben, Ergebnis bestätigen

Mit Handler-Funktionen, ohne asyncio (lt.on(...) und lt.run_forever()):

from laptimer_client import Laptimer

lt = Laptimer("http://pi:5800", key="lt_…")

@lt.on("connected")            # nach dem Verbinden: erst dann starten, damit nichts fehlt
def start(ev):
    if ev.data["session"] is None:
        lt.start_session(race=True, laps=3)

@lt.on("pilot_pass")
def lap(ev):
    d = ev.data
    if d["kind"] == "lap":
        print(f'{d["pilot"]}: Runde {d["lap"]} in {d["lap_time_s"]:.2f} s, Platz {d["position"]}')

@lt.on("session_stopped")
def stopped(ev):
    print("Sieger:", ev.data["winner"])
    if ev.data["needs_confirmation"]:
        lt.confirm(ev.data["session_id"])
    lt.stop()                  # run_forever() beenden

lt.run_forever()               # bis lt.stop() oder Strg+C; verbindet sich neu, wenn nötig

"*" als Event-Name bekommt jedes Event. Handler dürfen auch async def sein. Eine Ausnahme im Handler wird ausgegeben, die Schleife läuft weiter.

Dasselbe mit asyncio

import asyncio
from laptimer_client import Laptimer

lt = Laptimer("http://pi:5800", key="lt_…")

async def main():
    async for ev in lt.events("pilot_pass", "session_stopped"):
        if ev.event == "pilot_pass" and ev.data["kind"] == "lap":
            print(ev.data["pilot"], ev.data["lap"], ev.data["lap_time_s"])
        elif ev.event == "session_stopped":
            break

asyncio.run(main())

lt.events(*namen) liefert die Events (ohne Namen: alle) und verbindet sich nach einem Abbruch selbst neu (reconnect=False beendet die Schleife stattdessen).

Befehle über dieselbe WebSocket-Verbindung (ohne Neuverbinden):

async with lt.connect() as conn:
    print(conn.initial_state["session"])         # Zustand beim Verbinden (aus „hello“)
    await conn.start_session(race=True, laps=3)  # alle Methoden wie oben, mit await
    ev = await conn.wait_for("session_stopped", timeout=600)
    await conn.confirm()
    async for ev in conn:                        # alle weiteren Events
        ...

Ein Event ist ein Event

Zwei Events kommen vom Client selbst (seq = 0): connected nach jedem (Neu-)Verbinden, data = der Zustand (wie state()), und lagged, wenn der Client zu langsam war (data["missed"] Events fehlen; Zustand neu holen).

Hook-Script

Hooks (Seite API & Hooks → Hooks) starten bei einem Event ein Script; das Event steht in LAPTIMER_JSON. Event.from_env() liest es:

#!/usr/bin/env python3
from pathlib import Path
from laptimer_client import Event, Laptimer

ev = Event.from_env()
if ev.event == "session_stopped" and ev.data["race"] and ev.data["reason"] == "finished":
    lt = Laptimer("http://localhost:5800", key=Path("api_key.txt").read_text().strip())
    lt.confirm(ev.data["session_id"])     # alle im Ziel: Ergebnis von selbst bestätigen

Ins Script-Ordner (Standard server/data/hooks/) gehören dann das Script (chmod +x), laptimer_client.py und api_key.txt. Hooks laufen im Script-Ordner.

Fertige Beispiele in examples/:

• race.py: Piloten setzen, Rennen starten, Countdown und Runden ausgeben, am Ende bestätigen oder verwerfen (python examples/race.py --url http://pi:5800 --key lt_… --laps 3 NITRO ACE).
• hook_autoconfirm.py: Hook, der fertige Rennen bestätigt und den Sieger in eine Datei schreibt.

Methoden

Alle gibt es auf Laptimer (HTTP, blockierend) und auf Connection (WebSocket, mit await). Kanäle sind Indizes ab 0, Zeiten µs seit 1970.

Nur auf Laptimer: events(*namen, reconnect=True, retry_s=2.0), connect(), on(event, handler), run_forever(), stop(), ws_url(). Nur auf Connection: hello, initial_state (Zustand beim Verbinden; await conn.state() holt ihn neu), commands, next_event(timeout=None), wait_for(*namen, timeout=None), close(); async for ev in conn endet, wenn die Verbindung zu ist.

Tests

Gegen einen laufenden Server, nicht den, mit dem ihr gerade fliegt (die Tests starten und löschen Sessions und legen Piloten an). Ohne LAPTIMER_API_KEY legen sie selbst einen Schlüssel an.

cd clients/python
python3 -m venv /tmp/lt-venv && /tmp/lt-venv/bin/pip install websockets
LAPTIMER_API_URL=http://localhost:5802 /tmp/lt-venv/bin/python -m unittest discover -s tests -v

Ohne websockets laufen nur die HTTP-Tests, die Event-Tests werden übersprungen.

Veröffentlichen (PyPI)

Die Version kommt aus dem Git-Tag und ist unabhängig von der Server-Version. Im Repo steht in laptimer_client/__init__.py nur der Platzhalter __version__ = "0.0.0.dev0"; der Workflow schreibt vor dem Bauen die echte Version hinein, pyproject.toml liest sie von dort. Ablauf:

1. Tag py-v<version> pushen, z. B. git tag py-v0.2.0 && git push origin py-v0.2.0 (Version nach PEP 440 in Normalform, also 0.2.0rc1, nicht 0.2.0-rc1).
2. Der Workflow .github/workflows/pypi.yml setzt die Version, baut sdist und Wheel, prüft sie (twine check, Import aus dem Wheel) und lädt sie per Trusted Publishing (ohne Token) auf PyPI.

Von Hand starten (Actions → PyPI → Run workflow) lädt auf TestPyPI statt PyPI, zum Ausprobieren, als <letzter py-v-Tag>.dev<Laufnummer> (z. B. 0.2.0.dev14), damit jeder Lauf eine neue Version hat: pip install --pre -i https://test.pypi.org/simple/ fpv-laptimer-client.

Einmal einrichten: auf pypi.org bzw. test.pypi.org unter Publishing → Add a new pending publisher das Projekt fpv-laptimer-client, Owner bli-blu-bla, Repo fpv-laptimer, Workflow pypi.yml, Environment pypi bzw. testpypi eintragen; im GitHub-Repo unter Settings → Environments die beiden Environments anlegen (bei pypi am besten mit Freigabe durch dich).

Lokal bauen und prüfen:

cd clients/python
python3 -m venv /tmp/lt-build && /tmp/lt-build/bin/pip install build twine
/tmp/lt-build/bin/python -m build                 # → dist/*.tar.gz, dist/*.whl
/tmp/lt-build/bin/twine check --strict dist/*

LICENSE ist ein Symlink auf die Lizenz im Repo; im Paket landet sie als Datei.