debug: make the scan check work out its own geometry, or say why it cannot
CI for csaxs_bec / test (push) Successful in 1m47s

First run against real data (S00473, S00254) found the files but stopped at
"metadata: none of the expected keys found" and "could not work out the line
geometry". Two wrong assumptions, both now fixed.

**Metadata location.** It was looked up at two hard-coded paths. BEC nests this
differently across versions, so it is now found by searching the tree for the
keys by basename, attributes included. Verified against metadata one level
deeper than the original code expected.

**Which signal defines the geometry.** The point count was taken from the
*longest* signal. Real files hold signals of several lengths -- S00473 has
86884 points in something while the per-point arrays are shorter -- so a
monitor on its own clock hijacked the calculation. It now takes the length
shared by the most signals, since per-point data outnumbers everything else.

When the geometry still cannot be inferred, the script no longer just gives up
and asks for --list. It now prints every signal grouped by length, and the
factor pairs that could plausibly be (lines x points-per-line). That turns a
dead end into a readable diagnosis in one run rather than a round trip. If a
length does not factor usefully at all it says so explicitly, because a padded
or truncated per-point array is itself a finding.

Also wires up --lines/--points, which were declared but never reached the
analysis.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
menzelandClaude Opus 5 committed 2026-08-25 09:59:03 +02:00
1 parent a5035d91cb
commit 90358f5222
1 file changed
+63 -26
+63 -26
View File
@@ -94,28 +94,36 @@ def numeric_signals(h5: h5py.File) -> dict[str, np.ndarray]:
return out
WANTED_META = ("scan_name", "scan_number", "num_points", "num_lines", "frames_per_trigger")
def read_metadata(h5: h5py.File) -> dict:
"""Pull the scan bookkeeping we need, tolerating layout differences."""
"""Find the scan bookkeeping wherever it lives.
BEC nests this differently across versions, so search the whole tree for the
keys by *basename* rather than assuming a path. Attributes count too.
"""
meta: dict = {}
for path in ("/entry/collection/metadata", "/entry/collection"):
if path not in h5:
continue
node = h5[path]
for key in (
"scan_name",
"scan_number",
"num_points",
"num_lines",
"frames_per_trigger",
):
if key in node:
try:
val = node[key][()]
meta[key] = val.item() if hasattr(val, "item") else val
except Exception: # pylint: disable=broad-except
pass
for key, val in getattr(node, "attrs", {}).items():
meta.setdefault(key, val)
def take(key: str, raw):
if key not in WANTED_META or key in meta:
return
try:
val = raw[()] if isinstance(raw, h5py.Dataset) else raw
if isinstance(val, np.ndarray) and val.size == 1:
val = val.reshape(-1)[0]
meta[key] = val.item() if hasattr(val, "item") else val
except Exception: # pylint: disable=broad-except
pass
def visit(name, obj):
take(name.rsplit("/", 1)[-1], obj)
for akey, aval in getattr(obj, "attrs", {}).items():
take(akey, aval)
h5.visititems(visit)
for akey, aval in h5["/"].attrs.items():
take(akey, aval)
return {k: (v.decode() if isinstance(v, bytes) else v) for k, v in meta.items()}
@@ -256,7 +264,7 @@ def check_line_motion(signals, grid, fast_axis) -> None:
# --------------------------------------------------------------------------- #
def analyse(path: str, do_list: bool) -> None:
def analyse(path: str, do_list: bool, force_lines=None, force_points=None) -> None:
print("=" * 80)
print(os.path.basename(path))
print("=" * 80)
@@ -277,11 +285,40 @@ def analyse(path: str, do_list: bool) -> None:
if not signals:
print(" no 1-D numeric signals found -- re-run with --list")
return
n = max(len(v) for v in signals.values())
grid = infer_grid(n, meta)
lengths: dict[int, list[str]] = {}
for name, arr in signals.items():
lengths.setdefault(arr.size, []).append(name)
print(f" {len(signals)} numeric signals, in {len(lengths)} different lengths:")
for size in sorted(lengths, reverse=True):
names = sorted(lengths[size])
shown = ", ".join(names[:6]) + (f", +{len(names) - 6} more" if len(names) > 6 else "")
print(f" {size:>8} points : {shown}")
# Prefer the length shared by the most signals: per-point data outnumbers
# whatever else is in the file. Taking the longest signal is wrong when a
# monitor is sampled on its own clock.
n = max(lengths, key=lambda k: len(lengths[k]))
if force_lines and force_points:
grid = (force_lines, force_points)
n = force_lines * force_points
else:
grid = infer_grid(n, meta)
if grid is None:
print(f" could not work out the line geometry for {n} points.")
print(" Pass --lines / --points, or send me the --list output.")
print(f"\n could not work out the line geometry for {n} points.")
print(" factor pairs that could plausibly be (lines x points-per-line):")
pairs = [(d, n // d) for d in range(2, int(n**0.5) + 1) if n % d == 0]
pairs += [(b, a) for a, b in pairs]
plausible = sorted({p for p in pairs if 2 <= p[0] <= 2000 and 4 <= p[1] <= 20000})
if plausible:
for a, b in plausible[:12]:
print(f" {a:>6} lines x {b:>6} points")
if len(plausible) > 12:
print(f" ... and {len(plausible) - 12} more")
else:
print(f" none -- {n} does not factor usefully, so the per-point arrays")
print(" may be padded or truncated, which is itself worth knowing.")
print("\n Re-run with --lines N --points M once you know the scan geometry")
print(" (the scan command that produced it gives both).")
return
mon = monitored_names(h5)
fast = pick_fast_axis(signals, grid)
@@ -312,7 +349,7 @@ def main(argv=None) -> int:
for path in paths:
try:
analyse(path, args.list)
analyse(path, args.list, args.lines, args.points)
except Exception as exc: # pylint: disable=broad-except
print(f" FAILED on {path}: {type(exc).__name__}: {exc}")
print()