Example 2: Testing fastcs-gatan over EPICS with aioca

aioca is a Python (asyncio) EPICS Channel Access client: caget, caput and camonitor as Python functions, so it can drive fastcs-gatan’s PVs where the EPICS command-line tools are not available. It is not a fastcs-gatan dependency; install it with uv pip install aioca. PV names are explained in Example 1.

Setup

Start the mock server and the driver with - epicsca: {} in the transport: list of gatan.yaml, then point the client at the local IOC:

export EPICS_CA_ADDR_LIST=127.0.0.1 EPICS_CA_AUTO_ADDR_LIST=NO

Interactive

python -m asyncio gives a Python prompt where await works directly:

>>> from aioca import caget, caput, camonitor, DBR_CHAR_STR, FORMAT_CTRL, FORMAT_TIME
>>> P = "GATAN:Acquisition:"
>>> await caget("GATAN:DmVersion")
50302

The main calls

# Numbers: write the setting PV, read back the _RBV PV
await caput(P + "DoseFracNumFrames", 20, wait=True)   # wait=True: wait until processed
await caget(P + "DoseFracNumFrames_RBV")               # 20

# Text PVs are character arrays: use DBR_CHAR_STR both ways (like caput -S)
await caput(P + "SaveDir", "/tmp/frames", datatype=DBR_CHAR_STR, wait=True)
await caget(P + "SavedPath", datatype=DBR_CHAR_STR)

# Choice PVs: the value is an index; FORMAT_CTRL adds the option names
st = await caget(P + "DoseFracState", format=FORMAT_CTRL)
st.enums[int(st)]                                      # 'DONE'

# Alarm status and timestamp: FORMAT_TIME
cmd = await caget(P + "AcquireDoseFractionated", format=FORMAT_TIME)
cmd.severity, cmd.timestamp                            # 2 = MAJOR alarm (last command failed)

# Several PVs at once: pass a list, get a list back
await caget([P + "FramesSaved", P + "SaveError"])

# Commands: write 1 to the command PV
await caput(P + "AcquireDoseFractionated", 1, wait=True)

# Watch changes: the callback runs on every update; close when finished
m = camonitor(P + "DoseFracState", lambda v: print(v.enums[int(v)]), format=FORMAT_CTRL)
m.close()

A command PV is never reset to 0 after a write, but every write of 1 runs the command again. Watch the status PVs (DoseFracBusy, DoseFracState), not the command PV.

A dose-fractionated exposure on the Talos K3

examples/dose_fractionation_aioca.py is the EPICS version of examples/dose_fractionation.py. Its defaults are the parameters of GatanDetectorClient’s session_runner.sh step 26, which worked on the real K3 (counting mode, binned frames, 5760 × 4092, gain-normalised, 0.1 s at 0.013316 s per frame = 8 frames, saved to X:\DoseFractions\kwd30084 as fmt_mrc).

python -m fastcs_gatan run Talos_K3_ca.yaml          # terminal 1: driver with EPICS CA
export EPICS_CA_ADDR_LIST=127.0.0.1 EPICS_CA_AUTO_ADDR_LIST=NO
python examples/dose_fractionation_aioca.py          # terminal 2

The script prints the settings and asks for ACQUIRE (--yes skips this). It then writes the setting PVs, runs the Camera:Select and Acquisition:ApplyReadMode commands, starts Acquisition:AcquireDoseFractionated, and watches Acquisition:DoseFracState with camonitor until it reports DONE or FAILED. A command that fails (its PV goes into MAJOR alarm) aborts the script before the exposure starts.

On a recording mock server it sends the plugin the same requests as step 26, byte for byte, from SetCurrentCamera to GetFileSaveResult; it then also sends SetK2Parameters2 to switch frame saving off again.

A single exposure on the Talos K3

examples/single_exposure_aioca.py takes one exposure and has the driver save the transferred frame as an MRC file on the driver’s machine. Its defaults are the parameters of GatanDetectorClient’s session_runner.sh step 6, the first frame that worked on the real K3 (K3 linear, scaling 0.25, gain-normalised, 1024 × 1024, 0.1 s, beam-blanked).

python -m fastcs_gatan run Talos_K3_ca.yaml          # terminal 1: driver with EPICS CA
export EPICS_CA_ADDR_LIST=127.0.0.1 EPICS_CA_AUTO_ADDR_LIST=NO
python examples/single_exposure_aioca.py             # terminal 2

The script asks for ACQUIRE (--yes skips this), writes the setting PVs and runs Camera:Select, Acquisition:ApplyReadMode and Acquisition:ApplyK2Parameters. It then writes 1 to Acquisition:AcquireImage and waits for the put to complete, which happens once the frame is transferred and saved (up to --timeout, default 120 s).

The frame goes to <LocalSaveDir>/<LocalSaveRoot>_NNNN.mrc (default frames/single_0001.mrc, then _0002, …; never overwritten). The driver writes the bytes from the socket unchanged after an MRC header, marked as 16-bit signed like GatanDetectorClient’s MRC files. The script makes --save-dir absolute, since a relative path would be resolved in the driver’s working directory, and prints Acquisition:LocalSavedPath at the end.

On a recording mock server it sends the plugin the same requests as step 6, byte for byte: SetCurrentCamera, SetReadMode, SetK2Parameters2, GetAcquiredImage.

A pytest system test

This test starts the mock server and the driver as subprocesses and talks to them only through PVs. It checks the status PVs, runs a full dose-fractionated exposure (watched with camonitor), and checks that a failing command puts the command PV into MAJOR alarm. Save it as e.g. tests/system/test_ca_system.py.

"""System test: fastcs-gatan over EPICS CA, against the mock GatanSocket server."""

import asyncio
import os
import subprocess
import sys
import time
from pathlib import Path

import pytest
import pytest_asyncio
from aioca import (
    DBR_CHAR_STR,
    FORMAT_CTRL,
    FORMAT_TIME,
    caget,
    camonitor,
    caput,
    purge_channel_caches,
)

MOCK_DIR = "/workspaces/GatanDetectorClient-0.1.0"
P = "GATAN:Acquisition:"

CONFIG = """\
controllers:
  - id: GATAN
    type: fastcs_gatan.GatanController
    ip: 127.0.0.1
    port: {port}
transport:
  - epicsca: {{}}
"""


@pytest.fixture(scope="module")
def ioc(tmp_path_factory):
    """Start the mock server and the driver (as an EPICS IOC) in subprocesses."""
    os.environ["EPICS_CA_ADDR_LIST"] = "127.0.0.1"
    os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "NO"
    port = 48891  # not 48890, so it doesn't clash with a mock you started by hand
    config = tmp_path_factory.mktemp("cfg") / "gatan.yaml"
    config.write_text(CONFIG.format(port=port))

    mock = subprocess.Popen(
        [sys.executable, "-m", "gatan_client.mock_server", "--k3", "--port", str(port)],
        cwd=MOCK_DIR,
    )
    time.sleep(1)
    driver = subprocess.Popen(
        [sys.executable, "-m", "fastcs_gatan", "run", str(config)],
        stdin=subprocess.PIPE,  # keeps the IPython shell waiting instead of exiting
    )
    try:
        yield
    finally:
        driver.kill()
        mock.kill()
        assert driver.stdin is not None
        driver.stdin.close()
        driver.wait()
        mock.wait()


@pytest_asyncio.fixture(autouse=True)
async def close_ca_channels():
    """aioca caches channels per event loop, and each test gets its own loop.
    Drop them while this test's loop is still running, or they fire callbacks
    into a closed loop when the IOC stops."""
    yield
    purge_channel_caches()


async def wait_for_ioc(timeout: float = 20) -> None:
    deadline = time.monotonic() + timeout
    while True:
        try:
            await caget("GATAN:DmVersion", timeout=1)
            return
        except Exception:
            if time.monotonic() > deadline:
                raise
            await asyncio.sleep(0.5)


@pytest.mark.asyncio
async def test_status_pvs(ioc):
    await wait_for_ioc()
    assert await caget("GATAN:DmVersion") == 50302  # the mock's DM version
    assert await caget("GATAN:Camera:NumCameras") > 0


@pytest.mark.asyncio
async def test_dose_fractionated_exposure(ioc, tmp_path: Path):
    await wait_for_ioc()
    await caput(P + "ReadMode", 3, wait=True)  # K3 linear: mock allows small frames
    await caput(P + "Width", 512, wait=True)
    await caput(P + "Height", 512, wait=True)
    await caput(P + "SaveDir", str(tmp_path), datatype=DBR_CHAR_STR, wait=True)
    await caput(P + "SaveRootName", "movie", datatype=DBR_CHAR_STR, wait=True)
    await caput(P + "DoseFracNumFrames", 10, wait=True)
    await caput(P + "DoseFracExposure", 0.5, wait=True)
    assert await caget(P + "DoseFracNumFrames_RBV") == 10

    # Watch the state PV, then start the exposure and wait until it reports DONE.
    states: list[str] = []
    done = asyncio.Event()

    def on_state(value):
        name = value.enums[int(value)]
        states.append(name)
        if name in ("DONE", "FAILED"):
            done.set()

    monitor = camonitor(P + "DoseFracState", on_state, format=FORMAT_CTRL)
    try:
        await caput(P + "AcquireDoseFractionated", 1, wait=True)
        await asyncio.wait_for(done.wait(), timeout=10)
    finally:
        monitor.close()

    message = await caget(P + "DoseFracMessage", datatype=DBR_CHAR_STR)
    assert states[-1] == "DONE", message
    assert await caget(P + "DoseFracBusy") == 0
    assert await caget(P + "FramesSaved") == 10
    assert await caget(P + "SaveError") == 0
    saved = await caget(P + "SavedPath", datatype=DBR_CHAR_STR)
    assert saved.endswith("movie.mrc")


@pytest.mark.asyncio
async def test_failing_command_raises_alarm(ioc):
    await wait_for_ioc()
    await caput(P + "SaveDir", "", datatype=DBR_CHAR_STR, wait=True)  # invalid
    await caput(P + "AcquireDoseFractionated", 1, wait=True)
    cmd = await caget(P + "AcquireDoseFractionated", format=FORMAT_TIME)
    assert cmd.severity == 2  # MAJOR alarm: the command raised

Things to know, all caught by the repo’s filterwarnings = "error" pytest setting:

• Drop aioca’s connections after each test. aioca caches connections per event loop, and pytest-asyncio gives each test its own loop; when the IOC stops, cached connections fire callbacks into closed loops (“Event loop is closed”). purge_channel_caches() in an async fixture fixes this.

• Close the driver’s stdin pipe after the process stops, or pytest reports a ResourceWarning.

• Start the driver with stdin=PIPE. Otherwise its IPython shell exits at once and takes the IOC with it.

• Use a different mock port (48891 here), so the test does not clash with a mock server started by hand on 48890.