nanover.recording.reading module

exception nanover.recording.reading.InvalidMagicNumber

Bases: Exception

The magic number read from a file is not the expected one.

The file may not be in the correct format, or it may be corrupted.

class nanover.recording.reading.MessageRecordingReader(io: BinaryIO)

Bases: object

Wraps a NanoVer recording of gRPC messages to provide fast and convenient random access.

close()

classmethod from_io(io: BinaryIO)

Read a recording from binary data source.

classmethod from_path(path: PathLike[str])

Read a recording from a filepath.

get_entry_at_index(index)

Return an entry for the nth message in the file.

Parameters:

index – Index of the message in the sequence of all messages in the data.

get_entry_at_offset(offset)

Read and return a message entry from a specific byte offset in the data.

Parameters:

offset – Offset in bytes to begin reading a message entry.

iter_messages(message_type: Callable[[], TMessage])

reindex()

Read the file from beginning to end and recording the byte position of each message for later access.

class nanover.recording.reading.Parseable(*args, **kwargs)

Bases: Protocol

ParseFromString(_: bytes) → bytes

class nanover.recording.reading.RecordingFileEntry(*, offset: int, timestamp: float, buffer: bytes)

Bases: object

buffer: bytes

offset: int

timestamp: float

exception nanover.recording.reading.UnsupportedFormatVersion(format_version: int, supported_format_versions: tuple[int])

Bases: Exception

The version of the file format is not supported by the parser.

nanover.recording.reading.buffer_to_frame_message(buffer)

nanover.recording.reading.buffer_to_message(buffer, message_type: Callable[[], TMessage])

nanover.recording.reading.buffer_to_state_message(buffer)

nanover.recording.reading.iter_buffers(io: BinaryIO)

nanover.recording.reading.iter_full_view(*, traj: PathLike[str] | None = None, state: PathLike[str] | None = None)

Iterate one or both of trajectory and state recording files, yield a timestamp and copies of both the current aggregate FrameData and the current aggregate state dictionary.

nanover.recording.reading.iter_recording_files(*, traj: PathLike[str] | None = None, state: PathLike[str] | None = None)

Iterate one or both of trajectory and state recording files, yield a timestamp and one or both of frame and update that occurred at that instant. Frame index is included in frame data under the key “index”.

nanover.recording.reading.iter_state_file(path)

Iterate over all state updates in a recording file.

Parameters:

path – Path of recording file to read from.

nanover.recording.reading.iter_state_recording(io: BinaryIO)

nanover.recording.reading.iter_trajectory_file(path)

Iterate over all frame updates in a recording file.

Parameters:

path – Path of recording file to read from.

nanover.recording.reading.iter_trajectory_recording(io: BinaryIO)

nanover.recording.reading.read(io: BinaryIO, count: int)

nanover.recording.reading.read_buffer(io: BinaryIO)

nanover.recording.reading.read_header(io: BinaryIO)

nanover.recording.reading.read_u128(io: BinaryIO)

nanover.recording.reading.read_u64(io: BinaryIO)

nanover.recording.reading.skip_buffer(io: BinaryIO)

nanover.recording.reading.skip_buffers(io: BinaryIO)

nanover.recording.reading.split_by_simulation_counter(*, traj: PathLike[str], state: PathLike[str] | None = None)

Split a trajectory recording (and optionally a corresponding state recording) into sequences that share the same simulation counter value.