Metadata-Version: 2.4
Name: epics-bridge
Version: 2.0.0
Summary: A generic bridge between EPICS IOCs and Python logic.
Author-email: Hugo Valim <hugo.valim@ess.eu>
License-Expression: MIT
Project-URL: Homepage, https://gitlab.esss.lu.se/hugovalim/epics-bridge
Project-URL: Repository, https://gitlab.esss.lu.se/hugovalim/epics-bridge
Project-URL: Issues, https://gitlab.esss.lu.se/hugovalim/epics-bridge/-/issues
Keywords: epics,p4p,control-system,daemon,ioc,process-variable
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy
Requires-Dist: p4p
Dynamic: license-file

# EPICS Bridge
![Python](https://img.shields.io/badge/python-3.9%2B-blue)
![License](https://img.shields.io/badge/license-MIT-green)

**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

```bash
# 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`.

```bash
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:

```epics
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.

```python
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.

```python
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

```python

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`):

```bash
# 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

```bash
# Unit tests (pure Python)
pytest -q

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