pacsys 0.2.1

Pure-python library for Fermilab control system

pacsys

Pure-Python library for Fermilab's control system.

About

ACNET (Accelerator Control NETwork) is the control system used at Fermilab's particle accelerators. PACSys provides a simple Python interface to interact with ACNET data without needing to understand the underlying protocols.

Features

• Read/Write/Stream any ACNET data types with synchronous or async APIs
• Multiple backends to connect to DPM, DMQ, and ACL
• Full DRF3 parser for data requests with automatic conversion
• Utilities for device database, SSH tunneling, and more
• Command-line tools like in EPICS - acget, acput, acmonitor, acinfo

Installation

pip install pacsys

Device API (recommended)

import pacsys
from pacsys import Device, Verify, KerberosAuth

# Create a device -- DRF is validated immediately
dev = Device("M:OUTTMP")

# Read different properties
temperature = dev.read()               # READING (scaled value)
setpoint = dev.setting()               # SETTING property
is_on = dev.status(field="on")         # STATUS field ON
alarm = dev.analog_alarm()             # ANALOG alarm

# Full reading with metadata
reading = dev.get()
print(f"{reading.value} {reading.units}")  # e.g. "72.5 DegF"

# Write with automatic readback verification
result = dev.write(72.5, verify=Verify(tolerance=0.5))
assert result.verified

# Control commands with shortcuts
dev.on()
dev.off()
dev.reset()

# Device database metadata (scaling, limits, units)
info = dev.info()
print(info.description)                # "Outside temperature"
print(info.reading.common_units)       # "DegF"
print(info.reading.min_val)            # 0.0

# Stream data
with dev.with_event("p,1000").subscribe() as stream:
    for reading, handle in stream.readings(timeout=10):
        print(reading.value)

# Immutable -- modifications return new instances
periodic_dev = dev.with_event("p,1000")
sliced_dev = dev.with_range(0, 10)

Backend API

import time
import pacsys

# Read a device value through global backend
temperature = pacsys.read("M:OUTTMP")
print(f"Temperature: {temperature}")

# Stream real-time data through global backend
with pacsys.subscribe(["M:OUTTMP@p,1000"]) as stream:
    for reading, handle in stream.readings(timeout=30):
        print(f"{reading.name}: {reading.value}")

# Stream with callback dispatch mode through dedicated DPM instance
# WORKER (default): callbacks on dedicated worker thread, protects event loop
# DIRECT: callbacks inline on reactor thread (lower latency)
with pacsys.dpm(dispatch_mode=pacsys.DispatchMode.DIRECT) as backend:
    handle = backend.subscribe(
        ["M:OUTTMP@p,1000"],
        callback=lambda r, h: print(r.value),
    )
    time.sleep(10)
    handle.stop()

# Write through authenticated DPM instance (requires kerberos ticket)
with pacsys.dpm(auth=pacsys.KerberosAuth(), role="testing") as backend:
    backend.write("Z:ACLTST", 72.5)

Async capabilities

Native async versions with same API surface.

import pacsys.aio as aio

# Module-level API (mirrors pacsys.read, pacsys.get, etc.)
value = await aio.read("M:OUTTMP")
reading = await aio.get("M:OUTTMP")

# Explicit async backend
async with aio.dpm(auth=pacsys.KerberosAuth()) as backend:
    await backend.write("Z:ACLTST", 72.5)

# Async streaming
async with await backend.subscribe(["M:OUTTMP@p,1000"]) as stream:
    async for reading, handle in stream.readings(timeout=30):
        print(f"{reading.name}: {reading.value}")

# AsyncDevice
from pacsys.aio import AsyncDevice

dev = AsyncDevice("M:OUTTMP", backend=backend)
temp = await dev.read()
await dev.on()

SSH Utilities

Port tunneling, SFTP, and interactive processes over multi-hop SSH.

import pacsys

# Execute commands with automatic Kerberos auth
with pacsys.ssh(["jump.fnal.gov", "target.fnal.gov"]) as ssh:
    result = ssh.exec("hostname")
    print(result.stdout) # target

# ACL can be run on the fly - beam switch, DB, etc.
with pacsys.ssh("clx01.fnal.gov") as ssh:
    result = ssh.acl("read M:OUTTMP") # "M:OUTTMP       =  72.500 DegF"

CLI Tools

EPICS-style command-line tools:

# Read devices
acget M:OUTTMP Z:ACLTST
acget --format json M:OUTTMP

# Write devices (requires authentication, kerberos attempted by default)
acput Z:ACLTST 72.5
acput -a kerberos -b dmq --verify --tolerance 0.5 Z:ACLTST 72.5

# Monitor (streaming on default event or custom one)
acmonitor M:OUTTMP
acmonitor -n 10 M:OUTTMP@p,500

# Device info (DB + property reads)
acinfo -v M:OUTTMP

Also aliased under pacsys-get, pacsys-put, pacsys-monitor, pacsys-info.

Requirements

• Python 3.10+
• For writes: Kerberos credentials with appropriate role or console class assigned
• For some utilities: must run on the controls network or have SSH access to it

Documentation

See the full documentation for guides, API reference, and protocol details.