remotivelabs.broker.recording_session

There is a RecordingSessionClient for managing recording session playback on the broker.

import asyncio

from remotivelabs.broker.recording_session import RecordingSessionClient


async def main() -> None:
    # Create a recording session client
    async with RecordingSessionClient(url="http://127.0.0.1:50051") as rs_client:
        # List available recording files on the broker
        files = await rs_client.list_recording_files()
        print(f"Available recording files: {files}")

        # open a recording session with the first file
        rs_ref = rs_client.get_session(str(files[0]))
        async with rs_ref as session:
            # start playback
            status = await session.play()
            print(f"Playback started with status: {status}")

        # exiting the async with block will close the session, but playback will continue if other sessions are open


if __name__ == "__main__":
    asyncio.run(main())
@dataclass
class File:

Represents a file or folder on the broker's file system.

File( path: str, type: remotivelabs.broker.recording_session.file.FileType, created_time: datetime.datetime, modified_time: datetime.datetime, size: int)
path: str
type: remotivelabs.broker.recording_session.file.FileType
created_time: datetime.datetime
modified_time: datetime.datetime
size: int
@dataclass
class PlaybackRepeat:

Playback repeat settings.

PlaybackRepeat(start_offset: int = 0, end_offset: int = 0)
start_offset: int = 0

Current cycle start in micro seconds.

end_offset: int = 0

Current cycle end in micro seconds.

class PlaybackMode(enum.Enum):

Playback modes for a recording session.

PLAYBACK_PLAYING = <PlaybackMode.PLAYBACK_PLAYING: 0>

Playing a file.

PLAYBACK_PAUSED = <PlaybackMode.PLAYBACK_PAUSED: 1>

Playback is paused.

PLAYBACK_CLOSED = <PlaybackMode.PLAYBACK_CLOSED: 2>

Playback is closed.

class RecordingSession:

Represents a recording session playback on the broker.

RecordingSession( client: RecordingSessionClient, path: str, force_reopen: bool = False)
path
async def open( self, force_reopen: bool = False) -> RecordingSessionPlaybackStatus:

Open recording session for playback. If force_reopen is True, close any existing session before opening.

Arguments:
  • force_reopen: Whether to force close any existing session before opening.
Returns:

The playback status after opening the session.

Raises:
  • RecordingSessionPlaybackError: On failures to open the session.
async def play( self, offset: int | None = None) -> RecordingSessionPlaybackStatus:

Start or continue playback of an open recording session.

Arguments:
  • offset: The offset in microseconds from which to start playback. If None, playback continues from the current position.
Returns:

The playback status after starting playback.

Raises:
  • RecordingSessionPlaybackError: On playback errors
async def pause( self, offset: int | None = None) -> RecordingSessionPlaybackStatus:

Pause playback of an open recording session.

Arguments:
  • offset: Target position in microseconds at where the playback cursor is put after pausing.
Returns:

The playback status after pausing playback.

Raises:
  • RecordingSessionPlaybackError: On playback errors
async def seek( self, offset: int) -> RecordingSessionPlaybackStatus:

Move playback to a target offset within the currently open recording session.

Arguments:
  • offset: Target position in microseconds.
Returns:

The playback status after seeking.

Raises:
  • RecordingSessionPlaybackError: On playback errors
async def close( self) -> RecordingSessionPlaybackStatus:

Close an open recording session.

async def set_repeat( self, start_offset: int | None = None, end_offset: int | None = None) -> RecordingSessionPlaybackStatus:

Set repeat mode for the currently open recording session.

Arguments:
  • start_offset: Start offset in microseconds for the repeat segment. If None, repeat from the start of the recording.
  • end_offset: End offset in microseconds for the repeat segment. If None, repeat until the end of the recording.
Note:

If both start_offset and end_offset are None, repeat mode will be removed.

Returns:

The playback status after setting repeat mode.

class RecordingSessionClient(remotivelabs.broker.client.BrokerClient):

Client for managing recording session playback on the broker.

async def list_recording_files( self, path: str | None = None) -> list[File]:

List recording files in a directory.

Arguments:
  • path: Optional path to the subdirectory containing the recording files.
Returns:

A list of File objects representing the recording files.

def playback_status( self) -> AsyncIterator[list[RecordingSessionPlaybackStatus]]:

Get continuous status of all open recording sessions.

Returns:

An async iterator yielding lists of RecordingSessionPlaybackStatus objects.

def get_session( self, path: str, force_reopen: bool = False) -> RecordingSession:

Return a RecordingSession for the given path.

Arguments:
  • path: The path to the recording session file.
  • force_reopen: Whether to force close any existing session before opening.
Returns:

A RecordingSession object.

class RecordingSessionPlaybackError(builtins.Exception):

Exception raised for errors during recording session playback.

@dataclass
class RecordingSessionPlaybackStatus:

Represents the playback status of a recording session.

RecordingSessionPlaybackStatus( path: str, mode: PlaybackMode, offset: int, repeat: PlaybackRepeat | None = None)
path: str
mode: PlaybackMode
offset: int
repeat: PlaybackRepeat | None = None