grabowski
719ddcf04a
label_printer.py renders a DK-11221 23x23 mm label (QR code linking to the lamp's buildfor.life page plus the ID) and sends it over the network backend (brother-ql-next). The station prints two after each successful measurement, one for the box and one for the lamp; printer failures never lose a run, and any lamp can be reprinted with uv run label_printer.py <ID>. Printer address via HPCS_LABEL_PRINTER or --printer.
393 lines
17 KiB
Python
393 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. 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.
|
|
6. Two ID labels (QR to the lamp's page + the ID, DK-11221 23x23 mm)
|
|
print on the Brother QL-820NWB: one for the box, one for the lamp.
|
|
Configure the printer via HPCS_LABEL_PRINTER or --printer; reprint
|
|
any lamp later with `uv run label_printer.py L0006`.
|
|
|
|
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 os
|
|
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.")
|
|
|
|
# Two ID labels on the QL-820NWB: one for the box, one for the lamp.
|
|
if not args.no_labels:
|
|
if args.printer:
|
|
try:
|
|
from label_printer import print_labels
|
|
print_labels(lamp_id, args.printer, copies=2)
|
|
print(f" Printed 2 ID labels for {lamp_id}.")
|
|
except Exception as e: # noqa: BLE001 — a printer problem must not lose the run
|
|
print(f" Label printing failed ({e}); reprint later with:")
|
|
print(f" uv run label_printer.py {lamp_id}")
|
|
else:
|
|
print(f" (no label printer configured; set HPCS_LABEL_PRINTER or use --printer,")
|
|
print(f" reprint later with: uv run label_printer.py {lamp_id})")
|
|
|
|
|
|
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")
|
|
parser.add_argument("--printer", default=os.environ.get("HPCS_LABEL_PRINTER", ""),
|
|
help="QL-820NWB address, e.g. tcp://192.168.1.50 "
|
|
"(default: HPCS_LABEL_PRINTER env)")
|
|
parser.add_argument("--no-labels", action="store_true",
|
|
help="Skip printing the two ID labels after each lamp")
|
|
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()
|