feat(aerotech): A3200 controller and motor for the SAXS-TT Aerotech stages

Npaq/A3200 via the ASCII command interface on the A3200 Windows PC:
enable, home (HOMEASYNC), MOVEABS, abort, fault ack, ~STATUS readback.
Simulated controller (sim_a3200), device configs for hardware and
simulation, tests. Commissioning checklist in devices/aerotech/README.md.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM

csaxs_bec/device_configs/sastt_aerotech.yaml

@@ -0,0 +1,91 @@
+############################################################
+# SAXS-TT Aerotech stages: Npaq-B-40B-40B/ULTRA, A3200 software on a Windows PC
+# (FireWire to the Npaq). BEC connects to the A3200 ASCII command interface on that PC.
+# A3200 setup: CommandSetup = 0x00010004 (enabled + Always Send EOS), CommandPort = 8000.
+#
+# TODO commissioning: host name of the A3200 PC, A3200 axis names, velocities, limits.
+# UU (ADRS-150 slipring follower) follows U on the controller and is not a BEC device.
+# Limits [0, 0] = no BEC soft limits; the controller's SoftwareLimitLow/High still apply.
+############################################################
+
+aerox:
+  description: Aerotech X, ALS25020 linear stage
+  deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
+  deviceConfig:
+    axis_Id: X
+    axis_nr: 0
+    host: a3200-saxstt.psi.ch
+    port: 8000
+    velocity: 1.0
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aeroy:
+  description: Aerotech Y, AVSI125 vertical lift
+  deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
+  deviceConfig:
+    axis_Id: Y
+    axis_nr: 1
+    host: a3200-saxstt.psi.ch
+    port: 8000
+    velocity: 0.5
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aeroz:
+  description: Aerotech Z, ATS20010 linear stage
+  deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
+  deviceConfig:
+    axis_Id: Z
+    axis_nr: 2
+    host: a3200-saxstt.psi.ch
+    port: 8000
+    velocity: 1.0
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aerorot:
+  description: Aerotech U, ABRT-200 rotation stage (UU follows on the controller)
+  deviceClass: csaxs_bec.devices.aerotech.a3200.A3200Motor
+  deviceConfig:
+    axis_Id: U
+    axis_nr: 3
+    host: a3200-saxstt.psi.ch
+    port: 8000
+    velocity: 10.0
+    egu: deg
+    tolerance: 0.001
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech

csaxs_bec/device_configs/simulated_omny/simulated_sastt_aerotech.yaml

@@ -0,0 +1,86 @@
+############################################################
+# Simulated SAXS-TT Aerotech stages (sim_a3200), same devices as sastt_aerotech.yaml.
+# Axes start enabled and homed (sim_enabled / sim_homed in deviceConfig to change).
+############################################################
+
+aerox:
+  description: Aerotech X, ALS25020 linear stage
+  deviceClass: csaxs_bec.devices.sim.sim_a3200.SimA3200Motor
+  deviceConfig:
+    axis_Id: X
+    axis_nr: 0
+    host: sim-a3200-saxstt
+    port: 8000
+    velocity: 1.0
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aeroy:
+  description: Aerotech Y, AVSI125 vertical lift
+  deviceClass: csaxs_bec.devices.sim.sim_a3200.SimA3200Motor
+  deviceConfig:
+    axis_Id: Y
+    axis_nr: 1
+    host: sim-a3200-saxstt
+    port: 8000
+    velocity: 0.5
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aeroz:
+  description: Aerotech Z, ATS20010 linear stage
+  deviceClass: csaxs_bec.devices.sim.sim_a3200.SimA3200Motor
+  deviceConfig:
+    axis_Id: Z
+    axis_nr: 2
+    host: sim-a3200-saxstt
+    port: 8000
+    velocity: 1.0
+    egu: mm
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech
+aerorot:
+  description: Aerotech U, ABRT-200 rotation stage (UU follows on the controller)
+  deviceClass: csaxs_bec.devices.sim.sim_a3200.SimA3200Motor
+  deviceConfig:
+    axis_Id: U
+    axis_nr: 3
+    host: sim-a3200-saxstt
+    port: 8000
+    velocity: 10.0
+    egu: deg
+    tolerance: 0.001
+    limits:
+    - 0
+    - 0
+    sign: 1
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
+  deviceTags:
+    - sastt_aerotech

csaxs_bec/devices/aerotech/README.md

@@ -0,0 +1,47 @@
+# Aerotech A3200 (SAXS-TT)
+
+Hardware: Npaq-B-40B-40B/ULTRA drive rack, commanded over FireWire by the A3200 software
+(SMC) on a Windows PC. The Npaq Ethernet port is for Modbus I/O only, so BEC talks to the
+A3200 ASCII command interface on the Windows PC (TCP, default port 8000).
+
+| BEC device | A3200 axis | Stage    | Type                                           |
+|------------|------------|----------|------------------------------------------------|
+| aerox      | X          | ALS25020 | linear                                         |
+| aeroy      | Y          | AVSI125  | vertical lift, 25 mm                           |
+| aeroz      | Z          | ATS20010 | linear                                         |
+| aerorot    | U          | ABRT-200 | rotation                                       |
+| -          | UU         | ADRS-150 | slipring follower of U, handled on the controller |
+
+Configs: `device_configs/sastt_aerotech.yaml` (hardware),
+`device_configs/simulated_omny/simulated_sastt_aerotech.yaml` (simulation, `sim_a3200.py`).
+
+Documentation: A3200 6.04.004 help (`A3200.chm`, topic "ASCII Command Interface"), copy at
+`~/gitea/SAXS-TT_docs` on the workstation.
+
+## Protocol
+
+- one AeroBasic line per command, terminated by `\n`
+- reply: `%` (+ data) success, `!` invalid syntax, `#` task fault (`~LASTERROR` gives the text)
+- `~STATUS (X, ProgramPositionFeedback) (X, AxisStatus) ...` reads several items at once
+- only asynchronous motion commands are used (`MOVEABS`, `HOMEASYNC`, `ABORT`); blocking
+  commands like `HOME` would block the socket until they finish
+- motion commands are never retried; status reads are retried once
+
+## Commissioning checklist
+
+1. A3200 runs on the Windows PC, all axes enable and move in Motion Composer.
+2. Export the parameter file (`.prme`) and keep a copy.
+3. Configuration Manager: `CommandSetup = 0x00010004` (ASCII interface enabled +
+   Always Send EOS), `CommandPort = 8000`, reset the controller.
+4. Windows firewall: allow inbound TCP 8000 from the BEC host.
+5. Check with `nc <host> 8000`, send `~VERSION`, expect `%6.04...`.
+6. Set host, axis names, velocities and limits in `sastt_aerotech.yaml`.
+7. Check `ProgramPositionFeedback` matches the position shown in Motion Composer.
+8. Check how UU follows U (gantry, slave or program) before rotating.
+
+## Next: continuous line scan
+
+Fast axis moving continuously, slow axis stepping between lines, start of each line
+triggered by position with PSO (position synchronized output) on the fast axis:
+`PSOCONTROL`, `PSOWINDOW`/`PSODISTANCE`, `PSOOUTPUT` drive the digital output when the
+armed position is crossed. To be implemented once the hardware is reachable.

csaxs_bec/devices/aerotech/a3200.py

@@ -0,0 +1,570 @@
+"""
+Aerotech A3200 motion controller (Npaq drive rack) via the A3200 ASCII command interface.
+
+The Npaq is commanded over FireWire by the A3200 software (SMC) running on a Windows PC.
+BEC talks to that PC through the ASCII command interface, a line-based TCP protocol:
+
+- a command is an AeroBasic line (e.g. ``MOVEABS X 5 10``) or a server command (``~STATUS ...``)
+- the reply starts with one character: ``%`` success, ``!`` invalid syntax, ``#`` task fault
+- data follows the success character (``%1.234``), the reply is terminated by EOS (``\\n``)
+
+Required controller configuration (Configuration Manager, then reset the controller):
+
+- CommandSetup = 0x00010004 (ASCII interface enabled + "Always Send EOS")
+- CommandPort  = 8000 (default used here)
+- the default command characters: EOS ``\\n``, success ``%``, invalid ``!``, fault ``#``
+
+Blocking AeroBasic commands (HOME, WAIT ...) block the socket until they finish, so only
+asynchronous commands are used for motion (MOVEABS, HOMEASYNC, ABORT).
+"""
+
+import threading
+import time
+
+import numpy as np
+from bec_lib import bec_logger
+from ophyd import Component as Cpt
+from ophyd import Device, PositionerBase, Signal
+from ophyd.status import wait as status_wait
+from ophyd.utils import LimitError, ReadOnlyError
+from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
+from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
+from prettytable import PrettyTable
+
+logger = bec_logger.logger
+
+EOS = "\n"
+ACK_CHAR = "%"
+NAK_CHAR = "!"
+FAULT_CHAR = "#"
+
+# Axis Status bits (A3200 help: Motion Composer Suite > System Diagnostics > Axis Status)
+AXISSTATUS_HOMED = 1 << 0
+AXISSTATUS_PROFILING = 1 << 1
+AXISSTATUS_HOMING = 1 << 4
+AXISSTATUS_ENABLING = 1 << 5
+AXISSTATUS_JOGGING = 1 << 8
+AXISSTATUS_DRIVE_PENDING = 1 << 9
+AXISSTATUS_MOVE_DONE = 1 << 22
+
+AXISSTATUS_BUSY_MASK = (
+    AXISSTATUS_PROFILING
+    | AXISSTATUS_HOMING
+    | AXISSTATUS_ENABLING
+    | AXISSTATUS_JOGGING
+    | AXISSTATUS_DRIVE_PENDING
+)
+
+# Drive Status bits
+DRIVESTATUS_ENABLED = 1 << 0
+DRIVESTATUS_CW_EOT_LIMIT = 1 << 1
+DRIVESTATUS_CCW_EOT_LIMIT = 1 << 2
+DRIVESTATUS_HOME_LIMIT = 1 << 3
+
+# Axis Fault bits
+AXIS_FAULT_NAMES = {
+    0: "PositionError",
+    1: "OverCurrent",
+    2: "CwEOTLimit",
+    3: "CcwEOTLimit",
+    4: "CwSoftLimit",
+    5: "CcwSoftLimit",
+    6: "AmplifierFault",
+    7: "PositionFbk",
+    8: "VelocityFbk",
+    9: "HallFault",
+    10: "MaxVelocity",
+    11: "EstopFault",
+    12: "VelocityError",
+    14: "ProbeFault",
+    15: "ExternalFault",
+    17: "MotorTemp",
+    18: "AmplifierTemp",
+    19: "EncoderFault",
+    20: "CommLost",
+    22: "GantryMisalign",
+    23: "FbkScalingFault",
+    24: "MrkSearchFault",
+    25: "SafeZoneFault",
+    26: "InPosTimeout",
+    27: "VoltageClamp",
+    28: "PowerSupply",
+    29: "MissedInterrupt",
+    30: "Internal",
+}
+
+
+def decode_axis_fault(fault: int) -> list[str]:
+    """Return the names of all fault bits set in an Axis Fault word."""
+    return [name for bit, name in AXIS_FAULT_NAMES.items() if fault & (1 << bit)]
+
+
+class A3200CommunicationError(Exception):
+    pass
+
+
+class A3200Error(Exception):
+    pass
+
+
+class A3200Controller(Controller):
+    """
+    Controller for an Aerotech A3200 system, one instance per (host, port).
+
+    Axes are addressed by their A3200 axis name (e.g. "X", "Y", "U"). The axis_nr given to
+    set_axis is only the slot in the controller's axis list.
+    """
+
+    _axes_per_controller = 32
+    USER_ACCESS = [
+        "describe",
+        "socket_put_and_receive",
+        "command",
+        "get_version",
+        "enable_axis",
+        "disable_axis",
+        "home_axis",
+        "fault_ack",
+        "abort_axis",
+        "abort_all_axes",
+        "axis_status",
+        "print_command_history",
+    ]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if not hasattr(self, "_rx_buffer"):
+            self._rx_buffer = ""
+
+    @threadlocked
+    def socket_put(self, val: str) -> None:
+        self.command_history.append(f"[PUT]: {val}")
+        self.sock.put(f"{val}{EOS}".encode())
+
+    @threadlocked
+    def socket_get(self, timeout: float = 5.0) -> str:
+        """Receive one EOS-terminated reply. The EOS character is stripped."""
+        start = time.time()
+        while EOS not in self._rx_buffer:
+            chunk = self.sock.receive().decode()
+            if chunk:
+                self._rx_buffer += chunk
+                continue
+            if time.time() - start > timeout:
+                partial, self._rx_buffer = self._rx_buffer, ""
+                raise A3200CommunicationError(
+                    f"No complete reply within {timeout} s. Received so far:"
+                    f" {partial!r}. Is 'Always Send EOS' set in CommandSetup?"
+                )
+            time.sleep(0.005)
+        response, self._rx_buffer = self._rx_buffer.split(EOS, 1)
+        self.command_history.append(f"[GET]: {response}")
+        return response
+
+    @threadlocked
+    def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str:
+        self.socket_put(val)
+        return self.socket_get()
+
+    @threadlocked
+    def command(self, cmd: str) -> str:
+        """
+        Send a command and return the reply data (without the success character).
+        Not retried: a motion command must not be sent twice.
+
+        Raises:
+            A3200Error: invalid command ('!') or task fault ('#'); for a fault the
+                controller's ~LASTERROR message is included.
+            A3200CommunicationError: unexpected reply.
+        """
+        reply = self.socket_put_and_receive(cmd)
+        if not reply:
+            raise A3200CommunicationError(f"Empty reply to '{cmd}'.")
+        code, data = reply[0], reply[1:].strip()
+        if code == ACK_CHAR:
+            return data
+        if code == NAK_CHAR:
+            raise A3200Error(f"Controller rejected '{cmd}' as invalid.")
+        if code == FAULT_CHAR:
+            raise A3200Error(f"'{cmd}' failed: {self.last_error()}")
+        raise A3200CommunicationError(f"Unexpected reply to '{cmd}': {reply!r}")
+
+    def last_error(self) -> str:
+        reply = self.socket_put_and_receive("~LASTERROR")
+        return reply[1:].strip() if reply.startswith(ACK_CHAR) else reply
+
+    def get_version(self) -> str:
+        return self.command("~VERSION")
+
+    @retry_once
+    @threadlocked
+    def get_status_items(self, items: list[tuple]) -> list[float]:
+        """
+        Read several status items in one round trip with ~STATUS.
+
+        Args:
+            items: tuples as in the ~STATUS syntax, e.g. [("X", "PositionFeedback")]
+        """
+        query = " ".join(f"({', '.join(str(i) for i in item)})" for item in items)
+        values = self.command(f"~STATUS {query}").split()
+        if len(values) != len(items):
+            raise A3200CommunicationError(
+                f"Expected {len(items)} values for ~STATUS {query}, received {values}"
+            )
+        return [float(v) for v in values]
+
+    def axis_status(self, axis: str) -> dict:
+        """Position, status words and decoded flags of one axis."""
+        position, axis_status, drive_status, axis_fault = self.get_status_items(
+            [
+                (axis, "ProgramPositionFeedback"),
+                (axis, "AxisStatus"),
+                (axis, "DriveStatus"),
+                (axis, "AxisFault"),
+            ]
+        )
+        axis_status, drive_status, axis_fault = (
+            int(axis_status),
+            int(drive_status),
+            int(axis_fault),
+        )
+        return {
+            "position": position,
+            "axis_status": axis_status,
+            "drive_status": drive_status,
+            "axis_fault": axis_fault,
+            "enabled": bool(drive_status & DRIVESTATUS_ENABLED),
+            "homed": bool(axis_status & AXISSTATUS_HOMED),
+            "moving": self._is_moving(axis_status),
+            "faults": decode_axis_fault(axis_fault),
+            "cw_limit": bool(drive_status & DRIVESTATUS_CW_EOT_LIMIT),
+            "ccw_limit": bool(drive_status & DRIVESTATUS_CCW_EOT_LIMIT),
+        }
+
+    @staticmethod
+    def _is_moving(axis_status: int) -> bool:
+        return bool(axis_status & AXISSTATUS_BUSY_MASK) or not axis_status & AXISSTATUS_MOVE_DONE
+
+    def is_axis_moving(self, axis: str) -> bool:
+        (axis_status,) = self.get_status_items([(axis, "AxisStatus")])
+        return self._is_moving(int(axis_status))
+
+    def get_position(self, axis: str) -> float:
+        (position,) = self.get_status_items([(axis, "ProgramPositionFeedback")])
+        return position
+
+    def check_ready_to_move(self, axis: str, require_homed: bool = True) -> None:
+        """Raise A3200Error if the axis is faulted, disabled or (optionally) not homed."""
+        status = self.axis_status(axis)
+        if status["faults"]:
+            raise A3200Error(
+                f"Axis {axis} has faults {status['faults']}. Clear them with fault_ack('{axis}')."
+            )
+        if not status["enabled"]:
+            raise A3200Error(f"Axis {axis} is not enabled. Use enable_axis('{axis}').")
+        if require_homed and not status["homed"]:
+            raise A3200Error(f"Axis {axis} is not homed. Use home_axis('{axis}').")
+
+    def move_absolute(self, axis: str, position: float, velocity: float) -> None:
+        """Start an asynchronous absolute move (MOVEABS returns before the move is done)."""
+        if velocity <= 0:
+            raise ValueError(f"Velocity must be positive, received {velocity}.")
+        self.command(f"MOVEABS {axis} {position:.6f} {velocity:.6f}")
+
+    def enable_axis(self, axis: str) -> None:
+        self.command(f"ENABLE {axis}")
+
+    def disable_axis(self, axis: str) -> None:
+        self.command(f"DISABLE {axis}")
+
+    def home_axis(self, axis: str, wait: bool = True, timeout: float = 300) -> None:
+        """Home an axis with HOMEASYNC (non-blocking on the socket), optionally wait."""
+        self.command(f"HOMEASYNC {axis}")
+        if not wait:
+            return
+        start = time.time()
+        time.sleep(0.1)
+        while self.is_axis_moving(axis):
+            if time.time() - start > timeout:
+                self.abort_axis(axis)
+                raise A3200Error(f"Homing of axis {axis} did not finish within {timeout} s.")
+            time.sleep(0.1)
+        status = self.axis_status(axis)
+        if not status["homed"]:
+            raise A3200Error(f"Homing of axis {axis} failed. Faults: {status['faults']}")
+
+    def fault_ack(self, axis: str) -> None:
+        self.command(f"FAULTACK {axis}")
+
+    def abort_axis(self, axis: str) -> None:
+        self.command(f"ABORT {axis}")
+
+    def abort_all_axes(self) -> None:
+        axes = [ax.axis_Id for ax in self._axis if ax is not None]
+        if axes:
+            self.command(f"ABORT {' '.join(axes)}")
+
+    def describe(self) -> None:
+        t = PrettyTable()
+        t.title = f"{self.__class__.__name__} on {self.sock.host}:{self.sock.port}"
+        t.field_names = ["Axis", "Name", "Enabled", "Homed", "Moving", "Faults", "Position"]
+        for axis in self._axis:
+            if axis is None:
+                continue
+            status = self.axis_status(axis.axis_Id)
+            t.add_row(
+                [
+                    axis.axis_Id,
+                    axis.name,
+                    status["enabled"],
+                    status["homed"],
+                    status["moving"],
+                    ", ".join(status["faults"]) or "-",
+                    f"{status['position'] * axis.sign:.4f}",
+                ]
+            )
+        print(t)
+
+
+class A3200SignalBase(SocketSignal):
+    def __init__(self, signal_name, **kwargs):
+        self.signal_name = signal_name
+        super().__init__(**kwargs)
+        self.controller = self.root.controller if hasattr(self.root, "controller") else None
+
+
+class A3200SignalRO(A3200SignalBase):
+    def __init__(self, signal_name, **kwargs):
+        super().__init__(signal_name, **kwargs)
+        self._metadata["write_access"] = False
+
+    def _socket_set(self, val):
+        raise ReadOnlyError("Read-only signals cannot be set")
+
+
+class A3200ReadbackSignal(A3200SignalRO):
+    @retry_once
+    @threadlocked
+    def _socket_get(self) -> float:
+        return self.controller.get_position(self.parent.axis_Id) * self.parent.sign
+
+    def read(self):
+        self._metadata["timestamp"] = time.time()
+        return super().read()
+
+
+class A3200SetpointSignal(A3200SignalBase):
+    def __init__(self, signal_name, **kwargs):
+        super().__init__(signal_name, **kwargs)
+        self.setpoint = 0.0
+
+    def _socket_get(self) -> float:
+        """Last setpoint sent; it is not read back from the controller."""
+        return self.setpoint
+
+    @threadlocked
+    def _socket_set(self, val: float) -> None:
+        self.setpoint = val
+        self.controller.move_absolute(
+            self.parent.axis_Id, val * self.parent.sign, self.parent.motor_velocity.get()
+        )
+
+
+class A3200MotorIsMoving(A3200SignalRO):
+    @threadlocked
+    def _socket_get(self) -> bool:
+        return self.controller.is_axis_moving(self.parent.axis_Id)
+
+    def get(self):
+        val = super().get()
+        if val is not None:
+            self._run_subs(sub_type=self.SUB_VALUE, value=val, timestamp=time.time())
+        return val
+
+
+class A3200AxisHomed(A3200SignalRO):
+    @threadlocked
+    def _socket_get(self) -> bool:
+        return self.controller.axis_status(self.parent.axis_Id)["homed"]
+
+
+class A3200Motor(Device, PositionerBase):
+    """
+    One axis of an Aerotech A3200 controller.
+
+    Args:
+        axis_Id: A3200 axis name, e.g. "X".
+        axis_nr: slot of this axis in the controller's axis list (unique per controller).
+        velocity: move speed in user units per second (A3200 user units: mm or deg).
+        egu: engineering units shown in BEC.
+        require_homed: refuse moves while the axis is not homed.
+    """
+
+    USER_ACCESS = ["controller", "enable", "disable", "home", "fault_ack", "status"]
+    readback = Cpt(A3200ReadbackSignal, signal_name="readback", kind="hinted")
+    user_setpoint = Cpt(A3200SetpointSignal, signal_name="setpoint")
+    motor_is_moving = Cpt(A3200MotorIsMoving, signal_name="motor_is_moving", kind="normal")
+    homed = Cpt(A3200AxisHomed, signal_name="homed", kind="config")
+    motor_velocity = Cpt(Signal, value=1.0, kind="config")
+    high_limit_travel = Cpt(Signal, value=0, kind="omitted")
+    low_limit_travel = Cpt(Signal, value=0, kind="omitted")
+
+    SUB_READBACK = "readback"
+    SUB_CONNECTION_CHANGE = "connection_change"
+    _default_sub = SUB_READBACK
+
+    def __init__(
+        self,
+        axis_Id,
+        prefix="",
+        *,
+        name,
+        axis_nr=0,
+        kind=None,
+        read_attrs=None,
+        configuration_attrs=None,
+        parent=None,
+        host="localhost",
+        port=8000,
+        limits=None,
+        sign=1,
+        velocity=1.0,
+        egu="mm",
+        tolerance=0.001,
+        require_homed=True,
+        socket_cls=SocketIO,
+        device_manager=None,
+        **kwargs,
+    ):
+        self.controller = A3200Controller(
+            socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
+        )
+        self.axis_Id = axis_Id
+        self.axis_nr = axis_nr
+        self.controller.set_axis(axis=self, axis_nr=axis_nr)
+        self.sign = sign
+        self.tolerance = tolerance
+        self.require_homed = require_homed
+        self._egu = egu
+        self.device_manager = device_manager
+
+        super().__init__(
+            prefix,
+            name=name,
+            kind=kind,
+            read_attrs=read_attrs,
+            configuration_attrs=configuration_attrs,
+            parent=parent,
+            **kwargs,
+        )
+        self.readback.name = self.name
+        self.motor_velocity.put(velocity)
+        self.controller.subscribe(
+            self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
+        )
+        self._update_connection_state()
+
+        if limits is not None:
+            assert len(limits) == 2
+            self.low_limit_travel.put(limits[0])
+            self.high_limit_travel.put(limits[1])
+
+    def wait_for_connection(self, timeout: float = 30.0) -> bool:
+        self.controller.on(timeout=timeout)
+        self.user_setpoint.setpoint = self.readback.get()
+
+    def destroy(self):
+        """Make sure to turn off the controller socket on destroy."""
+        self.controller.off(update_config=False)
+        return super().destroy()
+
+    @property
+    def limits(self):
+        return (self.low_limit_travel.get(), self.high_limit_travel.get())
+
+    @property
+    def low_limit(self):
+        return self.limits[0]
+
+    @property
+    def high_limit(self):
+        return self.limits[1]
+
+    @property
+    def egu(self):
+        return self._egu
+
+    def check_value(self, pos):
+        """Check that the position is within the soft limits"""
+        low_limit, high_limit = self.limits
+        if low_limit < high_limit and not (low_limit <= pos <= high_limit):
+            raise LimitError(f"position={pos} not within limits {self.limits}")
+
+    def _update_connection_state(self, **kwargs):
+        for walk in self.walk_signals():
+            walk.item._metadata["connected"] = self.controller.connected
+
+    @raise_if_disconnected
+    def move(self, position, wait=True, **kwargs):
+        """Move to a position. Returns a MoveStatus that finishes when the axis is done."""
+        # Check before the MoveStatus exists: a failed MoveStatus calls stop() (ABORT) from
+        # a callback thread, which could abort a move or homing started right after.
+        self.check_value(position)
+        self.controller.check_ready_to_move(self.axis_Id, require_homed=self.require_homed)
+        self._started_moving = False
+        timeout = kwargs.pop("timeout", 100)
+        status = super().move(position, timeout=timeout, **kwargs)
+        try:
+            self.user_setpoint.put(position, wait=False)
+        except Exception:
+            self._done_moving(success=False)
+            raise
+
+        def move_and_finish():
+            # MOVEABS is asynchronous; give the controller one cycle to set the busy bits
+            time.sleep(0.05)
+            while self.motor_is_moving.get():
+                val = self.readback.read()
+                self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
+                time.sleep(0.1)
+            val = self.readback.read()
+            self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
+            success = np.isclose(val[self.name]["value"], position, atol=self.tolerance)
+            if not success:
+                faults = self.controller.axis_status(self.axis_Id)["faults"]
+                logger.warning(
+                    f"{self.name} stopped at {val[self.name]['value']} instead of {position}."
+                    f" Faults: {faults}"
+                )
+            self._done_moving(success=success)
+
+        threading.Thread(target=move_and_finish, daemon=True).start()
+        try:
+            if wait:
+                status_wait(status)
+        except KeyboardInterrupt:
+            self.stop()
+            raise
+
+        return status
+
+    def stop(self, *, success=False):
+        self.controller.abort_axis(self.axis_Id)
+        return super().stop(success=success)
+
+    def enable(self) -> None:
+        self.controller.enable_axis(self.axis_Id)
+
+    def disable(self) -> None:
+        self.controller.disable_axis(self.axis_Id)
+
+    def home(self, wait: bool = True) -> None:
+        self.controller.home_axis(self.axis_Id, wait=wait)
+        val = self.readback.read()
+        self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
+
+    def fault_ack(self) -> None:
+        self.controller.fault_ack(self.axis_Id)
+
+    def status(self) -> dict:
+        return self.controller.axis_status(self.axis_Id)

csaxs_bec/devices/sim/sim_a3200.py

@@ -0,0 +1,298 @@
+"""
+Simulated Aerotech A3200 controller (ASCII command interface).
+
+Implements the command subset used by `A3200Controller`: ~VERSION, ~LASTERROR, ~STATUS,
+ENABLE, DISABLE, MOVEABS, MOVEINC, ABORT, HOMEASYNC, FAULTACK, ACKNOWLEDGEALL.
+Replies follow the real interface with "Always Send EOS": "%<data>\\n", "!\\n" or "#\\n".
+
+Simulated behavior:
+- constant-velocity motion with linear position interpolation (no ramps)
+- HOMEASYNC moves the axis to 0 at the home speed and sets the Homed bit
+- moving a disabled axis is a task fault ("#"), like on the real controller
+- faults can be injected via `SimA3200State.axis(name).fault`, FAULTACK clears them
+"""
+
+from __future__ import annotations
+
+import re
+import threading
+import time
+
+from bec_lib.logger import bec_logger
+
+from csaxs_bec.devices.aerotech.a3200 import (
+    AXISSTATUS_HOMED,
+    AXISSTATUS_HOMING,
+    AXISSTATUS_JOGGING,
+    AXISSTATUS_MOVE_DONE,
+    DRIVESTATUS_ENABLED,
+    A3200Motor,
+)
+from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry
+
+logger = bec_logger.logger
+
+_STATUS_ITEM_RE = re.compile(r"\(([^)]*)\)")
+
+
+class _SimA3200Axis:
+    """State of one simulated A3200 axis (positions in user units)."""
+
+    def __init__(self, name: str):
+        self.name = name
+        self.enabled = False
+        self.homed = False
+        self.homing = False
+        self.fault = 0
+        self.home_velocity = 10.0
+        self.pos = 0.0
+        self._move_start = 0.0
+        self._move_target = 0.0
+        self._move_t0 = 0.0
+        self._move_t1 = 0.0
+
+    def position(self) -> float:
+        now = time.time()
+        if now >= self._move_t1:
+            return self.pos
+        frac = (now - self._move_t0) / (self._move_t1 - self._move_t0)
+        return self._move_start + frac * (self._move_target - self._move_start)
+
+    def is_moving(self) -> bool:
+        moving = time.time() < self._move_t1
+        if not moving and self.homing:
+            self.homing = False
+            self.homed = True
+        return moving
+
+    def start_move(self, target: float, velocity: float):
+        current = self.position()
+        self._move_start = current
+        self._move_target = target
+        self._move_t0 = time.time()
+        self._move_t1 = self._move_t0 + abs(target - current) / abs(velocity)
+        self.pos = target
+
+    def abort(self):
+        self.pos = self.position()
+        self._move_t1 = 0.0
+        self.homing = False
+
+    def axis_status(self) -> int:
+        moving = self.is_moving()  # first: finishes a homing move
+        status = 0
+        if self.homed:
+            status |= AXISSTATUS_HOMED
+        if self.homing:
+            status |= AXISSTATUS_HOMING
+        if moving:
+            status |= AXISSTATUS_JOGGING
+        else:
+            status |= AXISSTATUS_MOVE_DONE
+        return status
+
+    def drive_status(self) -> int:
+        return DRIVESTATUS_ENABLED if self.enabled else 0
+
+
+class SimA3200State:
+    """Shared state of one simulated A3200 controller."""
+
+    VERSION = "6.4.4.16"
+
+    def __init__(self, host, port):
+        self.host = host
+        self.port = port
+        self._axes = {}
+        self._lock = threading.RLock()
+
+    def axis(self, name: str) -> _SimA3200Axis:
+        with self._lock:
+            key = name.upper()
+            if key not in self._axes:
+                self._axes[key] = _SimA3200Axis(key)
+            return self._axes[key]
+
+    def configure_axis(
+        self, name, initial_position=None, enabled=None, homed=None, home_velocity=None
+    ):
+        ax = self.axis(name)
+        if initial_position is not None:
+            ax.pos = float(initial_position)
+        if enabled is not None:
+            ax.enabled = bool(enabled)
+        if homed is not None:
+            ax.homed = bool(homed)
+        if home_velocity is not None:
+            ax.home_velocity = float(home_velocity)
+
+
+class _TaskFault(Exception):
+    pass
+
+
+class SimA3200Socket(SimSocketBase):
+    state_cls = SimA3200State
+
+    def __init__(self, host, port, socket_timeout: int = 2):
+        super().__init__(host, port, socket_timeout)
+        self._last_error = "No Error"
+
+    def handle_command(self, line: str):
+        try:
+            data = self._dispatch(line)
+        except _TaskFault as exc:
+            self._last_error = str(exc)
+            return "#\n"
+        except (ValueError, IndexError):
+            self._last_error = f"Invalid command: {line}"
+            return "!\n"
+        if data is None:
+            return "!\n"
+        return f"%{data}\n"
+
+    def _dispatch(self, line: str):
+        state = self.state
+        words = line.split()
+        cmd = words[0].upper()
+        args = words[1:]
+
+        if cmd == "~VERSION":
+            return state.VERSION
+        if cmd == "~LASTERROR":
+            return self._last_error
+        if cmd == "~STATUS":
+            return " ".join(self._status_item(item) for item in _STATUS_ITEM_RE.findall(line))
+        if cmd == "ENABLE":
+            for name in self._require_axes(args):
+                state.axis(name).enabled = True
+            return ""
+        if cmd == "DISABLE":
+            for name in self._require_axes(args):
+                ax = state.axis(name)
+                ax.abort()
+                ax.enabled = False
+            return ""
+        if cmd in ("MOVEABS", "MOVEINC"):
+            name, value, velocity = args[0], float(args[1]), float(args[2])
+            if velocity == 0:
+                raise _TaskFault("Speed of zero is not allowed.")
+            ax = self._ready_axis(name)
+            target = value if cmd == "MOVEABS" else ax.position() + value
+            ax.start_move(target, velocity)
+            return ""
+        if cmd == "HOMEASYNC":
+            for name in self._require_axes(args):
+                ax = self._ready_axis(name)
+                ax.homing = True
+                ax.homed = False
+                ax.start_move(0.0, ax.home_velocity)
+            return ""
+        if cmd == "ABORT":
+            for name in self._require_axes(args):
+                state.axis(name).abort()
+            return ""
+        if cmd == "FAULTACK":
+            for name in self._require_axes(args):
+                state.axis(name).fault = 0
+            return ""
+        if cmd == "ACKNOWLEDGEALL":
+            for ax in state._axes.values():
+                ax.fault = 0
+            return ""
+        logger.warning(f"[sim a3200] {self.host}:{self.port} unhandled command '{line}'")
+        return None
+
+    @staticmethod
+    def _require_axes(args):
+        if not args:
+            raise ValueError("No axis given.")
+        return args
+
+    def _ready_axis(self, name: str) -> _SimA3200Axis:
+        ax = self.state.axis(name)
+        if not ax.enabled:
+            raise _TaskFault("The axis is not enabled.")
+        if ax.fault:
+            raise _TaskFault("The axis has a fault.")
+        return ax
+
+    def _status_item(self, item: str) -> str:
+        parts = [p.strip() for p in item.split(",")]
+        if len(parts) < 2:
+            raise ValueError(f"Unsupported status item ({item})")
+        ax = self.state.axis(parts[0])
+        data_item = parts[1]
+        if data_item in (
+            "PositionFeedback",
+            "ProgramPositionFeedback",
+            "PositionCommand",
+            "ProgramPositionCommand",
+        ):
+            return f"{ax.position():.6f}"
+        if data_item == "AxisStatus":
+            return str(ax.axis_status())
+        if data_item == "DriveStatus":
+            return str(ax.drive_status())
+        if data_item == "AxisFault":
+            return str(ax.fault)
+        raise ValueError(f"Unsupported status item ({item})")
+
+
+class SimA3200Motor(A3200Motor):
+    """A3200Motor connected to the simulated A3200 controller.
+
+    Note: the signature must list all deviceConfig keys explicitly. The BEC device
+    server only passes config keys that appear in inspect.signature(dev_cls) and only
+    injects device_manager if it is an explicit parameter.
+    """
+
+    def __init__(
+        self,
+        axis_Id,
+        prefix="",
+        *,
+        name,
+        axis_nr=0,
+        host="localhost",
+        port=8000,
+        limits=None,
+        sign=1,
+        velocity=1.0,
+        egu="mm",
+        tolerance=0.001,
+        require_homed=True,
+        device_manager=None,
+        sim_initial_position=None,
+        sim_enabled=True,
+        sim_homed=True,
+        sim_home_velocity=None,
+        **kwargs,
+    ):
+        super().__init__(
+            axis_Id,
+            prefix,
+            name=name,
+            axis_nr=axis_nr,
+            host=host,
+            port=port,
+            limits=limits,
+            sign=sign,
+            velocity=velocity,
+            egu=egu,
+            tolerance=tolerance,
+            require_homed=require_homed,
+            socket_cls=SimA3200Socket,
+            device_manager=device_manager,
+            **kwargs,
+        )
+        state = SimStateRegistry.get(SimA3200State, host, port)
+        state.configure_axis(
+            axis_Id,
+            initial_position=(
+                None if sim_initial_position is None else sim_initial_position * sign
+            ),
+            enabled=sim_enabled,
+            homed=sim_homed,
+            home_velocity=sim_home_velocity,
+        )

tests/tests_devices/test_a3200.py

@@ -0,0 +1,207 @@
+import pytest
+from bec_server.device_server.tests.utils import DMMock
+from ophyd.utils import LimitError
+from ophyd_devices.tests.utils import SocketMock
+
+from csaxs_bec.devices.aerotech.a3200 import (
+    A3200CommunicationError,
+    A3200Controller,
+    A3200Error,
+    A3200Motor,
+    decode_axis_fault,
+)
+from csaxs_bec.devices.sim.sim_a3200 import SimA3200Motor, SimA3200State
+from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
+
+_HOST = "test-a3200.psi.ch"
+_PORT = 8000
+
+
+@pytest.fixture(autouse=True)
+def _reset_controllers():
+    A3200Controller._reset_controller()
+    SimStateRegistry.reset()
+    yield
+    SimStateRegistry.reset()
+    A3200Controller._reset_controller()
+
+
+@pytest.fixture
+def mock_motor():
+    motor = A3200Motor(
+        "X", name="aerox", host=_HOST, port=_PORT, socket_cls=SocketMock, device_manager=DMMock()
+    )
+    motor.controller.on()
+    motor.controller.sock.flush_buffer()
+    yield motor
+    motor.controller.off()
+
+
+def _sim_motor(axis_Id="X", axis_nr=0, **kwargs):
+    kwargs.setdefault("velocity", 1000.0)
+    kwargs.setdefault("sim_home_velocity", 1000.0)
+    motor = SimA3200Motor(
+        axis_Id,
+        name=f"aero{axis_Id.lower()}",
+        axis_nr=axis_nr,
+        host=_HOST,
+        port=_PORT,
+        device_manager=DMMock(),
+        **kwargs,
+    )
+    motor.wait_for_connection()
+    return motor
+
+
+# --- protocol (SocketMock) ---------------------------------------------------
+
+
+def test_command_returns_data_after_ack(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"%6.04.004.0016\n"]
+    assert mock_motor.controller.get_version() == "6.04.004.0016"
+    assert mock_motor.controller.sock.buffer_put == [b"~VERSION\n"]
+
+
+def test_reply_split_over_several_receives(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"%1.2", b"5 4194", b"305\n"]
+    assert mock_motor.controller.get_status_items([("X", "A"), ("X", "B")]) == [1.25, 4194305.0]
+
+
+def test_two_replies_in_one_receive_are_kept_apart(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"%\n%7.5\n"]
+    mock_motor.controller.enable_axis("X")
+    assert mock_motor.controller.get_position("X") == 7.5
+
+
+def test_fault_reply_raises_with_last_error(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"#\n", b"%The axis is not enabled.\n"]
+    with pytest.raises(A3200Error, match="The axis is not enabled."):
+        mock_motor.controller.command("HOME X")
+    assert mock_motor.controller.sock.buffer_put == [b"HOME X\n", b"~LASTERROR\n"]
+
+
+def test_invalid_reply_raises(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"!\n"]
+    with pytest.raises(A3200Error, match="invalid"):
+        mock_motor.controller.command("ENABLE")
+
+
+def test_motion_command_is_not_retried(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"#\n", b"%fault\n", b"%\n", b"%\n"]
+    with pytest.raises(A3200Error):
+        mock_motor.controller.move_absolute("X", 1.0, 2.0)
+    assert mock_motor.controller.sock.buffer_put == [
+        b"MOVEABS X 1.000000 2.000000\n",
+        b"~LASTERROR\n",
+    ]
+
+
+def test_missing_eos_times_out(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"%"]
+    with pytest.raises(A3200CommunicationError, match="Always Send EOS"):
+        mock_motor.controller.socket_get(timeout=0.05)
+
+
+def test_status_query_format(mock_motor):
+    mock_motor.controller.sock.buffer_recv = [b"%1.5 4194305 1 0\n"]
+    status = mock_motor.controller.axis_status("X")
+    assert mock_motor.controller.sock.buffer_put == [
+        b"~STATUS (X, ProgramPositionFeedback) (X, AxisStatus) (X, DriveStatus) (X, AxisFault)\n"
+    ]
+    assert status["position"] == 1.5
+    assert status["enabled"] and status["homed"]
+    assert not status["moving"]
+    assert status["faults"] == []
+
+
+def test_decode_axis_fault():
+    assert decode_axis_fault(0) == []
+    assert decode_axis_fault((1 << 0) | (1 << 2) | (1 << 11)) == [
+        "PositionError",
+        "CwEOTLimit",
+        "EstopFault",
+    ]
+
+
+# --- simulated controller ----------------------------------------------------
+
+
+def test_sim_move_and_readback():
+    motor = _sim_motor(sim_initial_position=1.0)
+    assert motor.readback.get() == pytest.approx(1.0)
+    motor.move(5.0, wait=True)
+    assert motor.readback.get() == pytest.approx(5.0)
+    assert not motor.motor_is_moving.get()
+
+
+def test_sim_sign_is_applied():
+    motor = _sim_motor(sign=-1, sim_initial_position=2.0)
+    assert motor.readback.get() == pytest.approx(2.0)
+    motor.move(-3.0, wait=True)
+    assert motor.readback.get() == pytest.approx(-3.0)
+    state = SimStateRegistry.get(SimA3200State, _HOST, _PORT)
+    assert state.axis("X").position() == pytest.approx(3.0)
+
+
+def test_sim_move_refused_when_disabled():
+    motor = _sim_motor(sim_enabled=False)
+    with pytest.raises(A3200Error, match="not enabled"):
+        motor.move(1.0, wait=True)
+    motor.enable()
+    motor.move(1.0, wait=True)
+    assert motor.readback.get() == pytest.approx(1.0)
+
+
+def test_sim_move_refused_when_not_homed():
+    motor = _sim_motor(sim_homed=False, sim_initial_position=3.0)
+    with pytest.raises(A3200Error, match="not homed"):
+        motor.move(1.0, wait=True)
+    motor.home()
+    assert "[PUT]: ABORT X" not in motor.controller.command_history
+    assert motor.status()["homed"]
+    assert motor.readback.get() == pytest.approx(0.0)
+    motor.move(1.0, wait=True)
+    assert motor.readback.get() == pytest.approx(1.0)
+
+
+def test_sim_unhomed_move_allowed_if_configured():
+    motor = _sim_motor(sim_homed=False, require_homed=False)
+    motor.move(2.0, wait=True)
+    assert motor.readback.get() == pytest.approx(2.0)
+
+
+def test_sim_fault_blocks_move_until_acknowledged():
+    motor = _sim_motor()
+    SimStateRegistry.get(SimA3200State, _HOST, _PORT).axis("X").fault = 1 << 2
+    with pytest.raises(A3200Error, match="CwEOTLimit"):
+        motor.move(1.0, wait=True)
+    motor.fault_ack()
+    motor.move(1.0, wait=True)
+    assert motor.readback.get() == pytest.approx(1.0)
+
+
+def test_sim_soft_limits():
+    motor = _sim_motor(limits=[-1, 1])
+    with pytest.raises(LimitError):
+        motor.move(2.0, wait=True)
+
+
+def test_sim_stop_aborts_motion():
+    motor = _sim_motor(velocity=1.0)
+    status = motor.move(10.0, wait=False)
+    motor.stop()
+    with pytest.raises(Exception):
+        status.wait(timeout=2)
+    assert motor.readback.get() < 1.0
+    assert not motor.motor_is_moving.get()
+
+
+def test_sim_two_axes_share_one_controller():
+    x = _sim_motor("X", axis_nr=0)
+    y = _sim_motor("Y", axis_nr=1)
+    assert x.controller is y.controller
+    x.move(1.0, wait=True)
+    y.move(2.0, wait=True)
+    assert x.readback.get() == pytest.approx(1.0)
+    assert y.readback.get() == pytest.approx(2.0)
+    x.controller.describe()