Metadata-Version: 2.4
Name: pacsys
Version: 0.2.2
Summary: Pure-python library for Fermilab control system
Author: Nikita Kuklev
License-Expression: GPL-3.0-or-later
Project-URL: Homepage, https://github.com/fast-iota/pacsys
Project-URL: Documentation, https://fast-iota.github.io/pacsys/
Project-URL: Repository, https://github.com/fast-iota/pacsys
Project-URL: Issues, https://github.com/fast-iota/pacsys/issues
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.10
Classifier: Operating System :: OS Independent
Classifier: Topic :: Scientific/Engineering
Classifier: Intended Audience :: Science/Research
Classifier: Development Status :: 4 - Beta
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy>=1.23.0
Requires-Dist: tqdm>=4.60.0
Requires-Dist: httpx>=0.23.0
Requires-Dist: websockets>=11.0
Requires-Dist: grpcio>=1.66.0
Requires-Dist: protobuf>=6.30.0
Requires-Dist: pika>=1.2.0
Requires-Dist: gssapi>=1.8.0
Requires-Dist: paramiko[gssapi]>=3.0
Provides-Extra: parquet
Requires-Dist: pyarrow>=14.0; extra == "parquet"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: ty; extra == "dev"
Requires-Dist: prek; extra == "dev"
Requires-Dist: tzdata; extra == "dev"
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"
Requires-Dist: types-grpcio; extra == "dev"
Requires-Dist: types-pika-ts; extra == "dev"
Requires-Dist: types-protobuf; extra == "dev"
Requires-Dist: pyarrow>=14.0; extra == "dev"
Provides-Extra: mcp
Requires-Dist: fastmcp; extra == "mcp"
Requires-Dist: tomli>=2.0; python_version < "3.11" and extra == "mcp"
Provides-Extra: doc
Requires-Dist: mkdocs; extra == "doc"
Requires-Dist: mkdocs-jupyter; extra == "doc"
Requires-Dist: mkdocs-macros-plugin; extra == "doc"
Requires-Dist: mkdocs-material; extra == "doc"
Dynamic: license-file

<h1 align="center">pacsys</h1>

<p align="center">Pure-Python library for Fermilab's control system.</p>

<p align="center">
  <a href="https://github.com/fast-iota/pacsys/actions/workflows/tests.yml"><img src="https://github.com/fast-iota/pacsys/actions/workflows/tests.yml/badge.svg" alt="Tests"></a>
  <a href="https://fast-iota.github.io/pacsys/"><img src="https://img.shields.io/badge/docs-available-blue" alt="Documentation"></a>
  <a href="https://github.com/fast-iota/pacsys/blob/master/LICENSE"><img src="https://img.shields.io/badge/license-GPL--3.0-green" alt="License: GPL-3.0"></a>
  <a href="https://www.python.org/"><img src="https://img.shields.io/badge/python-3.10%2B-blue" alt="Python 3.10+"></a>
</p>

## 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`
- **MCP server** for AI agent integration (Claude Code, etc.) with supervised writes

## Installation

```bash
pip install pacsys
```

## Device API (recommended)

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

# Block until next reading arrives
reading = dev.with_event("p,1000").await_next(timeout=5)
print(reading.value)

# 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

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

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

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

## MCP Server (AI Agent Integration)

Expose pacsys as an MCP server for AI agents like Claude Code. Read-only by default, granular write policy engine available.

```json
{
  "mcpServers": {
    "pacsys": {
      "command": "python",
      "args": ["-m", "pacsys.mcp"]
    }
  }
}
```

## Experimental Utilities

`pacsys.exp` provides high-level utilities for experiment workflows:

```python
from pacsys.exp import Monitor, read_fresh, watch, scan, DataLogger, CsvWriter, ParquetWriter

# Collect readings for 10 seconds
result = Monitor(["M:OUTTMP@p,1000", "G:AMANDA@e,8f"]).collect(duration=10)
print(result.mean("M:OUTTMP@p,1000"))
print(result.median("M:OUTTMP@p,1000"))

# Continuous monitoring with blocking wait
mon = Monitor(["M:OUTTMP@p,1000"])
mon.start()
reading = mon.await_next("M:OUTTMP@p,1000")  # block until next reading
snap = mon.snapshot()                          # all buffered data still available
mon.stop()

# Time-slice and export
sliced = result.slice("M:OUTTMP@p,1000", start=t0, end=t1)
timestamps, values = result.to_numpy("M:OUTTMP@p,1000")
df = result.to_dataframe(relative=True)  # elapsed seconds index

# Wait for fresh readings (with multi-count and stats)
results = read_fresh(["M:OUTTMP@p,1000"], count=10, timeout=5.0)
print(results[0].mean())  # windowed stats: mean, std, median, min, max

# Watch for a condition
reading = watch("M:OUTTMP@p,1000", lambda r: r.value > 75, timeout=30)

# Parameter scan with restore
result = scan("Z:ACLTST", ["M:OUTTMP"], values=[0.0, 1.0, 2.0], settle=0.5)

# Log to CSV
with DataLogger(["M:OUTTMP@p,1000"], writer=CsvWriter("log.csv")):
    time.sleep(60)

# Log to Parquet (typed columns, ZSTD compression)
with DataLogger(["M:OUTTMP@p,1000"], writer=ParquetWriter("log.parquet")):
    time.sleep(60)
```

## CLI Tools

EPICS-style command-line tools:

```bash
# 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](https://fast-iota.github.io/pacsys/) for guides, API reference, and protocol details.
