diff --git a/tools/station/display/render.py b/tools/station/display/render.py index ae678a8..fb7dc83 100644 --- a/tools/station/display/render.py +++ b/tools/station/display/render.py @@ -167,6 +167,25 @@ PHASE_LABELS = {"auto": "AUTO", "listen": "LISTEN"} ARMED_LABEL = "FREIGEGEBEN" ARMED_DETAIL_PREFIX = "Freigabe: " +# Woraus die Station die Signiervariante abgeleitet hat (products.classify_variant): +# "-dev" kann in der Produkt-ID stehen, in der Version, oder in beiden. Die Quelle steht +# mit auf dem Panel, weil genau hier am 2026-09-15 ein falsch signiertes Image +# ausgeliefert wurde: wer vor der Freigabe sieht, dass "dev" nur aus der Version kommt +# (oder dass "prod" bedeutet: nirgends ein "-dev"), kann eine Fehlklassifikation abfangen, +# bevor die Lampe das Image ablehnt. Kurzformen, weil die Beiwerkzeile schmal ist. +VARIANT_SOURCE_LABELS = {"both": "ID+Ver", "product_id": "ID", "version": "Ver"} + + +def _variant_label(last_seen: dict) -> str: + """"dev/Ver", "dev/ID", "dev/ID+Ver", "prod" - die Quelle nur, wo sie etwas aussagt. + Bei "prod" gibt es keine ("none"), und eine aeltere Serverfassung ohne das Feld + liefert wie bisher nur die Variante.""" + variant = str(last_seen.get("variant") or "?") + label = VARIANT_SOURCE_LABELS.get(last_seen.get("variant_source")) + if not label: + return variant + return variant + "/" + label + def _hhmm(ts: object) -> str: """"2026-09-13T13:04:02Z" -> "13:04". Alles andere (fehlt, falsch geformt, aus einer @@ -275,7 +294,8 @@ def blocks(state: dict) -> list[tuple[str, str]]: target = last_seen.get("target") variant = last_seen.get("variant") if target or variant: - out.append((f"{target or '?'} ({variant or '?'})", ROLE_DETAIL)) + out.append(("{} ({})".format(target or "?", _variant_label(last_seen)), + ROLE_DETAIL)) lamp_version = last_seen.get("lamp_version") if lamp_version: out.append((f"FW: {lamp_version}", ROLE_DETAIL)) diff --git a/tools/station/products.py b/tools/station/products.py index 1d0d8c3..110d516 100644 --- a/tools/station/products.py +++ b/tools/station/products.py @@ -1,17 +1,37 @@ """Produkt-IDs der alten Klyqa-Firmware und ihre Zuordnung zu ninlight-Zielen. Vollstaendig uebernommen aus qcl_product_info.c der alten Firmware -(~/workspace/qcx-w600-firmware). Der "-dev"-Anhang entscheidet zugleich ueber den -Signierschluessel: Lampen mit Dev-Produkt-ID akzeptieren nur den Entwicklungs-, -Produktionslampen nur den Produktionsschluessel. +(~/workspace/qcx-w600-firmware). + +Die Variante (dev/prod) entscheidet ueber den Signierschluessel: Lampen mit +Entwicklungs-Bootloader akzeptieren nur den Entwicklungs-, Produktionslampen nur den +Produktionsschluessel. Sie ist aber KEINE Eigenschaft der Produkt-ID, sondern eine +Eigenschaft der ANFRAGE - das ist der Befund vom 2026-09-15 an echter Hardware: + + /qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img + +Dieselbe Entwicklungs-G95, aber das "-dev" steht nur in der VERSION, nicht in der +Produkt-ID. Die Station hielt sie damit fuer eine Produktionslampe, lieferte das +produktionssignierte Image aus, und die Lampe lehnte es ab. Die Machbarkeitsprobe im +September hatte eine andere Entwicklungs-G95 mit "-dev" an BEIDEN Stellen gesehen, eine +Produktionslampe an keiner - alle drei Formen kommen also vor. + +Regel seither: "-dev" an einer der beiden Stellen (Produkt-ID oder Version) bedeutet +Entwicklungsgeraet. PRODUCTS bildet nur noch auf Geraetetyp und Anzeigename ab, beide +Schreibweisen der Produkt-ID fuehren auf denselben Eintrag; die Variante ermittelt +parse_request() aus beiden Teilen und meldet in .variant_source mit, WORAUS sie stammt +("both", "product_id", "version", "none"), damit eine Fehlklassifikation vor der Freigabe +am Panel und in status.json sichtbar ist statt erst an der Lampe. """ import re from collections import namedtuple -Product = namedtuple("Product", "product_id target variant name") -ParsedRequest = namedtuple("ParsedRequest", "product_id lamp_version product") +Product = namedtuple("Product", "product_id target name") +ParsedRequest = namedtuple( + "ParsedRequest", "product_id lamp_version product variant variant_source") -# (Produkt-ID ohne "@", Ziel, Anzeigename) - die Dev-Variante wird darunter erzeugt. +# (Produkt-ID ohne "@", Ziel, Anzeigename) - die "-dev"-Schreibweise wird darunter +# als zweiter Schluessel auf denselben Eintrag erzeugt. _BASE = [ ("klyqa.lighting.rgb-cw-ww.e27", "e27_rgb", "Klyqa E27 RGB CW/WW"), ("klyqa.lighting.rgb-cw-ww.e14", "e14_rgb", "Klyqa E14 RGB CW/WW"), @@ -25,6 +45,9 @@ _BASE = [ _UNSAFE = re.compile(r"[^0-9A-Za-z+\-_.]") # Die Produkt-ID darf kein "_" enthalten - daran trennen wir Version und ID. _PATH = re.compile(r"^/qcx_firmware_([^/_]+)_([^/_]+)_ota_light\.img$") +# "-dev" in der Version, aber nicht als Anfang eines laengeren Wortes ("-devel", "-dev2"): +# erlaubt sind danach nur Ende oder ein Trennzeichen wie "." (z.B. "V1.1-dev.3"). +_DEV_IN_VERSION = re.compile(r"-dev(?![0-9A-Za-z])") def websafe(product_id): @@ -36,30 +59,54 @@ def websafe(product_id): def _build(): t = {} for pid, target, name in _BASE: - for suffix, variant in (("", "prod"), ("-dev", "dev")): - key = websafe("@" + pid + suffix) - t[key] = Product(key, target, variant, name) + base_key = websafe("@" + pid) + product = Product(base_key, target, name) + # Beide Schreibweisen zeigen auf DENSELBEN Eintrag - "-dev" an der Produkt-ID + # sagt seit dem Hardwarebefund nichts mehr ueber das Ziel, nur noch (mit) etwas + # ueber die Variante, und die ermittelt parse_request(). + t[base_key] = product + t[websafe("@" + pid + "-dev")] = product # Das EVK hat nur eine Produktions-ID. Im Dev-Build meldet es sich laut # qcl_product_info.c als "@klyqa.lighting.rgb-cw-ww.e27-dev" - die EVK-Zeile ist dort # auskommentiert und durch die E27 ersetzt. Ein dev-signiertes EVK ist damit von einer # dev-signierten E27 RGB NICHT unterscheidbar und bekommt deren Image. Das EVK wird # ohnehin seriell geflasht; die Zeile steht nur der Vollstaendigkeit halber hier. + # Meldet es eine "-dev"-Version, gilt auch hier die Regel: dann wird evk_dev.img + # gesucht (und mangels Datei sauber mit 404 abgelehnt), nicht das falsch signierte. key = websafe("@klyqa.lighting.w600.evk1.2") - t[key] = Product(key, "evk", "prod", "W600 EVK v1.2") + t[key] = Product(key, "evk", "W600 EVK v1.2") return t PRODUCTS = _build() +def classify_variant(product_id, lamp_version): + """(variant, variant_source) aus Produkt-ID UND Version. + + "-dev" an einer der beiden Stellen bedeutet Entwicklungsgeraet; variant_source sagt, + an welcher es stand: "both", "product_id", "version" oder - bei prod - "none".""" + in_id = product_id.endswith("-dev") + in_version = bool(_DEV_IN_VERSION.search(lamp_version)) + if in_id and in_version: + return "dev", "both" + if in_id: + return "dev", "product_id" + if in_version: + return "dev", "version" + return "prod", "none" + + def parse_request(path): """Zerlegt einen angefragten Pfad. Gibt None zurueck, wenn er nicht dem Muster der alten Firmware entspricht, sonst ParsedRequest - dessen .product ist None, wenn die - Produkt-ID unbekannt ist. Unbekannt heisst 404, nicht raten.""" + Produkt-ID unbekannt ist. Unbekannt heisst 404, nicht raten. .variant/.variant_source + stehen auch bei unbekannter Produkt-ID, sie haengen nicht an der Tabelle.""" m = _PATH.match(path) if not m: return None pid, version = m.group(1), m.group(2) if pid != websafe(pid) or "/" in pid or ".." in pid.split("."): return None - return ParsedRequest(pid, version, PRODUCTS.get(pid)) + variant, source = classify_variant(pid, version) + return ParsedRequest(pid, version, PRODUCTS.get(pid), variant, source) diff --git a/tools/station/server.py b/tools/station/server.py index f9658fb..11d9cb1 100755 --- a/tools/station/server.py +++ b/tools/station/server.py @@ -259,7 +259,8 @@ class Station: def record_migration(self, ip: str, target: str | None, variant: str | None, product_id: str, lamp_version: str, product_name: str | None, - version: str, nbytes: int, status: str) -> None: + version: str, nbytes: int, status: str, + variant_source: str | None = None) -> None: """Protokolliert eine Migrationsanfrage. Anders als record() oben wird hier auch dann ein Eintrag angelegt (Status "listen"/"unknown"/"notfound"), wenn nichts ausgeliefert wurde - das ist der ganze Zweck der Horchphase. last_seen wird nur bei @@ -269,13 +270,15 @@ class Station: if status == "ok": self.served += 1 entry = {"ts": now_iso(), "ip": ip, "target": target, "variant": variant, + "variant_source": variant_source, "product_id": product_id, "lamp_version": lamp_version, "version": version, "bytes": nbytes, "status": status} self.recent.append(entry) del self.recent[:-RECENT_MAX] if target is not None: self.last_seen = {"product_id": product_id, "target": target, - "variant": variant, "product_name": product_name, + "variant": variant, "variant_source": variant_source, + "product_name": product_name, "lamp_version": lamp_version, "ip": ip, "ts": entry["ts"]} self.write_state() @@ -395,7 +398,7 @@ class Handler(BaseHTTPRequestHandler): self.reject(404, "not found") self.log(404, 0, started) return - pid, lamp_version, product = parsed + pid, lamp_version, product, variant, variant_source = parsed ip = self.client_address[0] if product is None: self.station.record_migration(ip, None, None, pid, lamp_version, None, "?", 0, @@ -404,7 +407,11 @@ class Handler(BaseHTTPRequestHandler): self.log(404, 0, started) return - target, variant, name = product.target, product.variant, product.name + # variant/variant_source kommen aus parse_request(), nicht mehr aus der + # Produkttabelle: das "-dev" kann in der Produkt-ID stehen, in der Version, oder in + # beiden (Hardwarebefund 2026-09-15, siehe products.py). Das entscheidet, welches + # der beiden unterschiedlich signierten Images ausgeliefert wird. + target, name = product.target, product.name version = self.station.image_version(target, variant) # claim_delivery() liest die Phase UND verbraucht eine armed-Freigabe atomar unter # station.lock (Fixrunde 1, C1) - zwischen dem Lesen und dem Verbrauchen darf keine @@ -415,7 +422,7 @@ class Handler(BaseHTTPRequestHandler): # last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase: # die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht. self.station.record_migration(ip, target, variant, pid, lamp_version, name, - version, 0, "listen") + version, 0, "listen", variant_source) self.reject(404, "listening only") self.log(404, 0, started) return @@ -427,7 +434,7 @@ class Handler(BaseHTTPRequestHandler): # dass es gar kein Image gibt - das darf nicht auf Kosten der Freigabe gehen. self.station.restore_armed(target) self.station.record_migration(ip, target, variant, pid, lamp_version, name, - version, 0, "notfound") + version, 0, "notfound", variant_source) self.reject(404, "no image for this device") self.log(404, 0, started) return @@ -440,7 +447,7 @@ class Handler(BaseHTTPRequestHandler): # (das ist der Punkt von C1) - bei einem Abbruch wird das hier zurueckgenommen. self.station.restore_armed(target) self.station.record_migration(ip, target, variant, pid, lamp_version, name, version, - sent, status) + sent, status, variant_source) def _serve_file(self, path: Path, started: float) -> tuple[int, str]: """Sendet `path` mit Range-Unterstuetzung; liefert (gesendete Bytes, Status). diff --git a/tools/station/tests/test_products.py b/tools/station/tests/test_products.py index 6a41ea3..c6c7a28 100644 --- a/tools/station/tests/test_products.py +++ b/tools/station/tests/test_products.py @@ -7,43 +7,98 @@ import products # noqa: E402 def test_table_is_complete(): - """Acht Geraete mal zwei Varianten, minus das EVK, das keine eigene Dev-ID hat.""" + """Sieben Geraete in beiden Schreibweisen der Produkt-ID, plus das EVK, das nur eine + Produktions-ID hat.""" assert len(products.PRODUCTS) == 15 targets = {p.target for p in products.PRODUCTS.values()} assert targets == {"evk", "e27_rgb", "e14_rgb", "gu10_rgb", "e27_cw", "e14_cw", "gu10_cw", "g95_cw"} -def test_production_id_maps_to_prod_variant(): +def test_table_carries_no_variant(): + """Die Variante ist eine Eigenschaft der Anfrage, nicht der Produkt-ID - sie darf gar + nicht mehr aus der Tabelle zu holen sein (Hardwarebefund 2026-09-15).""" + assert not hasattr(products.PRODUCTS["-klyqa.lighting.cw-ww.g95"], "variant") + assert "variant" not in products.Product._fields + + +def test_both_spellings_map_to_the_same_device(): + with_dev = products.PRODUCTS["-klyqa.lighting.cw-ww.g95-dev"] + without = products.PRODUCTS["-klyqa.lighting.cw-ww.g95"] + assert with_dev.target == without.target == "g95_cw" + assert with_dev.name == without.name == "Klyqa G95 Vintage CW/WW" + + +def test_production_id_maps_to_device_and_name(): p = products.PRODUCTS["-klyqa.lighting.rgb-cw-ww.gu10"] - assert (p.target, p.variant) == ("gu10_rgb", "prod") + assert p.target == "gu10_rgb" assert p.name == "Klyqa GU10 RGB CW/WW" -def test_dev_suffix_selects_dev_variant(): - p = products.PRODUCTS["-klyqa.lighting.cw-ww.g95-dev"] - assert (p.target, p.variant) == ("g95_cw", "dev") - - -def test_parses_real_request_from_g95(): - """Am 2026-09-13 an einer echten G95 protokolliert.""" +def test_parses_real_request_from_g95_dev_in_both_places(): + """Am 2026-09-13 an einer echten Entwicklungs-G95 protokolliert: "-dev" in Produkt-ID + UND Version.""" r = products.parse_request( "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img") assert r is not None assert r.product_id == "-klyqa.lighting.cw-ww.g95-dev" assert r.lamp_version == "V1.1.1.1800-dev" assert r.product.target == "g95_cw" - assert r.product.variant == "dev" + assert (r.variant, r.variant_source) == ("dev", "both") -def test_parses_real_request_from_gu10(): +def test_parses_real_request_from_g95_dev_only_in_version(): + """Am 2026-09-15 an einer echten Entwicklungs-G95 protokolliert - die Anfrage, an der + die Station bisher scheiterte: Produkt-ID OHNE "-dev", Version MIT. Sie muss trotzdem + als Entwicklungsgeraet gelten, sonst geht das produktionssignierte Image raus.""" + r = products.parse_request( + "/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img") + assert r is not None + assert r.product_id == "-klyqa.lighting.cw-ww.g95" + assert r.lamp_version == "V1.1.1.2357-dev" + assert r.product.target == "g95_cw" + assert (r.variant, r.variant_source) == ("dev", "version") + + +def test_parses_real_request_from_gu10_without_any_dev(): + """Die erfolgreich migrierte Produktionslampe: nirgends ein "-dev".""" r = products.parse_request( "/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img") assert r.product.target == "gu10_rgb" - assert r.product.variant == "prod" + assert (r.variant, r.variant_source) == ("prod", "none") assert r.lamp_version == "V1.1.0.1138" +def test_dev_only_in_product_id_is_dev(): + """Die vierte denkbare Kombination - so noch nicht am Geraet gesehen, aber die + Machbarkeitsprobe zeigt, dass die Produkt-ID das "-dev" tragen kann, und nichts + garantiert, dass die Version es dann auch tut.""" + r = products.parse_request( + "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800_ota_light.img") + assert r.product.target == "g95_cw" + assert (r.variant, r.variant_source) == ("dev", "product_id") + + +def test_dev_must_be_a_whole_word_in_the_version(): + """"-devel"/"-dev2" sind kein Entwicklungskennzeichen - sonst wuerde eine + Produktionslampe mit so einer Version das dev-signierte Image bekommen.""" + for version in ("V1.1.1.2357-devel", "V1.1.1.2357-dev2"): + r = products.parse_request( + "/qcx_firmware_-klyqa.lighting.cw-ww.g95_" + version + "_ota_light.img") + assert (r.variant, r.variant_source) == ("prod", "none"), version + # ... ein "-dev" vor einem Trennzeichen dagegen schon. + r = products.parse_request( + "/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1-dev.3_ota_light.img") + assert (r.variant, r.variant_source) == ("dev", "version") + + +def test_variant_is_classified_even_for_unknown_product_ids(): + r = products.parse_request( + "/qcx_firmware_-acme.lightbulb.xxl_V9.9-dev_ota_light.img") + assert r.product is None + assert (r.variant, r.variant_source) == ("dev", "version") + + def test_unknown_product_id_parses_but_has_no_product(): r = products.parse_request( "/qcx_firmware_-acme.lightbulb.xxl_V9.9_ota_light.img") diff --git a/tools/station/tests/test_render.py b/tools/station/tests/test_render.py index 6f83f15..cef9132 100644 --- a/tools/station/tests/test_render.py +++ b/tools/station/tests/test_render.py @@ -33,6 +33,26 @@ def test_migration_shows_detected_lamp(): assert "13:04" in text +def test_migration_shows_where_the_dev_classification_came_from(): + """Der Hardwarebefund vom 2026-09-15: die Station lieferte ein falsch signiertes Image + aus, ohne dass am Panel etwas darauf hindeutete. Steht die Quelle der Einstufung + daneben, sieht der Bediener VOR der Freigabe, worauf sie beruht.""" + def variant_line(last_seen): + state = {"mode": "migration", "phase": "listen", "served": 0, "recent": [], + "ssid": "qconnex-firmware", "last_seen": last_seen} + return [l for l in render.lines(state) if l.startswith("g95_cw (")][0] + + base = {"product_name": "Klyqa G95 Vintage CW/WW", "target": "g95_cw", + "lamp_version": "V1.1.1.2357-dev"} + assert variant_line(dict(base, variant="dev", variant_source="version")) == "g95_cw (dev/Ver)" + assert variant_line(dict(base, variant="dev", variant_source="product_id")) == "g95_cw (dev/ID)" + assert variant_line(dict(base, variant="dev", variant_source="both")) == "g95_cw (dev/ID+Ver)" + # prod hat keine Quelle - da gibt es nirgends ein "-dev", das man anzeigen koennte. + assert variant_line(dict(base, variant="prod", variant_source="none")) == "g95_cw (prod)" + # Eine aeltere Serverfassung ohne das Feld bleibt darstellbar. + assert variant_line(dict(base, variant="dev")) == "g95_cw (dev)" + + def test_recovery_mode_without_traffic(): lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery", "served": 0, "recent": []}) diff --git a/tools/station/tests/test_server.py b/tools/station/tests/test_server.py index 18aa001..89c53b6 100644 --- a/tools/station/tests/test_server.py +++ b/tools/station/tests/test_server.py @@ -594,6 +594,40 @@ def test_migration_auto_serves_known_product(migration_server): assert st["last_seen"]["variant"] == "dev" +def test_migration_dev_only_in_version_serves_the_dev_image(migration_server): + """Der Hardwarebefund vom 2026-09-15: Produkt-ID OHNE "-dev", Version MIT. Frueher + ging hier das produktionssignierte Image raus (200, und die Lampe lehnte es ab) - + jetzt muss das dev-signierte gehen, und status.json muss die Quelle mitfuehren.""" + srv, imgdir = migration_server + (imgdir / "migrate").mkdir(exist_ok=True) + (imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"DEV" * 1000) + (imgdir / "migrate" / "g95_cw_prod.img").write_bytes(b"PROD" * 1000) + r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img") + assert r.status == 200 + assert r.body == b"DEV" * 1000 + st = srv.status(lambda s: s.get("last_seen") is not None) + assert st["last_seen"]["target"] == "g95_cw" + assert st["last_seen"]["variant"] == "dev" + assert st["last_seen"]["variant_source"] == "version" + assert st["recent"][-1]["variant"] == "dev" + assert st["recent"][-1]["variant_source"] == "version" + + +def test_migration_production_lamp_still_gets_the_prod_image(migration_server): + """Die erfolgreich migrierte GU10: nirgends ein "-dev", also prod - und die Quelle + "none", damit am Panel/in status.json steht, worauf die Einstufung beruht.""" + srv, imgdir = migration_server + (imgdir / "migrate").mkdir(exist_ok=True) + (imgdir / "migrate" / "gu10_rgb_dev.img").write_bytes(b"DEV" * 1000) + (imgdir / "migrate" / "gu10_rgb_prod.img").write_bytes(b"PROD" * 1000) + r = srv.get("/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img") + assert r.status == 200 + assert r.body == b"PROD" * 1000 + st = srv.status(lambda s: s.get("last_seen") is not None) + assert st["last_seen"]["variant"] == "prod" + assert st["last_seen"]["variant_source"] == "none" + + def test_migration_listen_records_but_serves_nothing(migration_server): srv, imgdir = migration_server (imgdir / "migrate").mkdir(exist_ok=True)