grabowski
7ef2960254
Every run opens with the freshly allocated L#### and then asks for the EAN, which Enter skips. Typing an existing ID re-measures that lamp with all identification and rated values pre-filled, which also rewrites old bundles in the current metrics format. Unused IDs are never burned since numbering derives from written bundles.
369 lines
15 KiB
Python
369 lines
15 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. Every run opens with a freshly allocated ID (L0012, ...). The ID only
|
|
becomes real once the bundle is written, so aborting never burns one.
|
|
2. Scan the box with a keyboard-wedge barcode scanner (types EAN +
|
|
Enter), press Enter for a lamp without a barcode, or type an existing
|
|
ID (e.g. L0004) to re-measure that lamp with everything pre-filled,
|
|
which also rewrites its bundle in the current metrics format. Scanned
|
|
EANs are checksum-validated; lamps already measured with the same EAN
|
|
are listed (pick one to re-measure, or continue as the 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. The
|
|
website picks it up automatically (data-driven device list plus the
|
|
auto-bump deploy); an .md page in the web repo is only ever needed
|
|
for photos, where-to-buy links, or notes.
|
|
|
|
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")
|
|
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 measure_one(new_id, ean, lamps, args, data_repo, existing_id=None):
|
|
# The run already owns its fresh ID (new_id). Typing an existing ID at
|
|
# the scan prompt re-measures that lamp instead (all its identification
|
|
# pre-filled, bundle rewritten in the current format); an EAN may also
|
|
# redirect to a re-measure when it matches an existing lamp.
|
|
lamp_id, previous = new_id, {}
|
|
if existing_id:
|
|
lamp_id = existing_id
|
|
previous = lamps[existing_id]
|
|
ean = previous.get("ean", "")
|
|
print(f" Re-measuring {lamp_id}" + (f" (EAN {ean})" if ean else ""))
|
|
elif 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(f"Re-measure one of these (type its ID) or Enter to continue as {new_id}").strip()
|
|
if choice:
|
|
if choice.upper() not in same:
|
|
print(f" Unknown ID {choice!r}, aborting this run.")
|
|
return
|
|
lamp_id = choice.upper()
|
|
previous = same[lamp_id]
|
|
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)
|
|
print(" The lamp publishes automatically (auto-bump deploy); add an .md under")
|
|
print(" src/content/comparisons/lamps/ only for photos, buy links, or notes.")
|
|
|
|
|
|
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("--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. Every run starts a new lamp ID; the EAN is optional")
|
|
print("(scan it, or press Enter to skip). 'q' quits.")
|
|
while True:
|
|
# A fresh ID opens every run. It only becomes real once a bundle is
|
|
# written, so quitting or re-measuring never burns a number.
|
|
lamps = lamp_index(data_repo)
|
|
new_id = next_lamp_id(data_repo)
|
|
print(f"\n--- New lamp {new_id} ---")
|
|
try:
|
|
code = input("Scan EAN (Enter to skip, L#### re-measures an existing lamp, q quits): ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
break
|
|
if code.lower() in ("q", "quit", "exit"):
|
|
break
|
|
existing_id = None
|
|
ean = ""
|
|
if ID_RE.match(code.upper()):
|
|
existing_id = code.upper()
|
|
if existing_id not in lamps:
|
|
print(f" No lamp {existing_id} in the database.")
|
|
continue
|
|
elif code and code.lower() not in ("n", "none"):
|
|
ean = ean_normalize(code)
|
|
if not ean:
|
|
print(f" Not a valid EAN-13/UPC-A: {code!r} (Enter would skip the EAN)")
|
|
continue
|
|
try:
|
|
measure_one(new_id, ean, lamps, args, data_repo, existing_id=existing_id)
|
|
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()
|