Converter module#

Sub-package for conversion from bytes to Frame.

class galaxy_beep_frame.converter.Lexer(graph: Graph = <factory>, frame: Node[Token] | None = None, header: Node[Token] | None = None, size: int = 0, continuation_indicator: int | None = None, end: bytearray = <factory>, current_token: Token | None = None, state: State = State.START)#

Lexer convert a bytes into tokens graph.

Notes#

The graph structure is a hierarchical structure in the graph:

root (id: root) + frame (has frame)

  • header (has header) - message type (has message type) - channel number (has channel number) - message number (has message number) - continuation indicator (has continuation indicator) - sequence number (has sequence number) . answer number (has answer number)

  • payload

where: - + : many - - : exactly one - . : one or zero

Frame can be fetched from the graph. They are considered as completed when in a immutable state.

static connections_to_human(connections: set[Connection], indent: int = 1) → str#

Convert a list of connections to readable str rep.

feed(data: bytes) → list[Node[Token]]#

Lex bytes.

lex(reader: Reader[__annotationlib_name_1__], chunk_size: int = 100) → list[Node[Token]]#

Lex on stream of bytes.

Warnings#

This will keep every frames in memory!

lex_byte(char: int) → Node[Token] | None#

Read a byte with current context.

This function update context in function of the value of this byte.

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

Lex stream of bytes, yield frame node as they complete.

async lex_stream(reader: StreamReader, chunk_size: int = 100) → list[Node[Token]]#

Lex on async stream of bytes.

Warnings#

This will keep every frames in memory!

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

Lex async stream of bytes, yield node as they complete.

static node_to_human(node: Node[Token]) → str#

Convert a token node to a readable str repr.

class galaxy_beep_frame.converter.MutabilityState(token: Token)#

Information concerning mutability state.

is_mutable() → bool#

Inform about the mutability of the state.

class galaxy_beep_frame.converter.MutableState(token: Token)#

Mutable state.

end() → None#

Switch mutability of the token.

is_mutable() → Literal[True]#

Return the mutability of the state.

update(char: int) → None#

Update value stored in the token by appending a byte.

class galaxy_beep_frame.converter.Parser#

Convert a graph of token generated by lexer and validate it.

static parse(frames: list[Node[__annotationlib_name_1__]]) → list[Frame]#

Parse a list of frames.

Parameters#

frames: list[Node[Token]]

Frames to convert.

static parse_frame(frame: Node[__annotationlib_name_1__]) → Frame#

Convert a node from a graph to a Frame.

Parameters#

frame: Node[Token]

Frame to convert.

class galaxy_beep_frame.converter.State(*values)#

Possible Lexer state.

class galaxy_beep_frame.converter.Token(type: TokenType, value: bytearray = <factory>)#

Hold information about current state and value.

Start mutable (eg. can be modified) and can become irreversably immutable.

end() → None#

Update value stored in the token by appending a byte.

Notes#

After this method is called, the token will not be updatable anymore.

update(char: int) → None#

Update value stored in the token by appending a byte.

Notes#

This token should not be finished.

class galaxy_beep_frame.converter.TokenType(*values)#

Possible token tyes.