Metadata-Version: 2.4
Name: epics-bridge
Version: 4.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==2.4.2
Requires-Dist: p4p==4.2.2
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**
  Small utilities (for example, the `Timer` context manager)

- **epics_bridge.io**
  Synchronous P4P client wrapper and P4P-to-Python value unwrapping


---
## 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")
    # State 4: Skipped (e.g. trigger=False)
    field(FRVL, "4")
    field(FRST, "Skipped")
    field(FRSV, "NO_ALARM")
}

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


### 2. Define a Python PV Interface

Subclass `BasePVInterface` and create `PV` instances in your constructor. Call `super().__init__(prefixes=...)`, add your PVs; placeholder resolution runs automatically. Placeholders use Python format syntax `{key}` (e.g. `{main}`, `{p1}`) and are replaced from `prefixes`. Standard PVs (trigger, busy, heartbeat, task_status, task_duration) are created by the base.

```python
from epics_bridge import BasePVInterface, PV

class MotorInterface(BasePVInterface):
    def __init__(self, prefixes: dict | None = None) -> None:
        super().__init__(prefixes=prefixes)
        self.position_rbv = PV("{main}Pos:RBV")
        self.velocity_sp = PV("{main}Vel:SP")
        self.temperature = PV("{sys}Temp:Mon")
```

### 3. Implement Control Logic

Subclass `BridgeDaemon` and implement the synchronous `run_task()` method.
Use `pvget(PV or list of PVs)` to read (mutates each PV’s `.val` and `.raw`), then read `pv.val`.
Use `pvput(list of PVs)` to write (each PV’s `.val` is written to its channel). Let exceptions from `run_task()` bubble up; the base class guarantees cleanup and logs failures.

```python
from epics_bridge import BridgeDaemon, TaskStatus

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

        if velocity is None:
            return TaskStatus.IO_FAILURE

        self.iface.position_rbv.val = velocity * 0.5
        self.io.pvput([self.iface.position_rbv])

        return TaskStatus.SUCCESS
```

### 4. Run the Daemon

```python

def main():

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

    interface = MotorInterface(prefixes=prefixes)

    daemon = MotorControlDaemon(
        iface=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 (PVs created in constructor)
- `echo_daemon.py`: example `BridgeDaemon` subclass
- `entrypoint.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
E3_CMD_TOP="$(pwd)/examples/echo_daemon" run-iocsh examples/echo_daemon/st.cmd

# Terminal B: start daemon (logs under --log-dir)
python examples/echo_daemon/entrypoint.py --log-dir /tmp
```

## Testing

```bash
# Unit tests (pure Python)
pytest -m "not integration" -v
# or
conda run -n epics-bridge pytest -m "not integration" -v

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