matterdecode/gen-lifx-products.py

#!/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()