iocmng — IOC Manager Framework

A pluggable task/job framework for EPICS soft IOC applications on Kubernetes. Provides base classes for continuous tasks and one-shot jobs, a declarative rule/transform engine, and a REST API for dynamic plugin management at runtime.

Key Features

• TaskBase — base class for continuous, triggered, or reactive tasks
• JobBase — base class for one-shot jobs returning structured results
• DeclarativeTask — zero-code tasks driven entirely by config.yaml rules and transforms
• Wired inputs/outputs — automatically read/write external PVs (poll or monitor)
• Connection tracking — CONN_INP / CONN_OUT waveform PVs show live CA/PVA connection state per port
• Declarative rules — safe boolean expressions that fire actuators and set outputs
• Output latch — latch: true holds an output value until an operator writes CLEAR; latch_dir selects which transition latches (rise / fall / any)
• ALARM severity — alarm_on: MAJOR|MINOR wires EPICS alarm fields directly from config
• Transforms — computed outputs using built-in math/statistics/array functions
• Ring buffers — buffer_size accumulates time-series data for signal processing
• Built-in function library — mean, std, sqrt, clamp, moving_avg, derivative, and more
• Safe expression evaluator — AST-validated expressions; no arbitrary code execution
• REST API — add/remove/restart plugins at runtime from git repos or local paths
• EPICS soft IOC PVs — every task gets STATUS, MESSAGE, ENABLE, VERSION, CYCLE_COUNT, CLEAR, RESET PVs
• Per-plugin config.yaml — parameters, directional PVs, rules, transforms in one file
• PV client abstraction — transparent PVA (p4p) or CA (PyEPICS) access; ENABLE/CLEAR/RESET polled each cycle to catch external writes
• Plugin validation — syntax, inheritance, abstract methods checked before acceptance
• Docker image — ready-to-run container with the REST API
• Standalone runner — iocmng-run for local development without a server
• Optional Ophyd — device abstraction via ophyd/infn_ophyd_hal

Quick Start

Install

pip install iocmng

# With all optional dependencies
pip install iocmng[all]

Run the API Server

iocmng-server

# With configuration
IOCMNG_PORT=8080 IOCMNG_PREFIX=SPARC:CONTROL iocmng-server

Create a Task (Python)

my_monitor.py

from iocmng import TaskBase

class MyMonitor(TaskBase):
    def initialize(self):
        self.logger.info("Starting monitor")

    def execute(self):
        value = self.get_pv("READING")
        if value and value > self.parameters.get("threshold", 75):
            self.set_pv("ALARM", 1)

    def cleanup(self):
        pass

config.yaml

parameters:
  mode: continuous
  interval: 1.0
  threshold: 75.0

arguments:
  inputs:
    SETPOINT:
      type: float
      value: 50.0
      unit: "%"
      prec: 2
      low: 0
      high: 100
  outputs:
    READING:
      type: float
      value: 0.0
    ALARM:
      type: bool
      value: 0
      znam: "OK"
      onam: "ALARM"

Create a Declarative Task (Zero Code)

No Python needed — all logic lives in config.yaml:

my_interlock.py

from iocmng import DeclarativeTask

class MyInterlock(DeclarativeTask):
    pass

config.yaml

parameters:
  mode: continuous
  interval: 1.0
  pva: false

arguments:
  inputs:
    temp:
      type: float
      value: 0.0
      link: "DEVICE:TEMP"         # Wired to external PV
      link_mode: monitor           # monitor pushes values; poll reads each cycle
      buffer_size: 100             # Keep last 100 readings
    pressure:
      type: float
      value: 0.0
      link: "DEVICE:PRESSURE"
      link_mode: monitor
  outputs:
    avg_temp:
      type: float
      value: 0.0
    alarm:
      type: bool
      value: 0
      znam: "OK"
      onam: "ALARM"
      latch: true                  # Hold value until CLEAR is written
      latch_dir: "rise"            # Latch on 0→1 transition (default)
      alarm_on: MAJOR              # EPICS ALARM severity when output is 1

transforms:
  - output: avg_temp
    expression: "mean(temp_buf)"

rule_defaults:
  alarm: 0

rules:
  - id: OVER_TEMP
    condition: "mean(temp_buf) > 80 or pressure > 2.0"
    message: "Temperature or pressure limit exceeded"
    outputs:
      alarm: 1

Run Standalone (no server)

iocmng-run -m my_interlock --config config.yaml --prefix MY:IOC --name interlock

Deploy via REST API

curl -X POST http://localhost:8080/api/v1/plugins \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-interlock",
    "git_url": "https://github.com/org/my-tasks.git",
    "path": "plugins/interlock/",
    "auto_start": true
  }'

Documentation

Task Modes

Built-in Functions

Available in rule conditions and transform expressions:

Extend with register("my_fn", callable) from iocmng.core.functions.

Default PVs

Every task automatically gets:

Environment Variables

Project Structure

src/iocmng/
├── __init__.py           # Exports: TaskBase, JobBase, DeclarativeTask, pv_client, run_ioc
├── declarative.py        # DeclarativeTask (zero-code tasks)
├── runner.py             # Standalone CLI runner (iocmng-run)
├── base/
│   ├── task.py           # TaskBase — continuous/triggered/reactive tasks
│   └── job.py            # JobBase — one-shot jobs
├── core/
│   ├── controller.py     # Central plugin manager
│   ├── loader.py         # Git clone + config loading
│   ├── validator.py      # Plugin validation
│   ├── plugin_spec.py    # PvArgumentSpec, PluginSpec, RuleSpec, TransformSpec
│   ├── safe_eval.py      # AST-validated expression evaluator
│   ├── functions.py      # Built-in function registry
│   └── pv_client.py      # PVA/CA abstraction layer
├── api/
│   ├── app.py            # FastAPI application
│   ├── routes.py         # REST endpoints
│   └── models.py         # Pydantic models
└── ophyd/
    └── factory.py        # Optional ophyd device creation

Development

pip install -e ".[dev]"
pytest tests/ -v
black .
flake8 .

License

MIT