Data Helpers¶

The pacsys.exp module provides high-level utilities for common accelerator physics workflows: monitoring channels, scanning parameters, logging data to files, and waiting for conditions.

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

All functions accept DRF strings or Device objects and use the global default backend unless one is explicitly provided.

Monitor¶

Subscribe to one or more channels and collect readings into ring buffers. Supports both blocking (collect) and non-blocking (start/stop) modes.

Blocking Collection¶

from pacsys.exp import Monitor

mon = Monitor(["M:OUTTMP@p,1000", "G:AMANDA@e,8f"])
result = mon.collect(duration=10.0)

print(result.mean("M:OUTTMP@p,1000"))    # Mean value
print(result.std("M:OUTTMP@p,1000"))     # Standard deviation
print(result.counts)                      # {drf: count} per channel
print(result.rate())                      # Readings/sec per channel

Alternatively, collect until each channel has a minimum number of readings:

result = mon.collect(count=100, timeout=30.0)

Non-Blocking (Start/Stop)¶

mon = Monitor(["M:OUTTMP@p,1000"])
mon.start()

# Peek without consuming data
snap = mon.snapshot()

# Swap out buffers and return old data
result = mon.flush()

mon.stop()

Or use the context manager:

with Monitor(["M:OUTTMP@p,1000"]) as mon:
    snap = mon.snapshot()

Waiting for New Data¶

Use await_next() to block until a new reading arrives on a specific channel:

with Monitor(["M:OUTTMP@p,1000"]) as mon:
    reading = mon.await_next("M:OUTTMP@p,1000", timeout=5.0)
    print(reading.value)

Use wait_until() to block until an arbitrary predicate is satisfied:

with Monitor(["M:OUTTMP@p,1000"]) as mon:
    result = mon.wait_until(
        lambda snap: snap.mean("M:OUTTMP@p,1000") > 72.0,
        timeout=30.0,
    )

Change Detection¶

Use tags and has_new() for cheap polling without copying buffers:

with Monitor(["M:OUTTMP@p,1000"]) as mon:
    old_tags = mon.tags
    # ... do other work ...
    if mon.has_new(old_tags):
        snap = mon.snapshot()

Channel Health¶

health() returns a ChannelHealth snapshot per channel (or one channel if drf is given): last reading, seconds since it arrived (gap), total received count, and a stale flag.

Passing stale_after to the constructor starts a watchdog that flags a channel stale when no reading arrives for that many seconds (with an initial grace period of stale_after after start). Transitions are logged and can trigger callbacks:

def alarm(drf, health):
    print(f"{drf} stale, {health.gap:.1f}s since last reading")

with Monitor(["M:OUTTMP@p,1000"], stale_after=5.0, on_stale=alarm, on_recover=print) as mon:
    ...
    if mon.health("M:OUTTMP@p,1000").stale:
        ...

Without stale_after, health() still works but stale is always False.

Constructor Options¶

Parameter	Type	Default	Description
`devices`	`list[DeviceSpec]`	required	DRF strings or `Device` objects
`buffer_size`	`int`	`10_000`	Max readings per channel (ring buffer)
`backend`	`Backend`	`None`	Backend to use (global default if `None`)
`stale_after`	`float`	`None`	Seconds without a reading before a channel is flagged stale
`on_stale`	`Callable[[str, ChannelHealth], None]`	`None`	Called when a channel becomes stale
`on_recover`	`Callable[[str, ChannelHealth], None]`	`None`	Called when a stale channel receives data again

MonitorResult¶

Returned by Monitor.collect(), snapshot(), and flush(). Provides statistics, slicing, and export methods.

Statistics¶

All stat methods accept an optional drf argument. If given, returns a single value; if omitted, returns a dict keyed by DRF.

Method	Description
`mean(drf?)`	Arithmetic mean
`std(drf?)`	Standard deviation (population)
`median(drf?)`	Median value
`min(drf?)`	Minimum value
`max(drf?)`	Maximum value
`rate(drf?)`	Readings per second
`last(n, drf?)`	Last n values

Properties¶

Property	Type	Description
`channels`	`dict[str, ChannelData]`	Per-channel data
`counts`	`dict[str, int]`	Reading count per channel
`elapsed`	`timedelta \\| None`	Duration between start and stop
`started`	`datetime \\| None`	When collection started
`stopped`	`datetime \\| None`	When collection stopped

Accessing Raw Data¶

values = result.values("M:OUTTMP@p,1000")          # list[Value]
timestamps = result.timestamps("M:OUTTMP@p,1000")  # list[datetime]

Time Slicing¶

from datetime import datetime, timezone

start = datetime(2026, 3, 5, 12, 0, tzinfo=timezone.utc)
end = datetime(2026, 3, 5, 12, 5, tzinfo=timezone.utc)
sliced = result.slice("M:OUTTMP@p,1000", start=start, end=end)
# Returns a ChannelData with only readings in [start, end]

Export to NumPy¶

timestamps, values = result.to_numpy("M:OUTTMP@p,1000")
# timestamps: float64 epoch seconds (UTC)
# values: float64

Export to pandas¶

# Single channel — indexed by timestamp
df = result.to_dataframe("M:OUTTMP@p,1000")

# All channels — flat table with drf column
df = result.to_dataframe()

# Relative timestamps (seconds since collection started)
df = result.to_dataframe("M:OUTTMP@p,1000", relative=True)

Relative export requires both the collection start time and a timestamp on every included reading; missing timestamps raise ValueError instead of producing an invalid index.

read_fresh¶

Wait for one or more fresh readings per channel via a temporary subscription. Consider it a pacsys.read() but with a lot more options and ability to collect multiple readings.

from pacsys.exp import read_fresh

# Single reading per channel (default)
results = read_fresh(["M:OUTTMP", "G:AMANDA"], default_event="p,1000")
for r in results:
    print(f"{r.drf}: {r.value}")

# Collect 10 readings and average
results = read_fresh(["M:OUTTMP@p,1000"], count=10, timeout=5.0)
print(results[0].mean())     # mean of last 10
print(results[0].std())      # standard deviation
print(results[0].median())   # median
print(results[0].values)     # all collected values

Parameters¶

Parameter	Type	Default	Description
`devices`	`list[DeviceSpec]`	required	DRF strings or `Device` objects
`count`	`int`	`1`	Readings to collect per channel
`default_event`	`str \\| None`	`None`	Event to apply if DRF has none
`timeout`	`float`	`5.0`	Max seconds to wait
`backend`	`Backend \\| None`	`None`	Backend to use

FreshResult¶

Each element in the returned list is a FreshResult:

Property / Method	Returns	Description
`value`	`Value`	Last reading's value
`values`	`list[Value]`	All ok values
`reading`	`Reading`	Last Reading object
`timestamp`	`datetime`	Last reading's timestamp
`timestamps`	`list[datetime]`	All timestamps
`length`	`int`	Number of ok values
`len(r)`	`int`	Total readings collected
`requested_count`	`int`	How many were requested

Windowed Statistics¶

Stats methods accept an optional n parameter:

Call	Window
`r.mean()`	Last `requested_count` values
`r.mean(-1)`	All values
`r.mean(8)`	Last 8 values (raises if < 8)

Available: mean, median, std, min, max.

watch¶

Block until a condition is met on a streaming channel.

from pacsys.exp import watch

# Wait for temperature to exceed 80
reading = watch(
    "M:OUTTMP@p,1000",
    condition=lambda r: r.ok and r.value > 80.0,
    timeout=60.0,
)
print(f"Crossed threshold at {reading.timestamp}: {reading.value}")

Parameter	Type	Default	Description
`device`	`DeviceSpec`	required	DRF string or `Device` (must include a streaming event)
`condition`	`Callable[[Reading], bool]`	required	Predicate; returns `True` to stop
`timeout`	`float`	`30.0`	Max seconds to wait
`backend`	`Backend \\| None`	`None`	Backend to use

Returns the Reading that satisfied the condition. Raises TimeoutError if the condition is not met within the timeout.

scan¶

Ramp a device through a series of values while reading other devices at each step. Automatically restores the original setting value on completion or error.

If restoration fails after the scan completes, ScanRestoreError.result retains the collected data with restored=False. A raised restoration exception is preserved as the cause. If the scan itself raises, restoration failure is logged and the original scan exception propagates.

from pacsys.exp import scan

result = scan(
    write_device="Z:ACLTST",
    read_devices=["M:OUTTMP", "G:AMANDA"],
    start=0.0,
    stop=10.0,
    steps=11,
    settle=0.5,
)

# Inspect results
for sv, step in zip(result.set_values, result.readings):
    print(f"Set {sv}: read {step['M:OUTTMP'].value}")

# Export to DataFrame
df = result.to_dataframe()

Explicit Values¶

result = scan(
    write_device="Z:ACLTST",
    read_devices=["M:OUTTMP"],
    values=[0.0, 2.5, 5.0, 7.5, 10.0],
)

Averaging and Abort¶

result = scan(
    write_device="Z:ACLTST",
    read_devices=["M:OUTTMP"],
    start=0.0, stop=10.0, steps=11,
    readings_per_step=5,  # Average 5 readings per step
    abort_if=lambda step: step["M:OUTTMP"].value > 100.0,
)
print(result.aborted)   # True if abort_if triggered or a write failed
print(result.restored)  # True if original value was restored

Parameters¶

Parameter	Type	Default	Description
`write_device`	`DeviceSpec`	required	Device to ramp
`read_devices`	`list[DeviceSpec]`	required	Devices to read at each step
`values`	`list[float]`	`None`	Explicit scan values
`start` / `stop` / `steps`	`float` / `float` / `int`	`None`	Linear range (alternative to `values`)
`settle`	`float`	`0.5`	Seconds to wait after each write
`readings_per_step`	`int`	`1`	Readings to average per step (arrays element-wise; non-numeric values raise)
`verify`	`bool \\| Verify`	`None`	Verify writes (see Writing)
`restore`	`bool`	`True`	Restore original setting on completion
`abort_if`	`Callable`	`None`	Abort predicate receiving step readings
`timeout`	`float \\| None`	`None`	Per-operation timeout
`backend`	`Backend \\| None`	`None`	Backend to use

ScanResult¶

Property	Type	Description
`write_device`	`str`	DRF of the ramped device
`read_devices`	`list[str]`	DRFs of the read devices
`set_values`	`list[float]`	Values that were written
`readings`	`list[dict[str, Reading]]`	Per-step readings keyed by DRF
`write_results`	`list[WriteResult]`	Per-step write results
`aborted`	`bool`	`abort_if` triggered or a write failed/unconfirmed before the last value
`restored`	`bool`	Whether original value was restored
`to_dataframe()`	`DataFrame`	Export with `set_value` + read columns

DataLogger¶

Subscribe to channels and log readings to a file via a pluggable writer. Runs in the background with periodic flushing.

from pacsys.exp import DataLogger, CsvWriter

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

Manual Control¶

dl = DataLogger(
    ["M:OUTTMP@p,1000", "G:AMANDA@e,8f"],
    writer=ParquetWriter("data.parquet"),
    flush_interval=10.0,
)
dl.start()
# ... do work ...
dl.stop()  # Flushes remaining data and closes the writer

Constructor Options¶

Parameter	Type	Default	Description
`devices`	`list[DeviceSpec]`	required	Channels to subscribe to
`writer`	`LogWriter`	required	Writer implementation
`flush_interval`	`float`	`5.0`	Seconds between flushes
`backend`	`Backend \\| None`	`None`	Backend to use

Properties¶

Property	Type	Description
`running`	`bool`	Whether the logger is actively collecting
`last_error`	`Exception \\| None`	Last write or subscription error, if any
`failed`	`bool`	True once a batch was dropped after exhausting write retries (logging continues) or the subscription failed (logging stops)
`dropped_count`	`int`	Readings lost so far (sticky until the next `start()`)

A dropped batch or failed subscription is never silent: stop() (and context-manager exit) flushes what was received, closes the writer, and raises RuntimeError chained to the subscription error or, for drops, to last_error.

Failed writes are retried up to 3 times before the batch is dropped. Errors are logged and available via last_error. If a batch is dropped after stop() begins, shutdown closes the writer and raises RuntimeError; if the flush worker is stuck, shutdown raises without closing the writer from another thread.

Writers¶

Writers implement the LogWriter protocol:

class LogWriter(Protocol):
    def write_readings(self, readings: list[Reading]) -> None: ...
    def close(self) -> None: ...

CsvWriter¶

Simple CSV output with columns: timestamp, drf, value, units, facility_code, error_code, message. Status fields are preserved from each reading; missing messages are empty. Error rows can be distinguished from successful empty text values, and warnings retain their values. Each batch is flushed to the operating system before write_readings() returns. Parquet files require a successful close() before they are readable.

from pacsys.exp import CsvWriter

writer = CsvWriter("output.csv")

ParquetWriter¶

Typed columnar output using Apache Parquet (requires pyarrow). Handles scalars, arrays, text, raw bytes, alarms, and status values in separate typed columns.

from pacsys.exp import ParquetWriter

writer = ParquetWriter("output.parquet")

Schema:

Column	Type	Content
`timestamp`	`timestamp[us, UTC]`	Reading timestamp
`drf`	`string`	DRF string
`value_type`	`string`	Value type name
`value`	`float64`	Float scalar values
`int_value`	`int64`	Integer and boolean scalar values
`value_array`	`list<float64>`	Array values
`value_text`	`string`	Text, JSON-encoded alarms/status, base64-encoded raw bytes
`error_code`	`int16`	ACNET error code
`units`	`string`	Engineering units
`cycle`	`int64`	Cycle number
`facility_code`	`int16`	ACNET facility identifier
`message`	`string`	Status message; null when absent, distinct from an empty string

You can also implement your own writer by conforming to the LogWriter protocol.

Data Helpers¶

Monitor¶

Blocking Collection¶

Non-Blocking (Start/Stop)¶

Waiting for New Data¶

Change Detection¶

Channel Health¶

Constructor Options¶

MonitorResult¶

Statistics¶

Properties¶

Accessing Raw Data¶

Time Slicing¶

Export to NumPy¶

Export to pandas¶

read_fresh¶

Parameters¶

FreshResult¶

Windowed Statistics¶

watch¶

scan¶

Explicit Values¶

Averaging and Abort¶

Parameters¶

ScanResult¶

DataLogger¶

Manual Control¶

Constructor Options¶

Properties¶

Writers¶

CsvWriter¶

ParquetWriter¶

See Also¶