#!/usr/bin/env python3 """Replay the Serval HTTP sequence that BEC + the ADASItpx 2.2 IOC send for a timepix_scan. No BEC, IOC or tpx3app needed: only Python 3 (stdlib) and a running Serval. Per scan point the IOC (asiTpx.cpp startMeasurement, release 2.2) does: 1. GET /detector/config, overwrite its own fields, PUT the whole config back (Tdc = ["P0123", "P0123"] because BEC enables TDC1 and TDC2, rising edge, all channels) 2. GET /server/destination, set Raw[0] (and Image), PUT it back 3. GET /measurement/start (nTriggers = 1, so Serval ends the measurement after one frame) BEC waits for the frame and starts the next point the same way. The script prints Serval's Tdc1/Tdc2 event rates per point. On this detector (2 boards x 4 chips) TDC1 at ~130 kHz reads ~1.04 MHz when all 8 chips see it and ~520 kHz when one board loses it. Switches to find which part of the sequence triggers the loss: --tdc2 / --no-tdc2 TDC2 "P0123" (BEC) or "" (Accos) --merge / --no-merge re-send Serval's full config (IOC 2.2) or only the IOC fields (IOC <= 2.1) --chain-mode NONE what the IOC sent before release 2.2 (BEC scans worked then) --one-measurement one measurement with nTriggers = points (Accos style) instead of one per point --configure-once send config + destination only before the first point, then only GET /measurement/start per point (no config back and forth) --software-trigger one measurement (SOFTWARESTART_TIMERSTOP, nTriggers = points), then GET /measurement/trigger/start per point (IOC: TriggerMode Software + cam1:TriggerSoftware); config is sent once --raw-base raw output, e.g. file:///data/SavedStream/ or tcp://connect@127.0.0.1:8451 (a tcp target needs a listener; omit for no raw output) Examples: python3 serval_bec_sequence.py --dry-run python3 serval_bec_sequence.py --points 40 --raw-base file:///data/SavedStream/ --raw-pattern bec_seq_%Hms_ python3 serval_bec_sequence.py --points 40 --no-tdc2 python3 serval_bec_sequence.py --points 40 --one-measurement python3 serval_bec_sequence.py --points 40 --configure-once python3 serval_bec_sequence.py --points 40 --software-trigger """ from __future__ import annotations import argparse import json import sys import time import urllib.error import urllib.request class Serval: def __init__(self, url: str, dry_run: bool): self.url = url.rstrip("/") self.dry_run = dry_run def _request(self, method: str, path: str, body: dict | None = None): data = None if body is None else json.dumps(body).encode() stamp = time.strftime("%H:%M:%S") print(f"{stamp} {method} {path}" + (f" {json.dumps(body)}" if body is not None else "")) if self.dry_run: return None req = urllib.request.Request(self.url + path, data=data, method=method) if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=30) as resp: text = resp.read().decode() except urllib.error.HTTPError as exc: sys.exit(f"{method} {path} failed: {exc.code} {exc.read().decode()}") try: return json.loads(text) if text else None except json.JSONDecodeError: return text def get(self, path: str): return self._request("GET", path) def put(self, path: str, body: dict): return self._request("PUT", path, body) def dashboard(self) -> dict: if self.dry_run: return {"Measurement": {"Status": "DA_IDLE"}} req = urllib.request.Request(self.url + "/dashboard") with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) def configure_detector( serval: Serval, args, n_triggers: int, trigger_mode: str = "AUTOTRIGSTART_TIMERSTOP" ) -> None: config = (serval.get("/detector/config") or {}) if args.merge else {} config["ExposureTime"] = args.exposure config["BiasEnabled"] = 1 # the IOC sends an int, not a JSON boolean config["BiasVoltage"] = args.bias config["ChainMode"] = args.chain_mode config["TriggerPeriod"] = args.period config["nTriggers"] = n_triggers config["TriggerDelay"] = 0.0 config["TriggerIn"] = 0 config["TriggerOut"] = 0 config["TriggerMode"] = trigger_mode config["Tdc"] = ["P0123", "P0123" if args.tdc2 else ""] serval.put("/detector/config", config) def configure_destination(serval: Serval, args) -> None: destination = (serval.get("/server/destination") or {}) if args.merge else {} if args.raw_base: raw = (destination.get("Raw") or [{}])[0] raw["Base"] = args.raw_base raw["FilePattern"] = args.raw_pattern if args.raw_base.startswith("tcp://"): raw.pop("SplitStrategy", None) destination["Raw"] = [raw] else: destination["Raw"] = None destination["Image"] = [] destination["Preview"] = None serval.put("/server/destination", destination) def wait_for_measurement(serval: Serval, timeout: float) -> tuple[float, float, int]: """Poll /dashboard until DA_IDLE; return max Tdc1/Tdc2 rates and dropped frames.""" tdc1 = tdc2 = 0.0 dropped = 0 deadline = time.monotonic() + timeout time.sleep(0.5) while time.monotonic() < deadline: meas = serval.dashboard().get("Measurement") or {} tdc1 = max(tdc1, meas.get("Tdc1EventRate") or 0) tdc2 = max(tdc2, meas.get("Tdc2EventRate") or 0) dropped = meas.get("DroppedFrames") or 0 if meas.get("Status") == "DA_IDLE": break time.sleep(0.5) else: print(" measurement did not finish in time") return tdc1, tdc2, dropped def report(tdc1: float, tdc2: float, dropped: int) -> None: full = 1.0e6 # rough "all 8 chips" TDC1 rate at the beamline flag = " <-- below ~1 MHz, a board may have lost the TDC" if 0 < tdc1 < 0.75 * full else "" print(f" Tdc1EventRate {tdc1:.0f} Tdc2EventRate {tdc2:.0f} DroppedFrames {dropped}{flag}") def wait_for_frame(serval: Serval, frames: int, timeout: float) -> tuple[float, float, int]: """Poll /dashboard until FrameCount reaches frames; return max Tdc1/Tdc2 rates and drops.""" tdc1 = tdc2 = 0.0 dropped = 0 deadline = time.monotonic() + timeout time.sleep(0.5) while time.monotonic() < deadline: meas = serval.dashboard().get("Measurement") or {} tdc1 = max(tdc1, meas.get("Tdc1EventRate") or 0) tdc2 = max(tdc2, meas.get("Tdc2EventRate") or 0) dropped = meas.get("DroppedFrames") or 0 if serval.dry_run or (meas.get("FrameCount") or 0) >= frames: break time.sleep(0.5) else: print(" frame did not arrive in time") return tdc1, tdc2, dropped def run_software_trigger(serval: Serval, args) -> None: """One measurement, config sent once, one software trigger per point.""" configure_detector(serval, args, args.points, trigger_mode="SOFTWARESTART_TIMERSTOP") configure_destination(serval, args) serval.get("/measurement/start") for point in range(1, args.points + 1): print(f"--- point {point}/{args.points}") serval.get("/measurement/trigger/start") report(*wait_for_frame(serval, point, args.period + 30)) if point < args.points: time.sleep(args.pause) serval.get("/measurement/stop") def main() -> None: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--url", default="http://localhost:8080") parser.add_argument("--points", type=int, default=40) parser.add_argument("--exposure", type=float, default=9.9979, help="s, BEC uses exp_time - 2.1 ms") parser.add_argument("--period", type=float, default=10.0, help="s") parser.add_argument("--bias", type=float, default=100.0) parser.add_argument("--chain-mode", default="LEADER", choices=["LEADER", "NONE"], help="LEADER = IOC 2.2 and Accos; NONE = what the IOC sent before 2.2") parser.add_argument("--tdc2", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--merge", action=argparse.BooleanOptionalAction, default=True) mode = parser.add_mutually_exclusive_group() mode.add_argument("--one-measurement", action="store_true") mode.add_argument("--configure-once", action="store_true") mode.add_argument("--software-trigger", action="store_true") parser.add_argument("--raw-base", default="") parser.add_argument("--raw-pattern", default="") parser.add_argument("--pause", type=float, default=1.0, help="s between points (BEC: ~1 s)") parser.add_argument("--dry-run", action="store_true", help="print the requests only") args = parser.parse_args() serval = Serval(args.url, args.dry_run) if args.software_trigger: run_software_trigger(serval, args) return runs = 1 if args.one_measurement else args.points n_triggers = args.points if args.one_measurement else 1 timeout = n_triggers * args.period + 30 for run in range(1, runs + 1): print(f"--- measurement {run}/{runs}") if run == 1 or not args.configure_once: configure_detector(serval, args, n_triggers) configure_destination(serval, args) serval.get("/measurement/start") report(*wait_for_measurement(serval, timeout)) if run < runs: time.sleep(args.pause) if __name__ == "__main__": main()