grabowski
fad8ec0813
Station prompts for it (default led), metrics.json carries it as 'type', and the scaffolded device page description and Type spec follow it.
378 lines
15 KiB
Python
378 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. 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}"
|
|
|
|
|
|
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.get("type"):
|
|
specs.append(("Type", meta["type"].upper() if meta["type"] in ("led", "cfl") else meta["type"].capitalize()))
|
|
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(",", " ")))
|
|
|
|
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"',
|
|
"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)
|
|
lamp_type = prompt("Type (led / halogen / cfl / incandescent / ...)",
|
|
previous.get("type", "led")).lower().strip()
|
|
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, "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()
|