The RHEEDStreamer class provides high-performance streaming of RHEED frames to the Atomscale platform. It bridges Python to a Rust/PyO3 backend for efficient, concurrent packaging and upload of grayscale frames.

Constructor

from atomscale.streaming import RHEEDStreamer

streamer = RHEEDStreamer(
    api_key="YOUR_API_KEY",
    endpoint="https://api.atomscale.ai/",
    verbosity=2,
)

str

required

API key for authentication.

str | None

default:"None"

Root API endpoint. If not provided, uses the default production endpoint or reads from AS_API_ENDPOINT environment variable.

int | None

default:"None"

Logging verbosity level. Set to 4 for detailed progress output.

initialize()

Start a new streaming session and return a data ID for subsequent operations.

data_id = streamer.initialize(
    fps=120.0,
    rotations_per_min=15.0,
    chunk_size=240,
    stream_name="Growth run 42",
    physical_sample="GaN-001",
)

Parameters

float

required

Frame rate of the capture in frames per second.

float

required

Sample rotation speed in rotations per minute. Set to 0.0 for stationary samples.

int

required

Number of frames per upload chunk. Must be at least 2 * fps, so each chunk covers 2 seconds or more of video.

str | None

default:"None"

Human-readable name for this stream. Helps identify runs in the UI.

str | None

default:"None"

Name of the physical sample to link this stream to. Names are matched case-insensitively. If no match is found, a new sample record is created.

str | None

default:"None"

UUID of the project to associate with this stream. When provided along with physical_sample, the sample is added to the project’s tracking list and marked as the project’s active tracking sample for growth monitoring.

list[str] | None

default:"None"

Tag names or UUIDs to attach to the data item. Names are matched case-insensitively against existing organization tags, and unknown names are created. UUIDs must reference existing tags.

Returns

str

Unique identifier for this streaming session. Use this ID with push(), run(), and finalize().

Example

data_id = streamer.initialize(
    fps=120.0,
    rotations_per_min=15.0,
    chunk_size=240,
    stream_name="MBE growth - GaN on sapphire",
    physical_sample="GaN-001",
)
print(f"Started stream: {data_id}")

run()

Stream frames using generator/pull mode. The streamer consumes chunks from an iterator and handles pacing and retry logic.

streamer.run(data_id, frame_iterator)

Parameters

str

required

The data ID returned by initialize().

Iterable[NDArray[np.uint8]]

required

Iterator yielding frame chunks. Each chunk should be a numpy array of shape (N, H, W) with dtype uint8, where N is the chunk size. A single (H, W) frame is treated as N = 1.A yielded item may instead be a (frames, capture_start_ms_utc) tuple, pairing the array with an explicit capture-start timestamp in milliseconds since the UNIX epoch (UTC). Use this when you have hardware or OS-level timestamps for each chunk.

Returns

None. Frames are uploaded as they are yielded by the iterator. The call blocks until every spawned upload task completes.

Example

import numpy as np
import time


def frame_chunks(frames, *, chunk_size=240, fps=120.0):
    """Yield frame chunks at the capture cadence."""
    seconds_per_chunk = chunk_size / fps
    for start in range(0, len(frames), chunk_size):
        yield frames[start : start + chunk_size]
        time.sleep(seconds_per_chunk)


# Buffered frames (e.g., loaded from disk)
frames = np.random.randint(0, 256, size=(1200, 300, 500), dtype=np.uint8)

data_id = streamer.initialize(fps=10.0, rotations_per_min=0.0, chunk_size=20)
streamer.run(data_id, frame_chunks(frames, chunk_size=20, fps=10.0))
streamer.finalize(data_id)

push()

Upload a single chunk of frames in callback/push mode. Use this when frames arrive live from the instrument.

streamer.push(data_id, chunk_idx, frames)

Parameters

str

required

The data ID returned by initialize().

int

required

Zero-based index of this chunk. Chunks must be pushed in order.

NDArray[np.uint8]

required

Numpy array of frames with shape (N, H, W) or (H, W) for single frames. Must have dtype uint8.

int | None

default:"None"

Capture-start timestamp for the first frame in the chunk, in milliseconds since the UNIX epoch (UTC). When omitted, the streamer samples the current time on entry to push().

Returns

None. The chunk is queued for upload.

Timestamps

Without an explicit capture_start_ms_utc, each chunk is stamped with the wall clock sampled the moment push() is entered, before any packaging work. The chunk’s end timestamp is then start + (n / fps) * 1000. Inter-chunk gaps therefore reflect real arrival jitter, while the span within a chunk follows the declared FPS. Pass an explicit timestamp when you have a hardware clock, such as a camera trigger or an OS monotonic-to-UTC conversion.

Example

import numpy as np
import time

fps = 120.0
chunk_size = 240
seconds_per_chunk = chunk_size / fps

data_id = streamer.initialize(
    fps=fps,
    rotations_per_min=15.0,
    chunk_size=chunk_size,
)

for chunk_idx in range(10):
    # Capture frames from instrument
    frames = np.random.randint(0, 256, size=(chunk_size, 300, 500), dtype=np.uint8)
    streamer.push(data_id, chunk_idx, frames)
    time.sleep(seconds_per_chunk)

time.sleep(1.0)  # Let in-flight uploads finish
streamer.finalize(data_id)

finalize()

Complete the streaming session and signal the server to begin final processing.

streamer.finalize(data_id)

Parameters

str

required

The data ID returned by initialize().

Returns

None. The stream is marked as complete and queued for final analysis.

Always call finalize() even if the upload fails part-way. This signals the server to clean up resources and mark the stream appropriately.

Example

try:
    data_id = streamer.initialize(fps=120.0, rotations_per_min=15.0, chunk_size=240)
    # ... push or run frames ...
finally:
    streamer.finalize(data_id)

Frame Format Requirements

RHEED frames must meet these requirements:

Property	Requirement
Data type	`numpy.uint8` (8-bit unsigned integer)
Shape	`(N, H, W)` for chunks or `(H, W)` for single frames
Color	Grayscale only
Values	0-255 intensity range

Example frame preparation

import numpy as np

# From a color image
color_frame = capture_from_camera()  # Shape: (H, W, 3)
gray_frame = np.mean(color_frame, axis=2).astype(np.uint8)

# From 16-bit data
raw_frame = capture_raw()  # Shape: (H, W), dtype: uint16
normalized = ((raw_frame / raw_frame.max()) * 255).astype(np.uint8)

# Batch into chunks
chunk_size = 240
frames = np.stack([capture_frame() for _ in range(chunk_size)])  # (240, H, W)

Best Practices

Recommendation	Rationale
Match the capture cadence	The server expects frames at the declared FPS
Use chunk sizes ≥ 2 seconds	Reduces network overhead and improves throughput
Always call `finalize()`	Allows graceful cleanup even on failure
Use distinct stream names	Makes runs easier to find later
Link to physical samples	Enables cross-run analysis and tracking

Error Handling

The streamer raises exceptions for connection failures and invalid parameters. Wrap streaming code in try/finally to ensure finalize() is called:

data_id = None
try:
    data_id = streamer.initialize(
        fps=120.0,
        rotations_per_min=15.0,
        chunk_size=240,
    )

    for chunk_idx, frames in enumerate(capture_frames()):
        streamer.push(data_id, chunk_idx, frames)

except Exception as e:
    print(f"Streaming failed: {e}")
finally:
    if data_id:
        streamer.finalize(data_id)

TimeseriesStreamer

The TimeseriesStreamer class streams scalar instrument data (temperature, pressure, growth rate, etc.) to the Atomscale platform. Like RHEEDStreamer, it uses a Rust/PyO3 backend.

View source on GitHub

Constructor

from atomscale.streaming import TimeseriesStreamer

streamer = TimeseriesStreamer(
    api_key="YOUR_API_KEY",
    points_per_chunk=100,
    verbosity=2,
)

str

required

API key for authentication.

str | None

default:"None"

Root API endpoint. If not provided, uses the default production endpoint or reads from AS_API_ENDPOINT environment variable.

int

default:"100"

Expected number of data points per chunk. Used for array positioning.

int | None

default:"None"

Logging verbosity level. Set to 4 for detailed progress output.

initialize()

Start a new timeseries streaming session.

data_id = streamer.initialize(
    stream_name="MBE growth - temperature log",
    synth_source_id=42,
    physical_sample="GaN-001",
    project_id="project-uuid",
)

Parameters

str | None

default:"None"

Human-readable name for this stream.

int | None

default:"None"

Growth instrument ID to link. Must belong to your organization.

str | None

default:"None"

Name or UUID of a physical sample to associate with the stream. Names are matched case-insensitively, or a new sample is created if no match is found.

str | None

default:"None"

list[str] | None

default:"None"

Tag names or UUIDs to attach to the data item. Names are matched case-insensitively against existing organization tags, and unknown names are created. UUIDs must reference existing tags.

Returns

str

Unique identifier for this streaming session.

push()

Upload a single chunk of data for one channel. Spawns an async upload task and returns immediately.

streamer.push(
    data_id,
    chunk_index=0,
    channel_name="temperature",
    timestamps=[1700000000.0, 1700000000.1],
    values=[25.0, 25.1],
    units="C",
)

Parameters

str

required

The data ID returned by initialize().

int

required

Zero-based index of this chunk. Chunks must be pushed in order.

str

required

Name of the data channel (e.g., “temperature”, “pressure”).

list[float]

required

Unix epoch timestamps in seconds.

list[float]

required

Measured values corresponding to each timestamp.

str | None

default:"None"

Optional units for the values (e.g., “C”, “mbar”).

Returns

None. The chunk is queued for upload.

push_multi()

Upload data for multiple channels in a single call.

streamer.push_multi(data_id, chunk_index=0, channels={
    "temperature": {
        "timestamps": [1700000000.0, 1700000000.1],
        "values": [25.0, 25.1],
        "units": "C",
    },
    "pressure": {
        "timestamps": [1700000000.0, 1700000000.1],
        "values": [1.0, 1.1],
        "units": "mbar",
    },
})

Parameters

str

required

The data ID returned by initialize().

int

required

Zero-based index of this chunk.

dict[str, dict]

required

Mapping of channel names to channel data. Each value dict should contain timestamps (list of floats), values (list of floats), and optionally units (string).

Returns

None.

run()

Stream data from an iterator. Blocks until all uploads complete.

def data_chunks():
    for i in range(10):
        timestamps = [time.time() + j * 0.1 for j in range(100)]
        values = [25.0 + j * 0.01 for j in range(100)]
        yield (timestamps, values)

streamer.run(
    data_id,
    channel_name="temperature",
    data_iter=data_chunks(),
    units="C",
)

Parameters

str

required

The data ID returned by initialize().

str

required

Name of the data channel.

Iterable[tuple[list[float], list[float]]]

required

Iterator yielding (timestamps, values) tuples for each chunk.

str | None

default:"None"

Optional units for the values.

Returns

None.

finalize()

Complete the streaming session and signal the server to begin processing.

streamer.finalize(data_id)

Parameters

str

required

The data ID returned by initialize().

Returns

None.

Always call finalize() even if the upload fails part-way. This signals the server to clean up resources and mark the stream appropriately.

​Constructor

​initialize()

​Parameters

​Returns

​Example

​run()

​Parameters

​Returns

​Example

​push()

​Parameters

​Returns

​Timestamps

​Example

​finalize()

​Parameters

​Returns

​Example

​Frame Format Requirements

​Example frame preparation

​Best Practices

​Error Handling

​TimeseriesStreamer

​Constructor

​initialize()

​Parameters

​Returns

​push()

​Parameters

​Returns

​push_multi()

​Parameters

​Returns

​run()

​Parameters

​Returns

​finalize()

​Parameters

​Returns

Constructor

initialize()

Parameters

Returns

Example

run()

Parameters

Returns

Example

push()

Parameters

Returns

Timestamps

Example

finalize()

Parameters

Returns

Example

Frame Format Requirements

Example frame preparation

Best Practices

Error Handling

TimeseriesStreamer

Constructor

initialize()

Parameters

Returns

push()

Parameters

Returns

push_multi()

Parameters

Returns

run()

Parameters

Returns

finalize()

Parameters

Returns