Module Contents

Classes

Data

logger

API

1
stanza.logger.session.logger

Value: getLogger(...)

1
class stanza.logger.session.SweepContext(session: stanza.logger.session.LoggerSession, name: str, x_label: str | list[str], y_label: str, metadata: dict[str, typing.Any], routine_name: str | None = None)

Context manager for streaming sweep data with live plotting.

Accumulates data points via append() calls, streams to live plot writers immediately, and writes the complete sweep to file writers on context exit.

Supports 1D sweeps (scalar x, scalar y) and 2D sweeps ([x0, x1], y).

Example:

1D sweep

with session.sweep(“signal”, “Time”, “Amplitude”) as s: … for x, y in data_stream: … s.append([x], [y]) # Live plot updates

2D sweep

with session.sweep(“heatmap”, [“V1”, “V2”], “Current”) as s: … for v1, v2 in voltage_pairs: … s.append([v1, v2], [current]) # Live heatmap updates

Sweep written to files here

Initialization

Initialize sweep context.

Parameters:

Parameters:

• session: Parent LoggerSession
• name: Sweep name
• x_label: X-axis label (string for 1D, list for 2D)
• y_label: Y-axis label
• metadata: User metadata dict
• routine_name: Optional routine name

1
append(
2
    x_data: list[float] | numpy.ndarray, y_data: list[float] | numpy.ndarray
3
) -> None

Append data points and stream to live plots.

1D: append([x], [y]) or append([x1, x2, …], [y1, y2, …]) 2D: append([x, y], [value])

1
end() -> None

Write accumulated sweep to file writers.

1
cancel() -> None

Cancel without persisting. Live updates remain visible.

1
__enter__() -> stanza.logger.session.SweepContext

Enter context manager.

1
__exit__(
2
    exc_type: type[BaseException] | None,
3
    exc_value: BaseException | None,
4
    traceback: typing.Any
5
) -> None

Exit context manager: end() on success, cancel() on exception.

1
class stanza.logger.session.LoggerSession(metadata: stanza.logger.datatypes.SessionMetadata, writer_pool: dict[str, stanza.logger.writers.base.AbstractDataWriter], writer_refs: list[str], base_dir: str | pathlib.Path, buffer_size: int = 1000, auto_flush_interval: float | None = 30.0)

Session for the logger.

1
session_id: str

1
_check_buffer_size_warning() -> None

Check if buffer has grown too large and log warning once.

1
initialize() -> None

Initialize the session.

Raises:

LoggerSessionError: If session is already initialized

1
finalize() -> None

Finalize the session.

1
close() -> None

Close the session (alias for finalize()).

1
log_measurement(
2
    name: str,
3
    data: dict[str, typing.Any],
4
    metadata: dict[str, typing.Any] | None = None,
5
    routine_name: str | None = None
6
) -> None

Log measurement data to a session.

1
log_analysis(
2
    name: str,
3
    data: dict[str, typing.Any],
4
    metadata: dict[str, typing.Any] | None = None,
5
    routine_name: str | None = None
6
) -> None

Log analysis data to a session.

1
log_sweep(
2
    name: str,
3
    x_data: list[float] | list[list[float]] | numpy.ndarray,
4
    y_data: list[float] | numpy.ndarray,
5
    x_label: str,
6
    y_label: str,
7
    metadata: dict[str, typing.Any] | None = None
8
) -> None

Log sweep data to a session.

1
log_parameters(parameters: dict[str, typing.Any]) -> None

Log parameters to a session.

1
sweep(
2
    name: str,
3
    x_label: str | list[str],
4
    y_label: str,
5
    metadata: dict[str, typing.Any] | None = None
6
) -> stanza.logger.session.SweepContext

Create streaming sweep context for live plotting.

Accumulates data via append(), writes complete sweep to files on exit. Supports 1D and 2D data. Exception in context = no file write. One active sweep per name; HDF5 auto-uniquifies sequential same-name sweeps.

Example:

1D sweep

with session.sweep(“signal”, “Time (s)”, “Amplitude”) as s: for x, y in data_stream: s.append([x], [y])

2D sweep

with session.sweep(“heatmap”, [“V1 (V)”, “V2 (V)”], “Current (A)”) as s: for v1, v2 in voltage_pairs: s.append([v1, v2], [current])

1
_should_flush() -> bool

Check if buffer should be flushed.

1
_stream_live_sweep_chunk(
2
    name: str,
3
    x: numpy.ndarray,
4
    y: numpy.ndarray,
5
    x_label: str | list[str],
6
    y_label: str,
7
    dim: int
8
) -> None

Stream data chunk to live plot writers only.

1
_write_completed_sweep(sweep_data: stanza.logger.datatypes.SweepData) -> None

Write complete sweep to file writers only (skip live/bokeh).

1
flush() -> None

Flush buffered data to all writers.

Raises:

LoggerSessionError: If any writer fails to write or flush data

1
__enter__() -> stanza.logger.session.LoggerSession

Enter the session context.

1
__exit__(
2
    exc_type: type[BaseException] | None,
3
    exc_value: BaseException | None,
4
    traceback: typing.Any
5
) -> None

Exit the session context.

1
__repr__() -> str