grabowski
9fdb12b533
Manufacturer input maps to one canonical spelling (seed list plus whatever the database and cache already use), so Philips is never philips/PHILIPS. Identity fields no longer duplicate into the device page specs (the site renders them from metrics.json); the weight scan ignores index.md so a fresh run starts at weight 1.
411 lines
17 KiB
Python
411 lines
17 KiB
Python
"""
|
|
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
|
|
<data-repo>/lamps/<ID>/.
|
|
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}"
|
|
|
|
|
|
# Canonical spellings for brands; extended at runtime by whatever is already
|
|
# in the database and the cache, so the first accepted spelling of a new brand
|
|
# becomes its canonical form for every later scan.
|
|
BRAND_SEED = [
|
|
"Philips", "OSRAM", "LAMPTAN", "TKL", "ST", "IKEA", "Panasonic",
|
|
"Toshiba", "GE", "Sylvania", "Xiaomi", "Opple", "EVE",
|
|
]
|
|
|
|
|
|
def canonical_brand(name, lamps):
|
|
"""Map a manufacturer string to its canonical capitalization."""
|
|
name = " ".join(name.split())
|
|
if not name:
|
|
return name
|
|
brands = {b.lower(): b for b in BRAND_SEED}
|
|
for entry in load_cache().values():
|
|
b = entry.get("manufacturer", "")
|
|
if b:
|
|
brands.setdefault(b.lower(), b)
|
|
for m in lamps.values():
|
|
b = m.get("manufacturer", "")
|
|
if b:
|
|
brands[b.lower()] = b # the database is the strongest source
|
|
return brands.get(name.lower(), name)
|
|
|
|
|
|
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") if f.name != "index.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']})"
|
|
# Identity fields (ID, EAN, type, dimmable, variant) live in metrics.json,
|
|
# which the detail page renders directly — only human-editable extras like
|
|
# the rated values go into the page's specs front matter.
|
|
rated = meta.get("rated", {})
|
|
specs = []
|
|
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(",", " ")))
|
|
|
|
type_word = {"led": "LED", "cfl": "CFL"}.get(meta.get("type", "led"), meta.get("type", "LED"))
|
|
lines = ["---", f'title: "{title}"']
|
|
if meta["manufacturer"]:
|
|
lines.append(f'manufacturer: "{meta["manufacturer"]}"')
|
|
lines += [
|
|
f'description: "Household {type_word} lamp, measured in our integrating sphere."',
|
|
f'csv: "/data/comparisons/lamps/{lamp_id}/spd.csv"',
|
|
# a bare "specs:" would parse as YAML null and fail the site's schema
|
|
*(["specs:"] if specs else []),
|
|
*[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 and commit the web repo (the submodule bump lands there")
|
|
print(" automatically via comparison-data's bump-website workflow; git pull first).")
|
|
|
|
|
|
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", canonical_brand(manufacturer, lamps))
|
|
canon = canonical_brand(manufacturer, lamps)
|
|
if canon != manufacturer:
|
|
print(f" Using canonical brand name: {canon}")
|
|
manufacturer = canon
|
|
model = prompt("Model", model)
|
|
lamp_type = prompt("Type (led / halogen / cfl / incandescent / ...)",
|
|
previous.get("type", "led")).lower().strip()
|
|
prev_dim = previous.get("dimmable")
|
|
dim_default = "" if prev_dim is None else ("y" if prev_dim else "n")
|
|
dim_raw = prompt("Dimmable? (y / n, Enter if unknown)", dim_default).lower()
|
|
dimmable = True if dim_raw.startswith("y") else False if dim_raw.startswith("n") else None
|
|
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 "", "type": lamp_type, "dimmable": dimmable,
|
|
"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/<ID>/)")
|
|
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()
|