Frame module#

Sub-package about frame organization.

class galaxy_beep_frame.frame.Answer(state: FrameState, continuation_indicator: ContinuationIndicator, answer_number: int)#

Answer type.

build(size: int) → bytes#: Build the header of this Frame.

class galaxy_beep_frame.frame.ContinuationIndicator(*values)#: Possible Continuation Indicator.

class galaxy_beep_frame.frame.Frame(payload: dataclasses.InitVar[bytes], channel_number: dataclasses.InitVar[int], message_number: dataclasses.InitVar[int], sequence_number: dataclasses.InitVar[int], answer_number: dataclasses.InitVar[int | None], continuation_indicator: dataclasses.InitVar[ForwardRef('ContinuationIndicator', is_class=True, owner=<class 'galaxy_beep_frame.frame.base.Frame'>)], possible_header_type: dataclasses.InitVar[ForwardRef('PossibleHeaderType', is_class=True, owner=<class 'galaxy_beep_frame.frame.base.Frame'>)])#

A frame of the BEEP protocol.

Notes#

See https://datatracker.ietf.org/doc/html/rfc3080.

static async_read(reader: StreamReader, chunk_size: int = 100) → AsyncIterator[__annotationlib_name_1__]#

Generate Frame object asynchronously.

Parameters#

reader: StreamReader: An StreamReader of bytes from asyncio containing the frames.
chunk_size: int: Size of each chunk to read.

build() → bytes#: Build a Frame.

static read(reader: Reader[__annotationlib_name_1__], chunk_size: int = 100) → Iterator[__annotationlib_name_2__]#

Generate Frame object on the fly from a stream of bytes.

Parameters#

reader: Reader[bytes]: Stream of bytes containing the frames.
chunk_size: int: Size of each chunk to read.

write(writer: Writer[__annotationlib_name_1__]) → None#

Write this Frame into a stream of bytes.

Parameters#

writer: Writer[bytes]: Stream of bytes to write the frame on.

class galaxy_beep_frame.frame.FrameState(frame: Frame, payload: bytes, channel_number: int, message_number: int, sequence_number: int, answer_number: dataclasses.InitVar[int | None], continuation_indicator: dataclasses.InitVar[ForwardRef('ContinuationIndicator', is_class=True, owner=<class 'galaxy_beep_frame.frame.state.FrameState'>)], possible_header_type: dataclasses.InitVar[ForwardRef('PossibleHeaderType', is_class=True, owner=<class 'galaxy_beep_frame.frame.state.FrameState'>)])#

State of a BEEP frame.

write(writer: Writer[__annotationlib_name_1__]) → None#

Write this Frame state.

Parameters#

writer: Writer[bytes]: Stream of bytes to write the frame on.

class galaxy_beep_frame.frame.HeaderType(state: FrameState)#

Manage data about header type (MSG, RPY, etc.).

build(size: int) → bytes#: Build the header of this Frame.

static make(state: FrameState, continuation_indicator: ContinuationIndicator, possible_header_type: PossibleHeaderType, answer_number: int | None) → HeaderType#: Build HeaderType depending on argument.

class galaxy_beep_frame.frame.Message(state: FrameState, continuation_indicator: ContinuationIndicator)#

Message type.

build(size: int) → bytes#: Build the header of this Frame.

class galaxy_beep_frame.frame.Null(state: FrameState)#

Null type.

build(size: int) → bytes#: Build the header of this Frame.

class galaxy_beep_frame.frame.Reply(state: FrameState, continuation_indicator: ContinuationIndicator)#

Reply type.

build(size: int) → bytes#: Build the header of this Frame.