remotivelabs.broker.recording_session

@dataclass
class File:
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
class PlaybackCommand(enum.Enum):

An enumeration.

PLAYBACK_PLAY = <PlaybackCommand.PLAYBACK_PLAY: 0>

Play a file.

PLAYBACK_PAUSE = <PlaybackCommand.PLAYBACK_PAUSE: 1>

Pause playback.

PLAYBACK_SEEK = <PlaybackCommand.PLAYBACK_SEEK: 2>

Seek to offset but keep current state.

PLAYBACK_CLOSE = <PlaybackCommand.PLAYBACK_CLOSE: 3>

Stop and close playback.

@dataclass
class PlaybackOffset:

Current offset in micro seconds.

PlaybackOffset(offset_time: int = 0)
offset_time: int = 0
@dataclass
class PlaybackRepeat:

Playback repeat settings.

PlaybackRepeat(cycle_start_time: int = 0, cycle_end_time: int = 0)
cycle_start_time: int = 0

Current cycle start in micro seconds.

cycle_end_time: int = 0

Current cycle end in micro seconds.

class PlaybackStatus(enum.Enum):

An enumeration.

PLAYBACK_PLAYING = <PlaybackStatus.PLAYBACK_PLAYING: 0>

Playing a file.

PLAYBACK_PAUSED = <PlaybackStatus.PLAYBACK_PAUSED: 1>

Playback is paused.

PLAYBACK_CLOSED = <PlaybackStatus.PLAYBACK_CLOSED: 2>

Playback is closed.

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

TODO: We probably dont want to inherit from BrokerClient, but rather use composition to hide functionality not relevant for recording session operations. However, this will do for now.

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.
def playback_status( self) -> AsyncIterator[list[RecordingSessionPlaybackStatus]]:

Get continuous status of all open recording sessions.

async def perform_playback( self, command: RecordingSessionPlaybackCommand) -> RecordingSessionPlaybackStatus | RecordingSessionPlaybackError:

Perform playback command.

@dataclass
class RecordingSessionPlaybackCommand:
RecordingSessionPlaybackCommand( path: str, command: PlaybackCommand, offset: PlaybackOffset | None = None)
path: str
command: PlaybackCommand
offset: PlaybackOffset | None = None
@dataclass
class RecordingSessionPlaybackError:
RecordingSessionPlaybackError(error_message: str)
error_message: str
@dataclass
class RecordingSessionPlaybackStatus:
RecordingSessionPlaybackStatus( path: str, status: PlaybackStatus, offset: PlaybackOffset, repeat: PlaybackRepeat | None = None)
path: str
status: PlaybackStatus
offset: PlaybackOffset
repeat: PlaybackRepeat | None = None