epics-bridge 1.0.1

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.

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 process voluntarily terminates (exit(0)). 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

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 execute() method.

from epics_bridge.daemon import BridgeDaemon, TaskStatus

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

        if velocity is None:
            return TaskStatus.ERROR

        new_position = velocity * 0.5

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

        return TaskStatus.DONE

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()