pyg90alarm.cloud.protocol

Protocol implementation for G90 cloud communication.

This module defines the base classes and structures used for encoding and decoding messages that flow between G90 alarm devices and their cloud servers.

Combination of Python dataclasses and struct is used to define the protocol messages. The dataclass decorator is used to define the message classes, while the struct module is used to define the binary format of the messages.

Classes

`G90CloudBase`(context)	Base class for G90 cloud protocol messages.
`G90CloudHeader`(context, command, _source, ...)	Header for G90 cloud protocol messages.
`G90CloudHeaderVersioned`(context, command, ...)	Versioned header for G90 cloud protocol messages.
`G90CloudMessage`(context)	Base class for G90 cloud protocol messages with headers.
`G90CloudMessageContext`(local_ip, local_port, ...)	Context for G90 cloud messages.
`G90CloudStatusChangeReqMessageBase`(context)	Base class for status change request messages in the G90 cloud protocol.

Exceptions

`G90CloudError`	Base exception for G90 cloud protocol errors.
`G90CloudMessageInvalid`	Raised when a message is invalid or cannot be processed.
`G90CloudMessageNoMatch`	Raised when a message does not match the expected format or type.

exception pyg90alarm.cloud.protocol.G90CloudError

Bases: G90Error

Base exception for G90 cloud protocol errors.

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

exception pyg90alarm.cloud.protocol.G90CloudMessageNoMatch

Bases: G90CloudError

Raised when a message does not match the expected format or type.

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

exception pyg90alarm.cloud.protocol.G90CloudMessageInvalid

Bases: G90CloudError

Raised when a message is invalid or cannot be processed.

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

class pyg90alarm.cloud.protocol.G90CloudMessageContext(local_ip, local_port, remote_host, remote_port, cloud_ip, cloud_port, upstream_host, upstream_port, device_id)

Bases: object

Context for G90 cloud messages.

This class holds information about the local and remote hosts and ports, as well as the cloud server and upstream connection details.

class pyg90alarm.cloud.protocol.G90CloudBase(context)

Bases: object

Base class for G90 cloud protocol messages.

Provides methods for encoding and decoding messages to and from their wire representation.

classmethod from_wire(data, context)

Decode a message from its wire representation.

Parameters:

data (bytes) – The raw bytes of the message.
context (G90CloudMessageContext) – The message context.

Return type:

TypeVar(CloudBaseT, bound= G90CloudBase)

Returns:

An instance of the message class.

to_wire()

Encode the message to its wire representation.

Return type:: bytes
Returns:: The raw bytes of the message.

classmethod size()

Get the size of the message in bytes.

Return type:: int
Returns:: The size of the message.

classmethod format()

Get the format string for the message.

Return type:: str
Returns:: The format string.

class pyg90alarm.cloud.protocol.G90CloudHeader(context, command, _source, flag1, _destination, message_length)

Bases: G90CloudBase

Header for G90 cloud protocol messages.

Contains metadata about the message, such as its command, source, destination, and payload length.

property source: G90CloudDirection

Get the source direction of the message.

Returns:: The source direction.

property destination: G90CloudDirection

Get the destination direction of the message.

Returns:: The destination direction.

classmethod from_wire(data, context)

Decode a header from its wire representation.

Parameters:

data (bytes) – The raw bytes of the header.
context (G90CloudMessageContext) – The message context.

Return type:

TypeVar(CloudHeaderT, bound= G90CloudHeader)

Returns:

An instance of the header class.

property payload: bytes

Get the payload of the message.

Returns:: The raw bytes of the payload.

property payload_length: int

Get the length of the payload in bytes.

Returns:: The payload length.

matches(value)

Check if the header matches another header.

Parameters:: value (G90CloudHeader) – The header to compare against.
Return type:: bool
Returns:: True if the headers match, False otherwise.

classmethod format()

Get the format string for the message.

Return type:: str
Returns:: The format string.

classmethod size()

Get the size of the message in bytes.

Return type:: int
Returns:: The size of the message.

to_wire()

Encode the message to its wire representation.

Return type:: bytes
Returns:: The raw bytes of the message.

class pyg90alarm.cloud.protocol.G90CloudHeaderVersioned(context, command, _source, flag1, _destination, message_length, version=1, sequence=0)

Bases: G90CloudHeader

Versioned header for G90 cloud protocol messages.

Adds version and sequence information to the header.

property destination: G90CloudDirection

Get the destination direction of the message.

Returns:: The destination direction.

classmethod format()

Get the format string for the message.

Return type:: str
Returns:: The format string.

classmethod from_wire(data, context)

Decode a header from its wire representation.

Parameters:

data (bytes) – The raw bytes of the header.
context (G90CloudMessageContext) – The message context.

Return type:

TypeVar(CloudHeaderT, bound= G90CloudHeader)

Returns:

An instance of the header class.

matches(value)

Check if the header matches another header.

Parameters:: value (G90CloudHeader) – The header to compare against.
Return type:: bool
Returns:: True if the headers match, False otherwise.

property payload: bytes

Get the payload of the message.

Returns:: The raw bytes of the payload.

property payload_length: int

Get the length of the payload in bytes.

Returns:: The payload length.

classmethod size()

Get the size of the message in bytes.

Return type:: int
Returns:: The size of the message.

property source: G90CloudDirection

Get the source direction of the message.

Returns:: The source direction.

to_wire()

Encode the message to its wire representation.

Return type:: bytes
Returns:: The raw bytes of the message.

class pyg90alarm.cloud.protocol.G90CloudMessage(context)

Bases: G90CloudBase

Base class for G90 cloud protocol messages with headers.

Provides methods for encoding and decoding messages with headers, as well as handling responses.

classmethod from_wire(data, context)

Decode a message from its wire representation.

Parameters:

data (bytes) – The raw bytes of the message.
context (G90CloudMessageContext) – The message context.

Return type:

TypeVar(CloudMessageT, bound= G90CloudMessage)

Returns:

An instance of the message class.

to_wire()

Encode the message to its wire representation.

Return type:: bytes
Returns:: The raw bytes of the message.

wire_responses(context)

Get the wire representations of the responses to this message.

Parameters:: context (G90CloudMessageContext) – The message context.
Return type:: List[bytes]
Returns:: A list of raw bytes for the responses.

classmethod matches(value)

Check if the message matches another message.

Parameters:: value (G90CloudMessage) – The message to compare against.
Return type:: bool
Returns:: True if the messages match, False otherwise.

classmethod header_matches(value)

Check if the header matches the expected header for this message type.

Parameters:: value (G90CloudHeader) – The header to compare against.
Return type:: bool
Returns:: True if the headers match, False otherwise.

classmethod format()

Get the format string for the message.

Return type:: str
Returns:: The format string.

classmethod size()

Get the size of the message in bytes.

Return type:: int
Returns:: The size of the message.

class pyg90alarm.cloud.protocol.G90CloudStatusChangeReqMessageBase(context)

Bases: G90CloudMessage

Base class for status change request messages in the G90 cloud protocol.

Provides methods for handling status change requests and their timestamps.

property timestamp: datetime

Get the timestamp as a datetime object.

Returns:: The message timestamp converted to a datetime object with UTC timezone.

classmethod matches(value)

Check if the message matches the expected type and format.

Parameters:: value (G90CloudMessage) – The message to compare against.
Return type:: bool
Returns:: True if the messages match, False otherwise.

classmethod format()

Get the format string for the message.

Return type:: str
Returns:: The format string.

classmethod from_wire(data, context)

Decode a message from its wire representation.

Parameters:

data (bytes) – The raw bytes of the message.
context (G90CloudMessageContext) – The message context.

Return type:

TypeVar(CloudMessageT, bound= G90CloudMessage)

Returns:

An instance of the message class.

classmethod header_matches(value)

Check if the header matches the expected header for this message type.

Parameters:: value (G90CloudHeader) – The header to compare against.
Return type:: bool
Returns:: True if the headers match, False otherwise.

classmethod size()

Get the size of the message in bytes.

Return type:: int
Returns:: The size of the message.

to_wire()

Encode the message to its wire representation.

Return type:: bytes
Returns:: The raw bytes of the message.

wire_responses(context)

Get the wire representations of the responses to this message.

Parameters:: context (G90CloudMessageContext) – The message context.
Return type:: List[bytes]
Returns:: A list of raw bytes for the responses.