Skip to content

Streaming Data

Streaming lets you receive continuous updates from devices as they change.

Basic Streaming

Subscribe to a device with a periodic event:

import pacsys

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

The @p,1000 means "send data every 1000 milliseconds." The timeout is a total wall-clock timeout from the first .readings() call (not a per-reading idle timeout). See DRF Events for all event types.

Always Use Context Manager

The with statement ensures the subscription is properly cleaned up. Without it, you must call stream.stop() manually.

Multiple Devices

Subscribe to several devices in one subscription:

with pacsys.subscribe([
    "M:OUTTMP@p,1000",   # Every second
    "G:AMANDA@p,500",    # Every 500ms
]) as stream:
    for reading, handle in stream.readings(timeout=60):
        print(f"{reading.name}: {reading.value}")

Readings arrive interleaved from all devices.

Stopping Early

Stop when a condition is met:

with pacsys.subscribe(["M:OUTTMP@p,1000"]) as stream:
    for reading, handle in stream.readings(timeout=60):
        print(f"Temperature: {reading.value}")
        if reading.value > 100:
            stream.stop()

Breaking out of the for loop also works - the context manager calls stop() on exit.

After stop() returns, reading callbacks still queued for that handle are discarded (an on_error callback is always delivered). A callback that is already executing when stop() is called runs to completion (an async callback stopped from another task is cancelled at its next await). A stream that ends on its own still delivers every queued reading.

handle.dropped counts readings discarded because the iterator buffer or the callback queue was full (cumulative for the life of the handle; a throttled warning is logged as well).

Callback Mode

For background streaming, use a callback instead of iteration:

import pacsys
import time

def on_reading(reading, handle):
    print(f"{reading.name}: {reading.value}")
    if reading.value > 100:
        handle.stop()

handle = pacsys.subscribe(
    ["M:OUTTMP@p,1000"],
    callback=on_reading,
)

# Do other work while data streams in the background
time.sleep(30)

# Clean up
handle.stop()
pacsys.shutdown()

The callback runs on a dedicated worker thread by default -- keep it fast to avoid delaying delivery of other readings.

Dispatch Mode

By default, callbacks run on a dedicated worker thread (DispatchMode.WORKER), which protects the reactor from slow callbacks. For lower latency, use DispatchMode.DIRECT to run callbacks inline on the reactor thread (i.e. the event loop thread):

from pacsys import DispatchMode

with pacsys.dpm(dispatch_mode=DispatchMode.DIRECT) as backend:
    handle = backend.subscribe(
        ["M:OUTTMP@p,1000"],
        callback=lambda r, h: print(r.value),
    )
    import time; time.sleep(10)
    handle.stop()

DIRECT mode

In DIRECT mode, slow callbacks block the reactor thread and delay all readings on that connection. Only use DIRECT when your callback is very fast.

Dispatch mode is configured per-backend, not per-subscription. The global backend (used by pacsys.subscribe()) always uses WORKER.

One worker thread serves every subscription on a backend, so one slow callback delays all of them and, if the queue fills, readings are dropped (handle.dropped). Give a slow consumer its own backend instance, or hand the work off to your own thread/executor from the callback.

Error Handling

Iterator Mode

Errors during streaming raise an exception when iterating:

try:
    with pacsys.subscribe(["M:OUTTMP@p,1000"]) as stream:
        for reading, handle in stream.readings(timeout=30):
            print(reading.value)
except Exception as e:
    print(f"Stream error: {e}")

Check handle.exc to inspect stored exceptions:

with pacsys.subscribe(["M:OUTTMP@p,1000"]) as stream:
    for reading, handle in stream.readings(timeout=5):
        print(reading.value)

    if stream.exc:
        print(f"Stream ended with error: {stream.exc}")

Callback Mode

Use on_error to handle connection errors:

def on_error(exc, handle):
    print(f"Connection error: {exc}")

handle = pacsys.subscribe(
    ["M:OUTTMP@p,1000"],
    callback=on_reading,
    on_error=on_error,
)

CombinedStream

Combine multiple independent subscriptions into a single iterable:

from pacsys import CombinedStream

with pacsys.dpm() as backend:
    sub1 = backend.subscribe(["M:OUTTMP@p,1000"])
    sub2 = backend.subscribe(["G:AMANDA@p,500"])

    with CombinedStream([sub1, sub2]) as combined:
        for reading, handle in combined.readings(timeout=30):
            print(f"{reading.name}: {reading.value}")

Each subscribe() call creates its own TCP connection (on DPM/HTTP), so subscriptions are truly independent - stopping one doesn't affect the others.

Delivery through CombinedStream is at-most-once: readings it has prefetched but not yet yielded are discarded if you exit the loop early or a subscription errors. Iterate each handle directly if you need every buffered reading. The shared buffer is bounded (same cap as a single handle); a slow consumer backpressures the source handles, which drop newest and count in handle.dropped as usual.

CombinedStream properties:

Property Description
stopped True when all subscriptions have stopped
exc First exception from any subscription
stop() Stop all subscriptions

Common Patterns

Periodic Sampling

import pacsys
import time

with pacsys.subscribe(["M:OUTTMP@p,5000"]) as stream:
    start = time.time()
    for reading, _ in stream.readings(timeout=60):
        elapsed = time.time() - start
        print(f"[{elapsed:.1f}s] {reading.value}")

Collecting Data

import pacsys

data = []

with pacsys.subscribe(["M:OUTTMP@p,1000"]) as stream:
    for reading, handle in stream.readings(timeout=10):
        data.append({
            "time": reading.timestamp,
            "value": reading.value,
        })

print(f"Collected {len(data)} samples")

TCLK Event Trigger

# Trigger on TCLK event $0F
with pacsys.subscribe(["M:OUTTMP@E,0F"]) as stream:
    for reading, _ in stream.readings(timeout=60):
        print(f"Event triggered: {reading.value}")

Device API

The Device class has a subscribe() method that builds the DRF automatically:

from pacsys import Device

dev = Device("M:OUTTMP@p,1000")

# Iterator mode
with dev.subscribe() as stream:
    for reading, _ in stream.readings(timeout=10):
        print(reading.value)

# Callback mode
handle = dev.subscribe(callback=lambda r, h: print(r.value))

Subscribe to specific properties or override the event:

dev = Device("M:OUTTMP")

with dev.subscribe(prop="setting", event="p,1000") as stream:
    for reading, _ in stream.readings(timeout=10):
        print(f"Setpoint: {reading.value}")

Device API Guide - full subscribe() reference

Explicit Backend

All backends support streaming with the same API:

import pacsys

# DPM/HTTP
with pacsys.dpm() as backend:
    with backend.subscribe(["M:OUTTMP@p,1000"]) as stream:
        for reading, _ in stream.readings(timeout=10):
            print(reading.value)

# DMQ (requires Kerberos)
from pacsys import KerberosAuth
with pacsys.dmq(auth=KerberosAuth()) as backend:
    with backend.subscribe(["M:OUTTMP@p,1000"]) as stream:
        for reading, _ in stream.readings(timeout=10):
            print(reading.value)

See Also