Updated matter decoder to show more info, and added a homekit decoder.

gen-lifx-products.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+# Regenerate LIFX's rows in products.tsv from LIFX's upstream product catalog.
+#
+# products.tsv has three columns, "<vendor id>\t<product id>\t<name>", sorted by
+# (vendor id, product id). It is consumed by matter-decode.sh's get_product_name():
+# when a decoded Matter QR's vendor ID has an equivalent in
+# vendor-equivalents.tsv, the product ID is looked up here (under any equivalent
+# id) to print a human name alongside it. LIFX's Matter vendor id is 4870; the
+# rows below are LIFX's, keyed under 4870.
+#
+# The table is derived from upstream LIFX's products.json -- refresh from there
+# rather than hand-editing.
+#
+# https://raw.githubusercontent.com/LIFX/products/refs/heads/master/products.json
+#
+# By default the script pulls the upstream file over the network (via curl,
+# falling back to urllib) and MERGES LIFX's rows into products.tsv, preserving
+# any other vendors' rows already there:
+#
+#     python3 gen-lifx-products.py
+#
+# A positional argument overrides the source with another URL or a local .json
+# file; --output (default: products.tsv) overrides the target. Merging is
+# idempotent: re-running replaces LIFX's (vendor 4870) rows in place and leaves
+# every other vendor's rows untouched. Requires only the Python 3 standard
+# library.
+#
+# NOTE: the upstream file carries only "pid" per product -- there is no separate
+# Matter/CHIP product-id field, so the lookup assumes LIFX's Matter product ID
+# equals their "pid". If that ever changes, adjust the mapping here.
+
+import argparse
+import json
+import shutil
+import subprocess
+import sys
+import urllib.request
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+
+# LIFX's Matter/CHIP manufacturer id.
+LIFX_VID = 4870
+
+# Pinned upstream source -- LIFX's product catalog.
+UPSTREAM = ("https://raw.githubusercontent.com/LIFX/products/"
+             "refs/heads/master/products.json")
+
+
+# fetch a URL to text via curl (preferred) or urllib (fallback).
+def fetch_url(url):
+    if shutil.which("curl"):
+        try:
+            out = subprocess.run(["curl", "-fsSL", url],
+                                 capture_output=True, check=True).stdout
+            return out.decode("utf-8")
+        except subprocess.CalledProcessError as e:
+            sys.stderr.write(e.stderr.decode("utf-8", "replace"))
+            sys.exit(f"error: failed to fetch {url}")
+    req = urllib.request.Request(url, headers={"User-Agent": "gen-lifx-products.py"})
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return resp.read().decode("utf-8")
+    except Exception as e:
+        sys.exit(f"error: failed to fetch {url}: {e}")
+
+
+# return JSON text for a URL or a local .json path.
+def read_source(source):
+    if source.startswith(("http://", "https://")):
+        return fetch_url(source)
+    path = Path(source)
+    if not path.exists():
+        sys.exit(f"error: no such file: {path}")
+    return path.read_text(encoding="utf-8")
+
+
+# return [(product_id, name), ...] from LIFX's products.json. The file is a list
+# of vendor entries, each with a "products" list of {"pid", "name"}; flatten all
+# and warn on a duplicate pid (later entry wins).
+def rows_from_json(json_text):
+    data = json.loads(json_text)
+    rows = []
+    seen = {}
+    for vendor in data:
+        for prod in vendor.get("products", []):
+            pid = prod.get("pid")
+            name = prod.get("name", "")
+            if pid is None:
+                continue
+            if pid in seen:
+                sys.stderr.write(
+                    f"warning: duplicate pid {pid} "
+                    f"({seen[pid]!r} vs {name!r}); later entry wins\n")
+            seen[pid] = name
+            rows.append((int(pid), name))
+    return rows
+
+
+# keep the (vendor, product, name) rows in `path` whose vendor is not `vendor`,
+# skipping blank and '#' comment lines. Used so a re-run replaces only this
+# vendor's rows and leaves the others' intact.
+def load_other_rows(path, vendor):
+    kept = []
+    p = Path(path)
+    if not p.exists():
+        return kept
+    for line in p.read_text(encoding="utf-8").splitlines():
+        if not line.strip() or line.lstrip().startswith("#"):
+            continue
+        cols = line.split("\t")
+        if len(cols) < 3 or cols[0].strip() == "":
+            sys.stderr.write(f"warning: skipping malformed row: {line!r}\n")
+            continue
+        try:
+            v = int(cols[0])
+            pid = int(cols[1])
+        except ValueError:
+            sys.stderr.write(f"warning: skipping malformed row: {line!r}\n")
+            continue
+        if v != vendor:
+            kept.append((v, pid, cols[2]))
+    return kept
+
+
+# merge this vendor's freshly parsed rows into the shared products.tsv: drop the
+# vendor's existing rows, add the new ones, re-sort by (vendor, product), write
+# back. Returns the number of rows written for this vendor.
+def merge_products(path, new_rows, vendor):
+    kept = load_other_rows(path, vendor)
+    combined = kept + [(vendor, pid, name) for (pid, name) in new_rows]
+    combined.sort(key=lambda r: (r[0], r[1]))
+    with open(path, "w", encoding="utf-8", newline="") as f:
+        for v, pid, name in combined:
+            f.write(f"{v}\t{pid}\t{name}\n")
+    return len(new_rows)
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description="Regenerate LIFX's rows (vendor 4870) in products.tsv from "
+                     "LIFX's upstream products.json, merged with other vendors "
+                     "(fetches upstream by default).")
+    ap.add_argument(
+         "source", nargs="?", default=UPSTREAM,
+        help="a URL or local .json file (default: LIFX's upstream products.json)")
+    ap.add_argument(
+         "-o", "--output", default=str(SCRIPT_DIR / "products.tsv"),
+        help="output path (default: products.tsv next to this script); "
+              "LIFX's rows are merged in, other vendors preserved")
+    args = ap.parse_args()
+
+    json_text = read_source(args.source)
+    rows = rows_from_json(json_text)
+    if not rows:
+        sys.exit("error: no products found in the source")
+
+    out = Path(args.output)
+    n = merge_products(out, rows, LIFX_VID)
+    sys.stderr.write(
+        f"merged {n} LIFX products into {out} (from {args.source})\n")
+
+
+if __name__ == "__main__":
+    main()