nanover.recording.reading module

class nanover.recording.reading.FrameRecordingEvent(*, timestamp: int | None, message: dict[str, Any], prev_frame: nanover.trajectory.frame_data.FrameData, next_frame: nanover.trajectory.frame_data.FrameData)

Bases: MessageEvent

classmethod make_empty()

next_frame: FrameData

prev_frame: FrameData

class nanover.recording.reading.MessageEvent(*, timestamp: int | None, message: dict[str, Any])

Bases: object

message: dict[str, Any]

timestamp: int | None

class nanover.recording.reading.MessageZipReader(zipfile: ZipFile)

Bases: object

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_message_from_entry(entry: RecordingIndexEntry) → dict[str, Any]

class nanover.recording.reading.NanoverRecordingReader(zipfile: ZipFile)

Bases: MessageZipReader

get_frame_from_entry(entry: RecordingIndexEntry) → FrameData | None

get_message_type_from_entry(entry: RecordingIndexEntry, type: str) → Any | None

get_state_from_entry(entry: RecordingIndexEntry) → DictionaryChange | None

iter_frame_updates()

iter_frames_full()

iter_max(): Iterate recording yielding recording events in timestamp order, with each event containing the full information of previous frame, previous state, current message, next frame, and next state.

iter_message_events()

iter_state_updates()

iter_states_full()

class nanover.recording.reading.RecordingEvent(*, prev_frame_event: nanover.recording.reading.FrameRecordingEvent, prev_state_event: nanover.recording.reading.StateRecordingEvent, next_frame_event: nanover.recording.reading.FrameRecordingEvent | None, next_state_event: nanover.recording.reading.StateRecordingEvent | None)

Bases: object

property next_frame

next_frame_event: FrameRecordingEvent | None

property next_state

next_state_event: StateRecordingEvent | None

property prev_frame

prev_frame_event: FrameRecordingEvent

property prev_state

prev_state_event: StateRecordingEvent

property timestamp: int | None

class nanover.recording.reading.RecordingIndexEntry(*, offset: int, length: int, metadata: dict[str, typing.Any] = <factory>)

Bases: object

length: int

metadata: dict[str, Any]

offset: int

read_from(io: IO[bytes])

class nanover.recording.reading.StateRecordingEvent(*, timestamp: int | None, message: dict[str, Any], prev_state: dict, next_state: dict)

Bases: MessageEvent

classmethod make_empty()

next_state: dict

prev_state: dict

nanover.recording.reading.iter_full_view(path: PathLike[str])

nanover.recording.reading.iter_recording_file(path: PathLike[str])

nanover.recording.reading.message_events_from_recording(path: PathLike[str])

nanover.recording.reading.parse_index(zipfile: ZipFile)

nanover.recording.reading.split_by_simulation_counter(path: PathLike[str])