From 817822594ca3697f91d96dd5c68288a43f91ef9e Mon Sep 17 00:00:00 2001 From: grabowski Date: Thu, 9 Jul 2026 12:13:03 +0700 Subject: [PATCH] Add lamp_station.py: barcode-driven measurement workflow Interactive loop around lamp_export.py: scan an EAN (checksum validated, optional for barcode-less lamps), resolve it against previous lamps / a local cache / upcitemdb.com, confirm identification, record the packaging's advertised values, measure per the published procedure, and commit the bundle to the comparison-data repo with a device page stub for the web repo. Every lamp is keyed by an internal ID (L0001, L0002, ...) so one EAN can cover several versions (daylight/warm white variants) and unbranded lamps work. lamp_export.py passes optional ean, variant, and rated (advertised values) fields through into metrics.json. --- README.md | 30 ++++ lamp_export.py | 16 ++- lamp_station.py | 372 ++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 415 insertions(+), 3 deletions(-) create mode 100644 lamp_station.py diff --git a/README.md b/README.md index 74cea37..3de5655 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,36 @@ relative, which TM-30 is invariant to. Sanity check of the implementation: `uv run tm30.py` reproduces the published values for the CIE FL2 illuminant (Rf 70, Rg 86). +## Lamp Station (barcode workflow) + +Interactive loop around `lamp_export.py` for building the comparison database +end to end. Every lamp gets an internal ID (L0001, L0002, ...) as primary key; +the EAN barcode is optional metadata, so one barcode can cover daylight/warm +white variants (distinct IDs + variant field) and unbranded lamps work too. + +Per lamp: scan the box (or 'n' for no barcode), the EAN is checksum-validated +and resolved to manufacturer/model (previously measured lamps, local cache, +then upcitemdb.com; GS1-restricted 20-29 barcodes are flagged as +retailer-internal), lamps already measured under the same EAN are offered for +re-measurement, you confirm the identification and type the ADVERTISED values +from the packaging (flux, CCT, power, CRI, lifetime, equivalent W -> stored +under "rated" in metrics.json for claimed-vs-measured comparison), insert the +lamp, and the tool measures per the published procedure (230 V / 50 Hz, 60 s +settle, 5 readings averaged), writes lamps// into the comparison-data +checkout, commits, and pushes. If the web repo is present it also scaffolds +the device page stub with the rated specs pre-filled (review and commit that +one manually). + +```bash +uv run lamp_station.py # scan -> measure -> commit -> push +uv run lamp_station.py --no-push # commit locally only +uv run lamp_station.py --no-commit # just write the bundle +uv run lamp_station.py --settle 0 --readings 1 # quick smoke test +``` + +Defaults assume `C:/dev/buildfor_life_web` with the comparison-data submodule +initialized; override with --data-repo / --web-repo. + ## Offline Parsing Parse previously captured USB traffic (pcap files from USBPcap): diff --git a/lamp_export.py b/lamp_export.py index d0736f9..17a35d0 100644 --- a/lamp_export.py +++ b/lamp_export.py @@ -172,9 +172,19 @@ def write_bundle(reading, out_dir, meta, n_readings, supply=None): for b in tm30["bins"]: w.writerow({k: (f"{v:.6g}" if isinstance(v, float) else v) for k, v in b.items()}) - # metrics.json — grouped scalars plus identification. - metrics = { - "name": meta["name"], + # metrics.json — grouped scalars plus identification. ean/variant are set + # by lamp_station.py when one barcode covers several versions of a lamp + # (e.g. daylight and warm white sharing box art and EAN). + metrics = {"name": meta["name"]} + if meta.get("ean"): + metrics["ean"] = meta["ean"] + if meta.get("variant"): + metrics["variant"] = meta["variant"] + if meta.get("rated"): + # Advertised values from the packaging, keyed like their measured + # counterparts so consumers can pair them (truth-in-advertising rows). + metrics["rated"] = meta["rated"] + metrics |= { "manufacturer": meta["manufacturer"], "model": meta["model"], "notes": meta["notes"], diff --git a/lamp_station.py b/lamp_station.py new file mode 100644 index 0000000..96c8f86 --- /dev/null +++ b/lamp_station.py @@ -0,0 +1,372 @@ +""" +Lamp measurement station — scan a barcode, measure, commit, done. + +Interactive loop for building the buildfor.life lamp comparison database. +Every lamp gets an internal ID (L0001, L0002, ...) as its primary key; the +EAN barcode is optional metadata. This keeps lamps distinct when one barcode +covers several versions (daylight / warm white in the same box art) and +supports unbranded lamps without any barcode. + + 1. Scan the box with a keyboard-wedge barcode scanner (types EAN + Enter), + or type 'n' for a lamp without a barcode. + 2. The EAN is checksum-validated. Lamps already measured with the same + EAN are listed: pick one to re-measure, or continue as a new lamp + (that is how same-EAN variants get their own IDs). Identification is + suggested from previous lamps, a local cache, then upcitemdb.com + (best effort; GS1-restricted 20-29 barcodes skip the online lookup). + 3. Confirm or edit manufacturer / model / variant / notes, then type the + ADVERTISED values from the packaging (lumen, CCT, watts, CRI, + lifetime, equivalent watts; Enter skips unknown ones). They are + stored under "rated" in metrics.json so the site can show + advertised-vs-measured deltas. + 4. Insert the lamp, press Enter: the built-in supply powers it per the + published procedure (230 V / 50 Hz, 60 s settle, 5 readings averaged) + and the bundle (spd.csv, tm30.csv, metrics.json) is written to + /lamps//. + 5. The bundle is committed to the comparison-data repo and pushed. + If --web-repo points at buildfor_life_web, a device page stub is + scaffolded when missing (review, fill remaining specs, commit + manually). + +Usage: + uv run lamp_station.py # defaults below + uv run lamp_station.py --data-repo D:/comparison-data --no-push + uv run lamp_station.py --settle 0 --readings 1 # quick smoke test +""" + +import argparse +import json +import re +import subprocess +import sys +import urllib.request +from pathlib import Path + +from hpcs6500 import find_hpcs_port +from lamp_export import average_readings, readings_from_device, write_bundle + +DEFAULT_DATA_REPO = Path(r"C:\dev\buildfor_life_web\apps\web\comparison-data") +DEFAULT_WEB_REPO = Path(r"C:\dev\buildfor_life_web") +CACHE_FILE = Path.home() / ".cache" / "hpcs6500" / "ean-cache.json" +LOOKUP_URL = "https://api.upcitemdb.com/prod/trial/lookup?upc={}" +ID_RE = re.compile(r"^L(\d{4,})$") + +# Advertised values typed from the packaging; keys match the measured +# metrics they are compared against on the site. +RATED_FIELDS = [ + ("Phi_lm", "Rated flux (lm)"), + ("CCT_K", "Rated CCT (K)"), + ("Power_W", "Rated power (W)"), + ("Ra", "Rated CRI (Ra)"), + ("lifetime_h", "Rated lifetime (h)"), + ("W_equiv", "Incandescent equivalent (W)"), +] + + +def ean_normalize(code): + """Return a valid EAN-13 (UPC-A gets a leading zero), or None.""" + code = re.sub(r"\D", "", code) + if len(code) == 12: + code = "0" + code + if len(code) != 13: + return None + digits = [int(c) for c in code] + check = (10 - sum(d * (3 if i % 2 else 1) for i, d in enumerate(digits[:12])) % 10) % 10 + return code if check == digits[12] else None + + +def is_restricted(ean): + """GS1 prefixes 20-29 are retailer-internal, never publicly resolvable.""" + return 20 <= int(ean[:2]) <= 29 + + +def lamp_index(data_repo): + """{id: metrics-dict} for every measured lamp.""" + lamps = {} + for f in sorted((data_repo / "lamps").glob("L*/metrics.json")): + if ID_RE.match(f.parent.name): + try: + lamps[f.parent.name] = json.loads(f.read_text()) + except ValueError: + pass + return lamps + + +def next_lamp_id(data_repo): + used = [int(ID_RE.match(d.name).group(1)) + for d in (data_repo / "lamps").glob("L*") if ID_RE.match(d.name)] + return f"L{max(used, default=0) + 1:04d}" + + +def load_cache(): + try: + return json.loads(CACHE_FILE.read_text()) + except (OSError, ValueError): + return {} + + +def save_cache(cache): + CACHE_FILE.parent.mkdir(parents=True, exist_ok=True) + CACHE_FILE.write_text(json.dumps(cache, indent=2)) + + +def lookup_ean(ean, lamps): + """Best-effort (manufacturer, model, source) for an EAN.""" + for m in lamps.values(): + if m.get("ean") == ean and (m.get("manufacturer") or m.get("model")): + return m.get("manufacturer", ""), m.get("model", ""), f"previous lamp {m['name']}" + + cache = load_cache() + if ean in cache: + return cache[ean]["manufacturer"], cache[ean]["model"], "local cache" + + if is_restricted(ean): + return "", "", "retailer-internal barcode, not resolvable" + + try: + req = urllib.request.Request(LOOKUP_URL.format(ean), + headers={"User-Agent": "hpcs6500-lamp-station"}) + with urllib.request.urlopen(req, timeout=10) as resp: + data = json.loads(resp.read()) + item = (data.get("items") or [{}])[0] + brand, title = item.get("brand", ""), item.get("title", "") + if brand or title: + return brand, title, "upcitemdb.com (verify against packaging)" + except Exception as e: # noqa: BLE001 — lookup is strictly best-effort + print(f" (online lookup failed: {e})") + return "", "", "no match" + + +def prompt(label, default=""): + text = input(f"{label} [{default}]: " if default else f"{label}: ").strip() + return text or default + + +def prompt_rated(previous=None): + """Advertised values from the box; Enter skips a field.""" + previous = previous or {} + print(" Advertised values from the packaging (Enter to skip):") + rated = {} + for key, label in RATED_FIELDS: + default = previous.get(key, "") + raw = prompt(f" {label}", str(default) if default != "" else "") + if not raw: + continue + try: + value = float(raw) + rated[key] = int(value) if value == int(value) else value + except ValueError: + print(f" (not a number, skipped: {raw!r})") + return rated + + +def git(repo, *args, check=True): + return subprocess.run(["git", "-C", str(repo), *args], + check=check, capture_output=True, text=True) + + +def commit_bundle(data_repo, lamp_id, label, existed, push): + git(data_repo, "add", f"lamps/{lamp_id}") + if not git(data_repo, "status", "--porcelain", f"lamps/{lamp_id}").stdout.strip(): + print("No data changes to commit.") + return + verb = "Re-measure" if existed else "Add" + git(data_repo, "commit", "-m", f"{verb} lamp {lamp_id} ({label})") + print(f"Committed to comparison-data: {verb} lamp {lamp_id} ({label})") + if push: + git(data_repo, "push") + print("Pushed.") + + +def scaffold_device_page(web_repo, lamp_id, meta): + pages = web_repo / "apps" / "web" / "src" / "content" / "comparisons" / "lamps" + if not pages.is_dir(): + return + page = pages / f"{lamp_id}.md" + if page.exists(): + return + weights = [int(m.group(1)) for f in pages.glob("*.md") + for m in [re.search(r"^weight:\s*(\d+)", f.read_text(), re.M)] if m] + title = meta["model"] or lamp_id + if meta["variant"]: + title = f"{title} ({meta['variant']})" + rated = meta.get("rated", {}) + specs = [("ID", lamp_id)] + if meta["ean"]: + specs.append(("EAN", meta["ean"])) + if meta["variant"]: + specs.append(("Variant", meta["variant"])) + if "Power_W" in rated: + power = f"{rated['Power_W']:g} W" + if "W_equiv" in rated: + power += f" ({rated['W_equiv']:g} W equivalent)" + specs.append(("Rated power", power)) + if "Phi_lm" in rated: + specs.append(("Rated flux", f"{rated['Phi_lm']:g} lm")) + if "CCT_K" in rated: + specs.append(("CCT (rated)", f"{rated['CCT_K']:g} K")) + if "Ra" in rated: + specs.append(("CRI (rated)", f"{rated['Ra']:g}")) + if "lifetime_h" in rated: + specs.append(("Rated lifetime", f"{rated['lifetime_h']:,} h".replace(",", " "))) + + lines = ["---", f'title: "{title}"'] + if meta["manufacturer"]: + lines.append(f'manufacturer: "{meta["manufacturer"]}"') + lines += [ + 'description: "Household LED lamp, measured in our integrating sphere."', + f'csv: "/data/comparisons/lamps/{lamp_id}/spd.csv"', + "specs:", + *[f' - {{ label: "{k}", value: "{v}" }}' for k, v in specs], + f"weight: {max(weights, default=0) + 1}", + "---", + "", + "Measured as purchased, no burn-in beyond the procedure's stabilization period.", + "", + ] + page.write_text("\n".join(lines)) + print(f"Scaffolded device page: {page}") + print(" -> review it, bump the submodule, commit the web repo.") + + +def measure_one(ean, args, data_repo): + lamps = lamp_index(data_repo) + + # Same-EAN lamps already in the database: re-measure or new variant. + lamp_id, previous = None, {} + if ean: + same = {i: m for i, m in lamps.items() if m.get("ean") == ean} + if same: + print(" Lamps already measured with this EAN:") + for i, m in same.items(): + variant = f", {m['variant']}" if m.get("variant") else "" + print(f" {i}: {m.get('manufacturer', '')} {m.get('model', '')}{variant}") + choice = prompt("Re-measure one of these (type its ID) or Enter for a NEW lamp/variant").strip() + if choice: + if choice.upper() not in same: + print(f" Unknown ID {choice!r}, aborting this scan.") + return + lamp_id = choice.upper() + previous = same[lamp_id] + + if lamp_id is None: + lamp_id = next_lamp_id(data_repo) + print(f" New lamp: {lamp_id}" + (f" (EAN {ean})" if ean else " (no barcode)")) + else: + print(f" Re-measuring {lamp_id}") + + if previous: + manufacturer, model = previous.get("manufacturer", ""), previous.get("model", "") + elif ean: + manufacturer, model, source = lookup_ean(ean, lamps) + print(f" Lookup: {manufacturer or '?'} / {model or '?'} ({source})") + else: + manufacturer = model = "" + + manufacturer = prompt("Manufacturer", manufacturer) + model = prompt("Model", model) + variant = prompt("Variant (e.g. daylight / warm white, Enter if none)", + previous.get("variant", "")).lower() + notes = prompt("Notes", previous.get("notes", "retailer-internal barcode, brand unknown" + if ean and is_restricted(ean) and not manufacturer else "")) + rated = prompt_rated(previous.get("rated")) + + if ean and (manufacturer or model): + cache = load_cache() + cache[ean] = {"manufacturer": manufacturer, "model": model} + save_cache(cache) + + input("Insert the lamp, close the sphere, press Enter to start ... ") + + port = args.port or find_hpcs_port() + if not port: + print("ERROR: HPCS 6500 not found. Connect the device or use --port.") + return + psu = {"mode": "ac", "voltage": args.voltage, "frequency": args.frequency, + "current": None, "settle": args.settle, "interval": args.interval, + "enable": True} + readings, supply = readings_from_device(port, args.readings, False, psu) + reading = average_readings(readings) + meta = {"name": lamp_id, "ean": ean or "", "variant": variant, + "manufacturer": manufacturer, "model": model, "notes": notes, + "rated": rated} + out_dir = data_repo / "lamps" / lamp_id + tm30 = write_bundle(reading, out_dir, meta, len(readings), supply) + + print(f"\n {reading.get('Phi_lm', 0):.0f} lm {reading.get('eta_lm_W', 0):.1f} lm/W " + f"{reading.get('CCT_K', 0):.0f} K Ra {reading.get('Ra', 0):.1f} " + f"TM-30 Rf {tm30['Rf']:.1f} / Rg {tm30['Rg']:.1f}") + if rated.get("Phi_lm"): + delta = reading.get("Phi_lm", 0) / rated["Phi_lm"] * 100 - 100 + print(f" vs advertised: {delta:+.1f} % flux", end="") + if rated.get("CCT_K"): + print(f", {reading.get('CCT_K', 0) - rated['CCT_K']:+.0f} K CCT", end="") + print() + print() + + label_parts = [x for x in (manufacturer, model) if x] + if variant: + label_parts.append(variant) + if ean: + label_parts.append(f"EAN {ean}") + if not args.no_commit: + commit_bundle(data_repo, lamp_id, ", ".join(label_parts) or "unidentified", + existed=bool(previous), push=not args.no_push) + if args.web_repo: + scaffold_device_page(Path(args.web_repo), lamp_id, meta) + + +def main(): + parser = argparse.ArgumentParser(description="Barcode-driven lamp measurement station") + parser.add_argument("--data-repo", default=str(DEFAULT_DATA_REPO), + help="comparison-data checkout (bundles land in lamps//)") + parser.add_argument("--web-repo", default=str(DEFAULT_WEB_REPO), + help="buildfor_life_web checkout for device page stubs ('' to skip)") + parser.add_argument("--port", help="COM port (auto-detect if omitted)") + parser.add_argument("--readings", type=int, default=5) + parser.add_argument("--voltage", type=float, default=230.0) + parser.add_argument("--frequency", type=float, default=50.0) + parser.add_argument("--settle", type=float, default=60.0, + help="Seconds of stabilization after PSU on (procedure: 60)") + parser.add_argument("--interval", type=float, default=1.0) + parser.add_argument("--no-commit", action="store_true", + help="Write the bundle but skip git entirely") + parser.add_argument("--no-push", action="store_true", + help="Commit locally but do not push") + args = parser.parse_args() + + data_repo = Path(args.data_repo) + # In a submodule checkout .git is a file, not a directory; exists() covers both. + if not (data_repo / ".git").exists(): + print(f"ERROR: {data_repo} is not a git checkout of comparison-data.") + sys.exit(1) + + print("Lamp station ready. Scan a barcode, 'n' for a lamp without one, 'q' to quit.") + while True: + try: + code = input("\nScan barcode: ").strip() + except (EOFError, KeyboardInterrupt): + break + if code == "": + continue + if code.lower() in ("q", "quit", "exit"): + break + if code.lower() in ("n", "none"): + ean = "" + else: + ean = ean_normalize(code) + if not ean: + print(f" Not a valid EAN-13/UPC-A: {code!r}") + continue + try: + measure_one(ean, args, data_repo) + except KeyboardInterrupt: + print("\n Aborted; lamp skipped. (PSU is switched off by the driver.)") + except (subprocess.CalledProcessError, SystemExit) as e: + err = getattr(e, "stderr", "") or e + print(f" ERROR: {err}") + print("Bye.") + + +if __name__ == "__main__": + main()