grabowski
8f667dc6cf
The linear 380-1050 nm axis gives spectrum chromaticities 0.09-0.13 (xy) away from the instrument's own x, y, so TM-30 and spd.csv inherit an axis error until it is confirmed. tools/make_fixtures.py regenerates the golden data from the python-final driver. The release profile moves to the workspace that builds the binary.
143 lines
6.2 KiB
Python
143 lines
6.2 KiB
Python
"""Golden fixtures for the Rust hpcs6500 driver, produced by the legacy Python driver.
|
|
|
|
Run it next to hpcs6500.py from tag python-final, with the USB captures in
|
|
C:\dev\HPCS\captures (not in git):
|
|
git show python-final:hpcs6500.py > /tmp/py/hpcs6500.py; cp tools/make_fixtures.py /tmp/py/
|
|
python /tmp/py/make_fixtures.py <out> # then copy <out>/fixtures and <out>/golden to tests/
|
|
pyfmt.json was trimmed to ~2000 cases afterwards.
|
|
|
|
Writes into <out>/:
|
|
fixtures/rNN_meas.bin, rNN_elec.bin raw blocks from real USB captures
|
|
golden/readings.json Python dict per fixture reading (floats as repr strings)
|
|
golden/format_*.txt format_reading / format_quick output
|
|
golden/csv_row.txt CSV row (without timestamp) as DictWriter writes it
|
|
golden/pyfmt.json repr / .6g / .2f / .0f / .1f of many floats
|
|
golden/parse_<capture>.txt `hpcs6500.py --parse` output, full and --quick --spectrum
|
|
"""
|
|
import contextlib
|
|
import csv
|
|
import io
|
|
import json
|
|
import math
|
|
import random
|
|
import struct
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
import hpcs6500 as h # noqa: E402
|
|
|
|
out = Path(sys.argv[1])
|
|
captures = Path(r"C:\dev\HPCS\captures")
|
|
(out / "fixtures").mkdir(parents=True, exist_ok=True)
|
|
(out / "golden").mkdir(parents=True, exist_ok=True)
|
|
|
|
pairs = [] # (name, meas, elec)
|
|
for f in sorted(captures.glob("*.pcap")):
|
|
if f.stat().st_size < 100:
|
|
continue
|
|
m = h.parse_pcap_messages(str(f))
|
|
b = [x for x in m if x["dir"] == "RX" and len(x["data"]) == 3904 and x["data"][:8] == b"HPCS6500"]
|
|
e = [x for x in m if x["dir"] == "RX" and len(x["data"]) == 1584]
|
|
for i, blk in enumerate(b):
|
|
el = e[i]["data"] if i < len(e) else None
|
|
harm = el is not None and abs(struct.unpack_from("<f", el, 544)[0] - 100) < 1
|
|
pairs.append((f.stem, i, blk["data"], el, harm))
|
|
|
|
print(len(pairs), "readings,", sum(p[4] for p in pairs), "with harmonics")
|
|
# pick: first, one with harmonics, one with zero electrical, and a few spread out
|
|
chosen = []
|
|
seen = set()
|
|
def pick(p):
|
|
key = (p[0], p[1])
|
|
if key not in seen:
|
|
seen.add(key)
|
|
chosen.append(p)
|
|
pick(pairs[0])
|
|
for p in pairs:
|
|
if p[4]:
|
|
pick(p)
|
|
break
|
|
for p in pairs:
|
|
if p[3] is not None and struct.unpack_from("<f", p[3], 8)[0] == 0:
|
|
pick(p)
|
|
break
|
|
for k in range(0, len(pairs), max(1, len(pairs) // 4)):
|
|
pick(pairs[k])
|
|
pick(pairs[-1])
|
|
|
|
dev = h.HPCS6500.__new__(h.HPCS6500)
|
|
|
|
|
|
def jsonable(v):
|
|
if isinstance(v, float):
|
|
return {"f": repr(v)}
|
|
if isinstance(v, list):
|
|
return [jsonable(x) for x in v]
|
|
return v
|
|
|
|
|
|
readings = []
|
|
for n, (cap, i, meas, elec, harm) in enumerate(chosen, 1):
|
|
name = f"r{n:02d}"
|
|
(out / "fixtures" / f"{name}_meas.bin").write_bytes(meas)
|
|
if elec is not None:
|
|
(out / "fixtures" / f"{name}_elec.bin").write_bytes(elec)
|
|
r = dev._parse_measurement(meas)
|
|
if elec is not None:
|
|
r.update(dev._parse_electrical(elec))
|
|
readings.append({"name": name, "source": f"{cap}#{i}", "harmonics": harm,
|
|
"keys": list(r.keys()), "values": {k: jsonable(v) for k, v in r.items()}})
|
|
(out / "golden" / f"format_{name}.txt").write_text(h.format_reading(r), encoding="utf-8", newline="\n")
|
|
(out / "golden" / f"format_{name}_spectrum.txt").write_text(h.format_reading(r, show_spectrum=True), encoding="utf-8", newline="\n")
|
|
(out / "golden" / f"quick_{name}.txt").write_text(h.format_quick(r, n) + "\n" + h.format_quick(r), encoding="utf-8", newline="\n")
|
|
buf = io.StringIO()
|
|
fieldnames = ["timestamp"] + list(h.FIELDS.keys()) + list(h.ELECTRICAL_FIELDS.keys())
|
|
w = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore")
|
|
w.writeheader()
|
|
row = {"timestamp": "T"}
|
|
row.update({k: v for k, v in r.items() if isinstance(v, (int, float))})
|
|
w.writerow(row)
|
|
(out / "golden" / f"csv_{name}.txt").write_bytes(buf.getvalue().encode())
|
|
(out / "golden" / "readings.json").write_text(json.dumps(readings, indent=1))
|
|
print("chosen:", [(c[0], c[1], c[4]) for c in chosen])
|
|
|
|
# --parse output for two real captures
|
|
for cap in ["usb_capture_20260204_123655_hub1", "usb_capture_20260204_134301_hub1"]:
|
|
for flags, suffix in (([], ""), (["--quick"], "_quick"), (["--harmonics"], "_harmonics")):
|
|
sys.argv = ["hpcs6500.py", *flags, "--parse", str(captures / f"{cap}.pcap")]
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
h.main()
|
|
(out / "golden" / f"parse_{cap}{suffix}.txt").write_text(buf.getvalue(), encoding="utf-8", newline="\n")
|
|
|
|
# float formatting golden
|
|
rng = random.Random(6500)
|
|
vals = [0.0, -0.0, 1.0, -1.0, 0.1, 0.5, 1.5, 2.5, 0.125, 0.375, 1e-5, 1e-4, 9.99995e-5, 123456.5, 999999.5,
|
|
1e15, 1e16, 1.2345678901234567e16, 1e22, 1e-7, 5e-324, 1.7976931348623157e308, 100.0, 380.0,
|
|
float("nan"), float("inf"), -float("inf"), 0.000123456789, 12345678.9, 0.999999, 0.9999995,
|
|
9.9999995, 99999.95, 1234567.0, 0.1 + 0.2]
|
|
vals += [380 + i * h.SPECTRUM_STEP for i in range(350)]
|
|
for meas_vals in readings:
|
|
for k, v in meas_vals["values"].items():
|
|
if isinstance(v, dict):
|
|
vals.append(float(v["f"]))
|
|
elif isinstance(v, list) and v and isinstance(v[0], dict):
|
|
vals.extend(float(x["f"]) for x in v[:60])
|
|
for _ in range(3000):
|
|
e = rng.uniform(-12, 18)
|
|
x = rng.choice([-1, 1]) * 10 ** e
|
|
vals.append(x)
|
|
vals.append(struct.unpack("<f", struct.pack("<f", x if abs(x) < 3e38 else 1.0))[0])
|
|
for _ in range(500):
|
|
vals.append(rng.randint(0, 10**7) / rng.choice([2, 4, 8, 10, 100, 1000, 16, 1024]))
|
|
rows = []
|
|
for v in vals:
|
|
rows.append({"bits": struct.unpack("<Q", struct.pack("<d", v))[0], "repr": repr(v), "g6": format(v, ".6g"),
|
|
"f2": format(v, ".2f") if abs(v) < 1e30 or math.isnan(v) or math.isinf(v) else None,
|
|
"f0": format(v, ".0f") if abs(v) < 1e30 or math.isnan(v) or math.isinf(v) else None,
|
|
"f1": format(v, ".1f") if abs(v) < 1e30 or math.isnan(v) or math.isinf(v) else None,
|
|
"round1": repr(round(v, 1)) if not (math.isnan(v) or math.isinf(v)) else None})
|
|
(out / "golden" / "pyfmt.json").write_text(json.dumps(rows))
|
|
print(len(rows), "float cases")
|