Python API Reference (monarch3)

The monarch3 Python module is a pybind11 binding of the Monarch3 C++ library. It exposes the same read/write workflow as the C++ API using Pythonic conventions: method and property names use snake_case, raw data buffers are returned as numpy.ndarray views, and files can be managed with Python’s with statement.

See How to use Monarch3 in Python for workflow examples and usage guidance.

Python bindings for the Monarch3 egg v3 file library

class monarch3.M3ChannelHeader

Bases: pybind11_object

Header information for a single data channel

property acquisition_rate
property bit_alignment
property bit_depth
property dac_gain
property data_format
property data_type_size
property frequency_min
property frequency_range
property number
property record_size
property sample_size
property source
property voltage_offset
property voltage_range
class monarch3.M3Header

Bases: pybind11_object

Egg file header: run metadata and stream/channel configuration

add_stream(*args, **kwargs)

Overloaded function.

  1. add_stream(self: monarch3.M3Header, source: str, acq_rate: typing.SupportsInt, rec_size: typing.SupportsInt, sample_size: typing.SupportsInt, data_type_size: typing.SupportsInt, data_format: typing.SupportsInt, bit_depth: typing.SupportsInt, bit_alignment: typing.SupportsInt, chan_vec: object = None) -> int

Add a single-channel stream; returns the stream number

  1. add_stream(self: monarch3.M3Header, source: str, n_channels: typing.SupportsInt, channel_format: typing.SupportsInt, acq_rate: typing.SupportsInt, rec_size: typing.SupportsInt, sample_size: typing.SupportsInt, data_type_size: typing.SupportsInt, data_format: typing.SupportsInt, bit_depth: typing.SupportsInt, bit_alignment: typing.SupportsInt, chan_vec: object = None) -> int

Add a multi-channel stream; returns the stream number

property channel_headers
property description
property egg_version
property filename
property n_channels
property n_streams
property run_duration
property stream_headers
property timestamp
class monarch3.M3Record

Bases: pybind11_object

A single data record: ID, timestamp (ns), and raw data bytes

get_data(self: monarch3.M3Record, nbytes: SupportsInt) → numpy.typing.NDArray[numpy.uint8]

Return a writable numpy uint8 view of the data buffer. nbytes should equal stream.channel_record_n_bytes (or stream_record_n_bytes). Use numpy.ndarray.view(dtype) to reinterpret as a different element type.

property record_id

Record ID (uint64)

property time

Timestamp in nanoseconds since the start of the run (uint64)

class monarch3.M3Stream

Bases: pybind11_object

Read/write access for a single data stream

property acquisition_id

ID of the most recently accessed acquisition

property channel_record_n_bytes

Size in bytes of a channel record data buffer

property channel_record_size

Number of samples in a channel record

property data_type_size

Size in bytes of each sample element

get_channel_data(self: monarch3.M3Stream, channel: SupportsInt) → numpy.typing.NDArray[numpy.uint8]

Return a writable numpy uint8 view of the data buffer for the given channel. Call after read_record(). Use .view(dtype) to reinterpret elements.

get_channel_record(self: monarch3.M3Stream, channel: SupportsInt) → monarch3.M3Record

Return the record object for the given channel index

get_stream_data(self: monarch3.M3Stream) → numpy.typing.NDArray[numpy.uint8]

Return a writable numpy uint8 view of the full (interleaved) stream record. Call after read_record(). Use .view(dtype) to reinterpret elements.

get_stream_record(self: monarch3.M3Stream) → monarch3.M3Record

Return the stream-level record object (all channels interleaved)

property is_interleaved

True if multi-channel data is stored interleaved in the file

property n_acquisitions

Number of acquisitions in this stream (valid after ReadRecord or after writing)

property n_channels

Number of channels in this stream

property n_records_in_acquisition

Number of records in the current acquisition

property n_records_in_file

Total number of records across all acquisitions in the file

read_record(self: monarch3.M3Stream, offset: SupportsInt = 0, if_new_acq_start_at_first_rec: bool = True) → bool

Read a record from the file.

If the last record read was [J], reads record [J+1+offset]. offset=0 (default): next record; offset=-1: reread current; offset<-1: step backward; offset>0: skip forward. Returns True on success, False when past end/start of file. Raises Monarch3Exception on error.

property record_count_in_acq

Number of records read/written in the current acquisition

property sample_size

Number of elements per sample (1 for real, 2 for complex)

property stream_record_n_bytes

Size in bytes of the full stream record data buffer

property stream_record_size

Number of samples in the full (interleaved) stream record

write_record(self: monarch3.M3Stream, is_new_acquisition: bool) → bool

Write the current record contents to the file.

is_new_acquisition=True starts a new acquisition group; is_new_acquisition=False continues the current acquisition. Returns True on success. Raises Monarch3Exception on error.

class monarch3.M3StreamHeader

Bases: pybind11_object

Header information for a single data stream

property acquisition_rate
property bit_alignment
property bit_depth
property channel_format
property channels
property data_format
property data_type_size
property n_acquisitions
property n_channels
property n_records
property number
property record_size
property sample_size
property source
class monarch3.Monarch3

Bases: pybind11_object

Top-level egg v3 file handle.

Use open_for_reading() or open_for_writing() to obtain an instance. Supports the context manager protocol (‘with’ statement).

finish_reading(self: monarch3.Monarch3) → None

Close the file after reading.

finish_writing(self: monarch3.Monarch3) → None

Flush and close the file after writing.

get_header(*args, **kwargs)

Overloaded function.

  1. get_header(self: monarch3.Monarch3) -> monarch3.M3Header

Return the file header (read access)

  1. get_header(self: monarch3.Monarch3) -> monarch3.M3Header

Return the file header (write access)

get_stream(*args, **kwargs)

Overloaded function.

  1. get_stream(self: monarch3.Monarch3, stream: typing.SupportsInt) -> monarch3.M3Stream

Return the stream object for the given stream index (read access)

  1. get_stream(self: monarch3.Monarch3, stream: typing.SupportsInt) -> monarch3.M3Stream

Return the stream object for the given stream index (write access)

static open_for_reading(filename: str) → monarch3.Monarch3

Open an existing egg file for reading. Returns a Monarch3 instance in the eOpenToRead state.

static open_for_writing(filename: str) → monarch3.Monarch3

Create or overwrite an egg file for writing. Returns a Monarch3 instance in the eOpenToWrite state.

read_header(self: monarch3.Monarch3) → None

Read header information from the file. Must be called after open_for_reading() before accessing data.

property state

Current state of the file handle

write_header(self: monarch3.Monarch3) → None

Write the header to file and prepare streams for writing. Must be called after configuring the header and adding streams.

exception monarch3.Monarch3Exception

Bases: Exception