tools: zero_amp_offsets runs the current-amp server offset auto-zero

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 Tagged releases that apps pin (see CONTRIBUTING.md "Releases and the apps that pin them").
 
+## Unreleased
+
+- `zero_amp_offsets` (`spec-write`, mutating): runs the ssrl-current-amp server's
+  Zero Offsets routine — each SR570's input offset set to the smallest ladder step
+  whose cscan box ADC reading clears +2 mV — and waits for it, aborting (offsets
+  restored by the server) past `wait_s`. Audited with `audited_device_call`. New
+  module `current_amp.py`; URL `BTH_CURRENT_AMP_URL` (BL4-1 default
+  `http://192.168.41.6:8570`, unset elsewhere so the tool refuses).
+
 ## v0.7.0 — 2026-10-12
 
 Additive; BL15-2 defaults unchanged (unset `BTH_STATION` behaves as before).

README.md

@@ -11,7 +11,7 @@ Provides:
 - **CHERFD continuous scanning** — plan, run and reduce continuous scans on the cscan box through the cherfd control server, at BL15-2 (Vortex) and BL4-1 (24-element Ge) (`cherfd`, `cherfd-read`, `cherfd-write` trees; `beamtimehero ref cherfd-operations`). Mocked unless `CHERFD_MOCK=0`.
 - **Deployment backends** — the same scan-read surface served from S3DF Postgres + pickled scan data; Slack messaging.
 - **Log reads** — beamline control log parsing, search; read-only access to the station's application logs (`tool list-app-logs` / `read-app-log` / `search-app-logs`).
-- **Station services (BL4-1)** — read-only gas mixer / ion-chamber HV state, Ge detector status and Motor Analyzer diagnostics (`tool get-sample-environment`, `get-ge-detector-status`, `get-motor-diagnostics`); dark currents and check_beam (`spec-write measure-darks`, `set-check-beam`).
+- **Station services (BL4-1)** — read-only gas mixer / ion-chamber HV state, Ge detector status and Motor Analyzer diagnostics (`tool get-sample-environment`, `get-ge-detector-status`, `get-motor-diagnostics`); dark currents, check_beam and amplifier offset auto-zero (`spec-write measure-darks`, `set-check-beam`, `zero-amp-offsets`).
 - **Provenance** — per-experiment provenance records and results bundles with checksummed manifests (`tool record-provenance`, `export-results-bundle`).
 - **Action logging** — every command writes to a local SQLite audit trail.
 - **Reference docs** — `beamtimehero ref <name>` to fetch bundled procedure docs.

src/beamtimehero_cli/current_amp.py

@@ -0,0 +1,128 @@
+"""Client for the station's SRS current-amplifier server (ssrl-current-amp).
+
+BL4-1 runs ``ssrl-current-amp`` (``/usr/local/projects/ssrl-current-amp``):
+a web UI and EPICS IOC for the SR570 amplifiers on I0/I1/I2/PIPS. Its
+**Zero Offsets** routine (``devices/autozero.py`` there) sets each detector's
+input offset current to the smallest 1-2-5 ladder step whose cscan box ADC
+reading clears a small positive margin (BL4-1: +2 mV). The server owns that
+routine and all of its guards (at most 8 offset changes per detector, the
+probe must respond 0.5-2x the nominal gain, readings must follow the
+prediction, a 300 s limit for the run, every offset restored on an error or
+abort). This module only starts it, waits for it and reports the result:
+
+  POST /api/autozero/start   409 if a run is already going
+  GET  /api/autozero         {enabled, status: {state, step, log, results}}
+  POST /api/autozero/abort   on our own deadline: the server restores offsets
+
+``zero_offsets`` writes hardware, so the tool handler wraps it in
+``audited_call.audited_device_call`` (justification, safety switch, action
+log). The URL is ``BTH_CURRENT_AMP_URL``, else the station default; a
+station without the server gets ``ok: false`` saying so.
+"""
+from __future__ import annotations
+
+import os
+import time
+from typing import Any, Callable, Optional
+
+from beamtimehero_cli import station
+
+URL_VAR = "BTH_CURRENT_AMP_URL"
+TIMEOUT_S = 5.0          # per HTTP request
+POLL_S = 1.0
+#: The server stops a run after its own max_run_s (300 s at BL4-1); wait a
+#: little longer so its limit, not ours, normally ends a stuck run.
+DEFAULT_WAIT_S = 360.0
+MAX_WAIT_S = 900.0
+
+
+def server_url() -> Optional[str]:
+    url = os.environ.get(URL_VAR) or station.default(URL_VAR)
+    return url.rstrip("/") if url else None
+
+
+def _request(method: str, path: str) -> dict[str, Any]:
+    """{ok, status, body} or {ok: false, error}."""
+    url = server_url()
+    if not url:
+        return {"ok": False, "error": f"{station.station_name()} has no current-amp "
+                                      f"server configured ({URL_VAR})"}
+    import requests
+    try:
+        r = requests.request(method, url + path, timeout=TIMEOUT_S,
+                             proxies={"http": None, "https": None})
+    except requests.RequestException as e:
+        return {"ok": False, "url": url + path, "error": f"{type(e).__name__}: {e}"}
+    try:
+        body = r.json()
+    except ValueError:
+        body = {"text": r.text[:500]}
+    if r.status_code >= 400:
+        detail = body.get("detail") if isinstance(body, dict) else None
+        return {"ok": False, "url": url + path, "status": r.status_code,
+                "error": detail or f"HTTP {r.status_code}"}
+    return {"ok": True, "url": url + path, "status": r.status_code, "body": body}
+
+
+def autozero_status() -> dict[str, Any]:
+    """The server's last (or current) zeroing run. Read-only."""
+    res = _request("GET", "/api/autozero")
+    if not res.get("ok"):
+        return res
+    body = res["body"] or {}
+    return {"ok": True, "enabled": bool(body.get("enabled")),
+            "status": body.get("status")}
+
+
+def _summary(status: dict) -> dict[str, Any]:
+    started, finished = status.get("started"), status.get("finished")
+    return {
+        "state": status.get("state"),
+        "results": status.get("results") or {},
+        "log": [e.get("msg") for e in status.get("log") or []],
+        "duration_s": (round(finished - started, 1)
+                       if started and finished else None),
+    }
+
+
+def zero_offsets(wait_s: float = DEFAULT_WAIT_S,
+                 sleep: Callable[[float], None] = time.sleep,
+                 clock: Callable[[], float] = time.monotonic) -> dict[str, Any]:
+    """Start the server's offset zeroing and wait for it to finish.
+
+    ``ok`` is true only when the run ended ``done``; per-detector outcomes
+    (zeroed / skipped / failed / restored) are in ``results``. If the run
+    is still going after ``wait_s`` it is aborted, which makes the server
+    restore every original offset.
+    """
+    wait_s = max(1.0, min(float(wait_s), MAX_WAIT_S))
+    pre = autozero_status()
+    if not pre.get("ok"):
+        return pre
+    if not pre["enabled"]:
+        return {"ok": False, "error": "offset zeroing is not enabled on the "
+                                      "current-amp server (config/autozero.json)"}
+    start = _request("POST", "/api/autozero/start")
+    if not start.get("ok"):
+        return start
+    deadline = clock() + wait_s
+    status: dict = {}
+    while True:
+        sleep(POLL_S)
+        cur = autozero_status()
+        if cur.get("ok"):
+            status = cur.get("status") or {}
+            if status.get("state") != "running":
+                break
+        if clock() >= deadline:
+            _request("POST", "/api/autozero/abort")
+            return {"ok": False,
+                    "error": f"zeroing still running after {wait_s:g} s; aborted "
+                             "(the server restores the original offsets)",
+                    "last_status": _summary(status) if status else cur}
+    out = _summary(status)
+    out["ok"] = out["state"] == "done"
+    if not out["ok"]:
+        out["error"] = (f"zeroing ended {out['state']!r}; the server restored the "
+                        "original offsets (see log)")
+    return out

src/beamtimehero_cli/station.py

@@ -49,6 +49,8 @@
         "BTH_GASMIXER_URL": "http://127.0.0.1:5006",
         "BTH_XSP32_GUI_URL": "http://192.168.41.6:8032",
         "BTH_MOTOR_ANALYZER_URL": "http://127.0.0.1:5011",
+        # SR570 current-amp server (ssrl-current-amp): zero_amp_offsets
+        "BTH_CURRENT_AMP_URL": "http://192.168.41.6:8570",
         # where the station's apps write their logs (applogs.py)
         "BTH_APP_LOG_ROOTS": (
             "fourone=/usr/local/projects/autonomous_fourone/logs,"

src/beamtimehero_cli/tool_catalog/definitions.py

@@ -7,6 +7,7 @@
 import json
 from pathlib import Path
 
+from beamtimehero_cli import current_amp as _current_amp
 from beamtimehero_cli import epics_ca as _epics_ca
 from beamtimehero_cli.science.exafs import policy as _exafs_policy
 from beamtimehero_cli.science.xrs import policy as _xrs_policy
@@ -330,6 +331,34 @@
             },
         },
     },
+    {
+        "type": "function",
+        "function": {
+            "name": "zero_amp_offsets",
+            "description": (
+                "BL4-1: auto-zero the SR570 current-amplifier input offsets (I0, I1, "
+                "I2, PIPS) with the current-amp server's Zero Offsets routine — each "
+                "detector's offset is set to the smallest ladder step whose cscan "
+                "box ADC reading is just above zero (+2 mV). Takes ~30 s; blocks "
+                "until done and returns per-detector before/after offsets and "
+                "voltages. Zero with the beam off (shutter closed): it cancels "
+                "whatever current is present. Not while a scan, a SPEC count or a "
+                "CHERFD stream is using the cscan box. On a failure or abort the "
+                "server restores every original offset. Rerun measure_darks "
+                "afterwards (the stored dark rates go stale)."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    **_J,
+                    "wait_s": {"type": "number", "default": _current_amp.DEFAULT_WAIT_S,
+                               "exclusiveMinimum": 0, "maximum": _current_amp.MAX_WAIT_S,
+                               "description": "Abort (offsets restored) if not finished by then (s)."},
+                },
+                "required": ["justification"],
+            },
+        },
+    },
     {
         "type": "function",
         "function": {
@@ -3634,7 +3663,7 @@
         "get_anchor", "set_anchor", "tracking",
     ]),
     ("CAT-6 Beam", ["get_beam_size", "get_beam_status", "get_counts", "get_counter", "request_gap_ownership", "capture_sample_image", "get_reference_image",
-                    "measure_darks", "set_check_beam", "get_check_beam"]),
+                    "measure_darks", "zero_amp_offsets", "set_check_beam", "get_check_beam"]),
     ("CAT-7 State", ["get_element", "get_crystal_set", "get_scan_number", "get_current_datafile", "get_plotselected_counter", "abort_current_scan"]),
     ("CAT-8 Orchestration", [
         "request_human_intervention", "post_status_update",

src/beamtimehero_cli/tool_catalog/lineage.py

@@ -1394,6 +1394,41 @@
         "source_detail": "SPEC macros in 41_spec/offset_subtraction.mac and shutter.mac.",
         "depends_on": [],
     },
+    "zero_amp_offsets": {
+        "long_description": (
+            "Starts ssrl-current-amp's Zero Offsets routine (devices/autozero.py "
+            "there) and polls it to the end. The server measures every cscan box "
+            "ADC, probes one offset step per detector for the sign of dV/dI, jumps "
+            "to the predicted zero crossing on the SR570 1-2-5 offset ladder and "
+            "steps until the reading clears target_min_v (+2 mV at BL4-1). Its own "
+            "guards (max 8 offset changes per detector, probe response 0.5-2x the "
+            "nominal gain, readings must follow the prediction, 300 s per run) "
+            "restore a detector's original offset; an error or abort restores all "
+            "of them. Past wait_s this tool aborts the run. Refused where no "
+            "server is configured."
+        ),
+        "python_func": (
+            "audited_call.audited_device_call('zero_amp_offsets', ..., "
+            "run=current_amp.zero_offsets(wait_s))"
+        ),
+        "spec_command": None,
+        # True: SR570 offset writes through the current-amp server, audited
+        # with audited_device_call (justification, safety switch, action log).
+        "mutates": True,
+        "output": (
+            "JSON: {ok, kind, action_id, result: {ok, state, results: {det: {adc, "
+            "sensitivity, offset_before, offset_after, v_before, v_after, status, "
+            "note}}, log[], duration_s}, elapsed_s}"
+        ),
+        "source": "epics_ioc",
+        "source_detail": (
+            "HTTP to the ssrl-current-amp server (BTH_CURRENT_AMP_URL; BL4-1 "
+            "default 192.168.41.6:8570): POST /api/autozero/start, GET "
+            "/api/autozero, POST /api/autozero/abort. The server writes the "
+            "SR570s over its Moxa and reads the ADCs with cscan_acq's CsBox."
+        ),
+        "depends_on": [],
+    },
     "set_check_beam": {
         "long_description": "check_beam_on / check_beam_off (41_spec/check_beam.mac): pause step scans while there is no beam.",
         "python_func": "spec_cmd.call('check_beam_on' | 'check_beam_off', [], justification)",

src/beamtimehero_cli/tool_catalog/tools_core.py

@@ -1185,6 +1185,24 @@ def t_measure_darks(arguments: dict) -> tuple[str, list[str]]:
     return _as_json(res), []
 
 
+def t_zero_amp_offsets(arguments: dict) -> tuple[str, list[str]]:
+    from beamtimehero_cli import current_amp
+    from beamtimehero_cli.audited_call import audited_device_call
+    url = current_amp.server_url()
+    if not url:
+        from beamtimehero_cli import station
+        return _as_json({"ok": False, "error": f"{station.station_name()} has no current-amp "
+                                               f"server configured ({current_amp.URL_VAR})"}), []
+    j = (arguments.get("justification") or "").strip()
+    wait_s = float(arguments.get("wait_s", current_amp.DEFAULT_WAIT_S))
+    res = audited_device_call(
+        "zero_amp_offsets", [f"wait_s={wait_s:g}"], j,
+        lambda: current_amp.zero_offsets(wait_s),
+        wire=f"POST {url}/api/autozero/start",
+    )
+    return _as_json(res), []
+
+
 def t_set_check_beam(arguments: dict) -> tuple[str, list[str]]:
     refused = _bl41_only("set_check_beam")
     if refused:
@@ -2954,6 +2972,7 @@ def t_overlay_chi_spectra(arguments: dict) -> tuple[str, list[str]]:
     "record_provenance": t_record_provenance,
     "get_provenance": t_get_provenance,
     "measure_darks": t_measure_darks,
+    "zero_amp_offsets": t_zero_amp_offsets,
     "set_check_beam": t_set_check_beam,
     "get_check_beam": t_get_check_beam,
     "get_crystal_set": t_get_crystal_set,

tests/test_agent_surface.py

@@ -292,7 +292,7 @@ def test_the_write_filter_is_the_only_thing_that_drops_anything(catalogue):
     build = build_surface(_aligner(), catalogue)
     mutating = sum(1 for t in build.tools if t.mutates)
     assert mutating == 2
-    assert len(build.tools) == len(TOOL_DEFINITIONS) - (42 + 17) + 2
+    assert len(build.tools) == len(TOOL_DEFINITIONS) - (43 + 17) + 2
 
 
 def test_a_branch_carries_its_nested_children(catalogue):
@@ -407,7 +407,7 @@ def leaves(node: dict):
         leaf for leaf in leaves(snapshot_parser(parser))
         if leaf["defaults"].get("_tool_name")
     ]
-    assert len(stamped) == 206 - (42 + 17) + 2
+    assert len(stamped) == 207 - (43 + 17) + 2
     assert all(leaf["defaults"]["_agent_role"] == "blaligner" for leaf in stamped)
 
 
@@ -855,8 +855,8 @@ def test_names_are_not_unique_but_no_shared_name_mutates(catalogue):
     names = Counter(
         (tdef.get("function") or {}).get("name") for tdef in TOOL_DEFINITIONS
     )
-    assert sum(names.values()) == 206
-    assert len(names) == 200
+    assert sum(names.values()) == 207
+    assert len(names) == 201
     duplicated = {name for name, n in names.items() if n > 1}
     assert duplicated == {
         "get_latest_scan", "list_scans", "read_scan",
@@ -866,15 +866,15 @@ def test_names_are_not_unique_but_no_shared_name_mutates(catalogue):
 
     # And the index really is keyed by path, so both of a pair survive.
     paths = {"/".join(p) for p in catalogue.index()}
-    assert len(paths) == 206
+    assert len(paths) == 207
     assert {"spec-file/list_scans", "s3df/list_scans"} <= paths
 
 
 def test_exactly_the_write_branches_mutate(catalogue):
     mutating = {
         "/".join(tool.path) for tool in catalogue.index().values() if tool.mutates
     }
-    assert len(mutating) == 42 + 17
+    assert len(mutating) == 43 + 17
     assert {p.split("/")[0] for p in mutating} == {"spec-write", "cherfd-write"}