SGN-EPICS

SGN-EPICS connects SGN-TS streaming pipelines to EPICS Channel Access, using caproto for both client and server roles. It provides four elements:

• EpicsSource — CA client source: monitors existing PVs (camonitor) and emits their values as a uniformly sampled stream.
• EpicsIOCSource — CA server source: serves its own soft IOC of writable PVs; values written by external clients (caput) enter the stream.
• EpicsSink — CA client sink: writes stream values to existing PVs (caput), once per stride.
• EpicsIOCSink — CA server sink: serves stream values as its own soft IOC of read-only PVs, timestamped with stream time.

Timing model

EPICS PVs update asynchronously; the sources convert them to uniform time series by sample-and-hold: at every tick of a GPS-aligned sample clock (rate, a power of 2, default 16 Hz) the most recent value of each channel is latched into the stream. Ticks before a PV's first value are emitted as gaps (EpicsSource) or carry the initial value (EpicsIOCSource). Channel Access has no history, so the sources are live-only: start must be None; duration/end may bound the stream. Only scalar numeric (or enum) PVs are supported.

The sinks update EPICS once per stride (default 1 s) with the last non-gap sample of each channel; all-gap strides leave the PV holding its previous value.

Data types

Every element accepts channels as either a list of PV names (all float64) or a dict mapping PV name to a numpy dtype. Only dtypes with an exact Channel Access native type are accepted; anything else raises ValueError:

The dtype sets the emitted series dtype (sources), the type values are cast to before writing (EpicsSink), and the served CA native type (IOC elements). ENUM PVs can be monitored by EpicsSource (values arrive as the numeric enum index) but cannot be served, since a bare numpy dtype cannot carry enum strings. String and waveform (array) PVs are not supported.

Alarm severity

EpicsSource honors EPICS alarm severity: monitor events whose severity exceeds max_severity (default 2 = MAJOR) are recorded as gaps, so INVALID-severity values — untrustworthy by definition — never enter the stream, while values merely in MINOR/MAJOR alarm still flow. Set max_severity=3 to disable the filter.

EpicsIOCSink sets severity on the PVs it serves: UDF/INVALID until a channel's first sample arrives (the never-processed convention), TIMEOUT/INVALID while a channel is in gap or after EOS (the last value stays readable but is flagged stale), and NO_ALARM while data flows.

EpicsSink and EpicsIOCSource are unaffected: a Channel Access client write cannot set severity, and caputs to served setpoints carry none.

Caveats

All IOC elements in a process serve their PVs through one shared Channel Access server (one IOC identity: one name-search socket, one TCP port, one beacon stream), started when the first element registers channels and stopped when the last withdraws them. Duplicate PV names across elements are rejected, and interfaces is process-wide: every IOC element must agree on it.

Multiple CA-serving processes on one host still resolve names correctly only when clients search via broadcast addresses (the normal EPICS configuration). With a unicast address list (for example EPICS_CA_ADDR_LIST=127.0.0.1 in loopback test setups) the kernel delivers each search datagram to only one of the sockets sharing the UDP port, so only one process's PVs will resolve — avoid running more than one CA-serving process per host in unicast-only environments. Within one process this is not a concern.

Installation

pip install sgn-epics

Example

Monitor two PVs for 60 seconds and serve a derived value as a soft IOC:

from sgn import Pipeline
from sgnts.transforms import Adder
from sgn_epics import EpicsIOCSink, EpicsSource

src = EpicsSource(
    name="src",
    channels=["H1:PEM-EY_WIND_ROOF_WEATHER_MPH", "H1:PEM-EX_WIND_ROOF_WEATHER_MPH"],
    rate=16,
    duration=60,
)
add = Adder(name="add", sink_pad_names=list(src.channels))
snk = EpicsIOCSink(name="snk", channels=["H1:GRD-WIND_SUM"], stride=1.0)

pipeline = Pipeline()
pipeline.connect(src, add)
pipeline.connect(add, snk, link_map={"H1:GRD-WIND_SUM": add.source_pad_names[0]})
pipeline.run()

Serve writable setpoint PVs as a soft IOC and mirror the values to typed readback PVs on an existing IOC. Operators caput to the served PVs; each value enters the stream sample-and-hold style and is written out once per stride, cast to its channel's dtype (so the offset goes over Channel Access as a native LONG):

from sgn import Pipeline
from sgn_epics import EpicsIOCSource, EpicsSink

settings = EpicsIOCSource(
    name="settings",
    channels={"X2:TST-GAIN": "float32", "X2:TST-OFFSET_CTS": "int32"},
    initial_values={"X2:TST-GAIN": 1.0},
    rate=16,
)
mirror = EpicsSink(
    name="mirror",
    channels={"X2:TST-GAIN_RB": "float32", "X2:TST-OFFSET_CTS_RB": "int32"},
    stride=1.0,
)

Pipeline().connect(
    settings,
    mirror,
    link_map={
        "X2:TST-GAIN_RB": "X2:TST-GAIN",
        "X2:TST-OFFSET_CTS_RB": "X2:TST-OFFSET_CTS",
    },
).run()