Add a RAWSEL sweep example that captures one RAW file per channel

README.md

@@ -233,6 +233,12 @@ Commands release the GIL while they run, so a blocking `expose()` leaves the res
 
 Logging follows `LOG_STDERR` from the `.cfg`; pass `log_to_stderr=` to override it per session. The C++ log always goes to its daily file under `LOGPATH`.
 
+`python/examples/raw_channel_sweep.py` uses the module to capture a pre-CDS waveform from several channels. The Archon covers one channel per exposure, so the script loops `RAWSEL`, exposes and fetches, leaving one RAW file per channel next to each paired image.
+
+```bash
+$ python python/examples/raw_channel_sweep.py --config lris2.cfg --fits-dir /tmp/images --channels 0-3,18-21
+```
+
 ## Frame Outputs
 
 Every instrument publishes each acquired frame to one or more outputs, configured entirely via `.cfg` file keys (`Camera::Interface::configure_frame_outputs()` builds them from `Camera::apply_config_overrides()`, called once at startup for every instrument, not just HISPEC). Both outputs are independent; either, both, or neither can be enabled per instrument.

docs/python/index.md

@@ -143,3 +143,15 @@ The C++ log always goes to its daily file under `LOGPATH` regardless.
 `python/examples/shm_read_frames.py` is a streaming shared-memory consumer that blocks on the
 stream's semaphore and flags frames it missed. See
 [frame outputs](../configuration/frame-outputs.md) for what it needs.
+
+`python/examples/raw_channel_sweep.py` captures a pre-CDS waveform from each of several channels.
+The Archon covers one channel per exposure, so the script loops `RAWSEL`, exposes, and fetches,
+ending with one RAW file per channel alongside each paired image:
+
+```bash
+python python/examples/raw_channel_sweep.py --config lris2.cfg \
+    --fits-dir /tmp/images --channels 0-3,18-21
+```
+
+It reports the module inventory from `SYSTEM` rather than inferring a slot from `RAWSEL`, for the
+reason given under [raw samples](../commands/controller.md#raw-samples).

python/examples/raw_channel_sweep.py

@@ -0,0 +1,223 @@
+"""Capture a pre-CDS RAW waveform from each of several Archon channels.
+
+The Archon captures raw samples for one channel per exposure, selected by
+RAWSEL, so covering more than one channel means one exposure each. This script
+loops that: set RAWSEL, expose, fetch, and record which file resulted.
+
+RAWSEL is applied before the exposure because the controller acquires raw
+samples while it fills the frame buffer, so enabling it afterwards cannot
+populate that buffer.
+
+No slot is inferred from RAWSEL. The manual maps four channels per slot while
+the Archon GUI offers 72 selections, which is eighteen per slot, and the two
+disagree on a mixed chassis. The module inventory is reported from SYSTEM
+instead, and each file's header carries RAWSEL beside every candidate slot type.
+
+    python raw_channel_sweep.py --config lris2.cfg --fits-dir /tmp/images \\
+        --channels 0-3,18-21
+
+Needs a camera_interface module built for an Archon controller, and an ACF that
+defines the six RAW keys, since camerad can only write keys it already loaded.
+It loads that ACF, so do not point it at a controller someone else is using.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import pathlib
+import sys
+import time
+from types import MappingProxyType
+from typing import Final
+
+import camera_interface
+
+# From archongui/src/archon.h, which carries the full table where the Archon
+# manual stops at "16+: Unknown"
+MODULE_NAMES: Final = MappingProxyType({
+    0: "None", 1: "Driver", 2: "AD", 3: "LVBias", 4: "HVBias", 5: "Heater",
+    6: "Atlas", 7: "HS", 8: "HVXBias", 9: "LVXBias", 10: "LVDS", 11: "HeaterX",
+    12: "XVBias", 13: "ADF", 14: "ADX", 15: "ADLN", 16: "DriverX", 17: "ADM",
+    18: "Unknown",
+})
+
+# Slots an AD or ADM module may occupy, and therefore the ones RAWSEL addresses
+AD_SLOTS: Final = range(5, 9)
+
+RAW_COMMAND: Final = "raw"
+
+# The FITS writer queues and drops by design, so a caller wanting the file has
+# to wait for it rather than expect it synchronously
+WRITE_TIMEOUT_S: Final = 60.0
+POLL_INTERVAL_S: Final = 0.2
+
+
+class SweepError(Exception):
+    """A step of the sweep failed in a way that should stop it."""
+
+
+def parse_channels(spec: str) -> list[int]:
+    """Expand a channel spec such as "0-3,7,18-21" into a sorted channel list."""
+    channels: set[int] = set()
+    for part in spec.split(","):
+        piece = part.strip()
+        if not piece:
+            continue
+        if "-" in piece:
+            low, _, high = piece.partition("-")
+            start, end = int(low), int(high)
+            if end < start:
+                raise ValueError(f"descending range: {piece}")
+            channels.update(range(start, end + 1))
+        else:
+            channels.add(int(piece))
+    if not channels:
+        raise ValueError("no channels selected")
+    return sorted(channels)
+
+
+def read_module_types(camera: camera_interface.Camera) -> dict[int, int]:
+    """Return MODn_TYPE for each AD-capable slot, parsed from the SYSTEM report."""
+    report = camera.native("SYSTEM")
+    types: dict[int, int] = {}
+    for field in report.split():
+        key, _, value = field.partition("=")
+        for slot in AD_SLOTS:
+            if key == f"MOD{slot}_TYPE":
+                types[slot] = int(value)
+    return types
+
+
+def describe_modules(types: dict[int, int]) -> str:
+    """Render the AD-capable slots as "slot 5: ADM" entries for logging."""
+    if not types:
+        return "no module types reported"
+    return ", ".join(f"slot {slot}: {MODULE_NAMES.get(kind, kind)}"
+                     for slot, kind in sorted(types.items()))
+
+
+def wait_for_raw_file(fits_dir: pathlib.Path,
+                      already_present: set[pathlib.Path]) -> pathlib.Path:
+    """Return the RAW file that appears after a fetch, or raise on timeout."""
+    deadline = time.monotonic() + WRITE_TIMEOUT_S
+    while time.monotonic() < deadline:
+        new = sorted(set(fits_dir.glob("*_raw.fits")) - already_present)
+        if new:
+            return new[-1]
+        time.sleep(POLL_INTERVAL_S)
+    raise SweepError(f"no new *_raw.fits appeared in {fits_dir} "
+                     f"within {WRITE_TIMEOUT_S}s")
+
+
+def capture_channel(camera: camera_interface.Camera, channel: int,
+                    geometry: list[str],
+                    fits_dir: pathlib.Path) -> pathlib.Path:
+    """Expose with RAWSEL on one channel, fetch the capture, return its file."""
+    settings = ["RAWENABLE", "1", "RAWSEL", str(channel), *geometry]
+    camera.controller_cmd(RAW_COMMAND, "set " + " ".join(settings))
+
+    already_present = set(fits_dir.glob("*_raw.fits"))
+    camera.expose("1")
+    summary = camera.controller_cmd(RAW_COMMAND, "read")
+    print(f"  RAWSEL={channel}: {summary}")
+    return wait_for_raw_file(fits_dir, already_present)
+
+
+def geometry_settings(args: argparse.Namespace) -> list[str]:
+    """Build the RAW geometry key and value pairs the caller chose to override."""
+    overrides = {
+        "RAWSAMPLES": args.samples,
+        "RAWSTARTLINE": args.start_line,
+        "RAWENDLINE": args.end_line,
+        "RAWSTARTPIXEL": args.start_pixel,
+    }
+    return [token
+            for key, value in overrides.items() if value is not None
+            for token in (key, str(value))]
+
+
+def sweep(camera: camera_interface.Camera, channels: list[int],
+          geometry: list[str],
+          fits_dir: pathlib.Path) -> tuple[dict[int, pathlib.Path], dict[int, str]]:
+    """Capture each channel in turn, returning the files and any failures."""
+    captured: dict[int, pathlib.Path] = {}
+    failures: dict[int, str] = {}
+    for channel in channels:
+        try:
+            captured[channel] = capture_channel(camera, channel, geometry, fits_dir)
+        except (SweepError, RuntimeError) as failure:
+            print(f"  RAWSEL={channel}: FAILED {failure}", file=sys.stderr)
+            failures[channel] = str(failure)
+    return captured, failures
+
+
+def run(args: argparse.Namespace) -> int:
+    """Sweep the requested channels and report where each capture landed."""
+    controller = camera_interface.controller_name()
+    if controller != "archon":
+        raise SweepError(f"RAW capture is Archon-only, but this module is "
+                         f"built for {controller}")
+
+    channels = parse_channels(args.channels)
+    fits_dir = pathlib.Path(args.fits_dir)
+    if not fits_dir.is_dir():
+        raise SweepError(f"not a directory: {fits_dir}")
+
+    # The binding exposes no __enter__, so closing() is what guarantees the
+    # controller is released even when a capture raises
+    with contextlib.closing(camera_interface.Camera(args.config,
+                                                    log_to_stderr=True)) as camera:
+        camera.open()
+        camera.load()
+        camera.power("on")
+        if args.exptime is not None:
+            camera.exptime(str(args.exptime))
+
+        print(f"instrument={camera_interface.instrument_name()} "
+              f"exptime={camera.exptime()}")
+        print(f"modules: {describe_modules(read_module_types(camera))}")
+        print(f"sweeping RAWSEL {channels}")
+
+        try:
+            captured, failures = sweep(camera, channels,
+                                       geometry_settings(args), fits_dir)
+        finally:
+            # Leave capture off so a later exposure does not silently carry raw data
+            camera.controller_cmd(RAW_COMMAND, "set RAWENABLE 0")
+
+    print(f"\ncaptured {len(captured)} of {len(channels)} channels")
+    for channel, path in captured.items():
+        print(f"  RAWSEL={channel} -> {path.name} ({path.stat().st_size} bytes)")
+    return 1 if failures else 0
+
+
+def main() -> int:
+    """Parse arguments, run the sweep, and report pass or fail."""
+    parser = argparse.ArgumentParser(description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--config", required=True, help="camerad .cfg to use")
+    parser.add_argument("--fits-dir", required=True,
+                        help="directory the FITS writer is configured to use")
+    parser.add_argument("--channels", default="0-3",
+                        help="RAWSEL values, e.g. \"0-3,18-21\" (default: 0-3)")
+    parser.add_argument("--exptime", type=float,
+                        help="exposure time, in the unit the config selects")
+    parser.add_argument("--samples", type=int, help="RAWSAMPLES override")
+    parser.add_argument("--start-line", type=int, help="RAWSTARTLINE override")
+    parser.add_argument("--end-line", type=int, help="RAWENDLINE override")
+    parser.add_argument("--start-pixel", type=int, help="RAWSTARTPIXEL override")
+    args = parser.parse_args()
+
+    try:
+        return run(args)
+    except (SweepError, ValueError) as failure:
+        print(f"FAIL: {failure}", file=sys.stderr)
+        return 1
+    except RuntimeError as failure:
+        print(f"FAIL: camera command failed: {failure}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())