epics-bridge 2.1.0

A generic bridge between EPICS IOCs and Python logic.

EPICS Bridge

EPICS Bridge is a high-availability Python framework designed for implementing a robust EPICS-Python interface. It provides a structured environment for bridging external control logic with the EPICS control system, emphasizing synchronous execution, fault tolerance, and strict process monitoring.

This library addresses the common reliability challenges like preventing silent stalls ("zombie processes") and handling network IO failures deterministically.

Documentation

Comprehensive project documentation lives in docs/README.md.

System Architecture

The core of epics-bridge relies on a Twin-Thread Architecture that decouples the control logic from the monitoring signal.

1. Synchronous Control Loop (Main Thread)

The primary thread executes the user-defined logic in a strict, synchronous cycle:

1. Trigger: Waits for an input event or timer.
2. Run Task: Executes user-defined task
3. Acknowledge: Updates the task status and completes the handshake.

2. Isolated Heartbeat Monitor (Daemon Thread)

A separate, isolated thread acts as an internal watchdog. It monitors the activity timestamp of the Main Thread.

• Operational: Pulses the Heartbeat PV as long as the Main Thread is active.
• Stalled (Zombie Protection): If the Main Thread hangs (e.g., infinite loop, deadlocked IO) for longer than the defined tolerance, the Heartbeat thread ceases pulsing immediately. This alerts external watchdogs (e.g., the IOC or alarm handler) that the process is unresponsive.

3. Automatic Recovery ("Suicide Pact")

To support containerized environments (Docker, Kubernetes) or systemd supervisors, the daemon implements a fail-fast mechanism. If network connectivity is lost or IO errors persist beyond a configurable threshold (max_stuck_cycles), the watchdog performs a hard-kill of the process (os._exit(1)). This allows the external supervisor to perform a clean restart of the service.

4. Logger

Output important messages in the daemon shell to a configured log file.

Installation

# Install the package
pip install .

# Install test dependencies
pip install -r requirements-test.txt

Conda environment (recommended for integration tests)

Integration tests run a real IOC and require EPICS tooling. A working reference environment is provided in environment.yml.

conda env create -f environment.yml
conda activate epics-bridge
pip install -e .

Project Structure

• epics_bridge.daemon Main control loop, heartbeat logic, and failure handling

• epics_bridge.io Synchronous P4P client wrapper with strict error handling

• epics_bridge.base_pv_interface PV template definitions and prefix validation

• epics_bridge.utils Utilities for converting P4P data into native Python types

---

Quick Start

1. EPICS Interface

There should be a standard epics db to handle the basic functionalities of the daemon and any amount of specialized dbs to fulfill the intended functionality.

The standard db should always be loaded by the IOC that interfaces with the daemon. These are its contents:

record(bo, "$(P)Trigger") {
    field(DESC, "Start Task")
    field(ZNAM, "Idle")
    field(ONAM, "Run")
}

record(bi, "$(P)Busy") {
    field(DESC, "Task Running Status")
    field(ZNAM, "Idle")
    field(ONAM, "Busy")
}

record(bi, "$(P)Heartbeat") {
    field(DESC, "Daemon Heartbeat")
}

record(mbbi, "$(P)TaskStatus") {
    field(DESC, "Last Cycle Result")
    field(DTYP, "Raw Soft Channel")

    # State 0: Success (Green)
    field(ZRVL, "0")
    field(ZRST, "Success")
    field(ZRSV, "NO_ALARM")

    # State 1: Logic Failure (Yellow - e.g. Interlock)
    field(ONVL, "1")
    field(ONST, "Task Fail")
    field(ONSV, "MINOR")

    # State 2: EPICS IO Failure (Yellow - e.g. PV Read/Write Error)
    field(TWVL, "2")
    field(TWST, "IO Failure")
    field(TWSV, "MINOR")

    # State 3: Exception (Red - Software/Hardware Crash)
    field(THVL, "3")
    field(THST, "Code Crash")
    field(THSV, "MAJOR")
}

record(ai, "$(P)TaskDuration") {
    field(DESC, "Task duration")
    field(PREC, "2")
    field(EGU,  "s")
}

2. Define a Python PV Interface

Use a dataclass to define EPICS PV templates. Standard PVs (trigger, busy, heartbeat, task_status) are provided automatically.

from dataclasses import dataclass
from epics_bridge.base_pv_interface import BasePVInterface

@dataclass
class MotorInterface(BasePVInterface):
    position_rbv: str = "{main}Pos:RBV"
    velocity_sp: str  = "{main}Vel:SP"
    temperature: str  = "{sys}Temp:Mon"

3. Implement Control Logic

Subclass BridgeDaemon and implement the synchronous run_task() method.

from epics_bridge.daemon import BridgeDaemon, TaskStatus

class MotorControlDaemon(BridgeDaemon):
    def run_task(self) -> TaskStatus:
        velocity = self.io.pvget(self.interface.velocity_sp)

        if velocity is None:
            return TaskStatus.IO_FAILURE

        new_position = velocity * 0.5

        self.io.pvput({
            self.interface.position_rbv: new_position
        })

        return TaskStatus.SUCCESS

4. Run the Daemon

def main():

    prefixes = {
        "main": "IOC:MOTOR:01:",
        "sys": "IOC:SYS:"
    }

    interface = MotorInterface(prefixes=prefixes)

    daemon = MotorControlDaemon(
        interface=interface,
    )

    daemon.start()

if __name__ == "__main__":
    main()

Example: Echo daemon (IOC + daemon)

This repository includes a complete example under examples/echo_daemon/:

• st.cmd: IOC startup script (loads examples/base_interface.db + echo-specific DBs)
• echo_interface.py: PV interface dataclass
• echo_daemon.py: example BridgeDaemon subclass
• main.py: runnable entrypoint which sets up logging and starts the daemon

Typical workflow (requires EPICS + pvxs tooling; easiest via environment.yml):

# Terminal A: start IOC
run-iocsh examples/echo_daemon/st.cmd

# Terminal B: start daemon and write logs to a file
python examples/echo_daemon/main.py /tmp/echo-daemon.log

Testing

# Unit tests (pure Python)
pytest -q

# Integration tests (IOC + daemon)
pytest -m integration -v