Poll Time Series Updates

When streaming RHEED data, use the polling utilities in atomscale.timeseries to receive analysis updates as they arrive. These helpers always read the RHEED timeseries for data_id, so they work only with RHEED videos. Each poll returns a DataFrame indexed by ["View ID", "Frame Number"], the same shape as get_rheed_timeseries(). Four entry points cover different concurrency needs:

Function	Use case
`iter_poll`	Synchronous loop that blocks between polls
`start_polling_thread`	Background thread for GUI or acquisition loops
`aiter_poll`	Async iterator for `asyncio` applications
`start_polling_task`	Fire-and-forget asyncio task

Setup

from atomscale import Client
from atomscale.timeseries import (
    iter_poll,
    aiter_poll,
    start_polling_thread,
    start_polling_task,
)

client = Client(api_key="YOUR_API_KEY")
data_id = "YOUR_STREAMING_DATA_ID"


def latest_frame(df):
    """Return the newest frame number in the result, for deduplication."""
    if df.empty or "Frame Number" not in df.index.names:
        return None
    return df.index.get_level_values("Frame Number").max()

Synchronous polling

Loop over iter_poll to fetch fresh rows on a fixed cadence. The helper waits interval seconds between polls.

for idx, result in enumerate(
    iter_poll(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=10,
        distinct_by=latest_frame,
        max_polls=3,
    ),
    start=1,
):
    print(f"Poll {idx}: latest frame -> {latest_frame(result)}")
    print(result.tail())

Parameters

Parameter	Description
`interval`	Seconds between polls (default: 1.0)
`last_n`	Only fetch the last N rows (None for all)
`distinct_by`	Function to extract a key for deduplication
`max_polls`	Stop after this many polls
`fire_immediately`	Poll immediately or wait one interval first (default: True)
`jitter`	Random delay (0 to jitter) added to each sleep
`until`	Predicate function; stop when it returns True
`on_error`	Handler called with the exception when a poll fails

Polling errors are never raised. A failed poll is skipped and polling continues on the next tick, so pass an on_error handler to log failures, or re-raise inside it to stop polling.

Background thread polling

Use start_polling_thread when you can’t block the main thread (GUI, acquisition loop):

collected = []


def on_result(result):
    print(f"Thread received {len(result)} rows")
    collected.append(result)


stop_event = start_polling_thread(
    client,
    data_id=data_id,
    interval=10.0,
    last_n=10,
    max_polls=5,
    distinct_by=latest_frame,
    on_result=on_result,
)

# Later, to stop polling:
# stop_event.set()

The function returns a threading.Event that you can set to stop polling gracefully.

Async polling

Two helpers integrate with asyncio:

Async iterator

aiter_poll yields results without blocking the event loop:

import asyncio


async def stream_updates():
    async for result in aiter_poll(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=10,
        distinct_by=latest_frame,
        max_polls=3,
    ):
        print(f"Async poll received {len(result)} rows")
        print(result.tail())


asyncio.run(stream_updates())

Background task

start_polling_task creates a background task that invokes an optional async handler:

async def handle_async(result):
    print(f"Task handler received {len(result)} rows")


async def main():
    task = start_polling_task(
        client,
        data_id=data_id,
        interval=5.0,
        last_n=5,
        max_polls=3,
        distinct_by=latest_frame,
        on_result=handle_async,
    )
    await task


asyncio.run(main())

Deduplication

The distinct_by parameter accepts a function that extracts a hashable key from each result. Only results with a new key are yielded, so you won’t process the same data twice when polling faster than updates arrive. Key on the newest Frame Number in the index. Frame numbers are absolute and present on every row, so the maximum grows each time new frames are analyzed. Avoid keying on the last row (df.iloc[-1]): rows are grouped by view, so the last row is not necessarily the newest frame in a rotating video. UNIX Timestamp can also be 0 for every row when the recording has no absolute start time.

def latest_frame(df):
    if df.empty or "Frame Number" not in df.index.names:
        return None
    return df.index.get_level_values("Frame Number").max()

Drift correction

All polling utilities use drift-corrected scheduling to maintain the requested cadence even if individual polls are slow. Missed ticks are skipped rather than accumulated.

Introduction

Guides

References

Setup

Synchronous polling

Parameters

Background thread polling

Async polling

Async iterator

Background task

Deduplication

Drift correction

Next steps

Stream RHEED

Client Reference

​Setup

​Synchronous polling

​Parameters

​Background thread polling

​Async polling

​Async iterator

​Background task

​Deduplication

​Drift correction

​Next steps

Stream RHEED

Client Reference

Setup

Synchronous polling

Parameters

Background thread polling

Async polling

Async iterator

Background task

Deduplication

Drift correction

Next steps