ddf.mqtt_communication package

Submodules

ddf.mqtt_communication.communicator module

Communicator using ASN1 as transmission file format.

class ddf.mqtt_communication.communicator.AntCommunicatorMqtt(client_id: UUID, base_topic: str, alias: str, mqtt_host: str, mqtt_port: int = 1883, autorun: bool = True)

Bases: Communicator

Implementation of Communicator for ASN1.

get_messages() → list[Message]

send_message(msg: Message) → None

ddf.mqtt_communication.communicator.translate_mode(mode: ControlCommand) → ControlCommand

Map a ControlCommand from ddf to asn1.

ddf.mqtt_communication.communicator.translate_mode_inv(command: ControlCommand) → ControlCommand

Map a ControlCommand from asn1 to ddf.

Inverse of translate_mode.

ddf.mqtt_communication.internal module

class ddf.mqtt_communication.internal.InternalCommunicator(client_id: uuid.UUID, base_topic: str, alias: str, mqtt_host: str, mqtt_port: int = 1883, autorun: bool = True, _lossnode_id: uuid.UUID | None = None, _queue: queue.Queue[ddf.mqtt_communication.message.Message] = <factory>)

Bases: object

alias: str

Human readable name for this node.

autorun: bool = True

base_topic: str

Base topic for all connected nodes.

client_id: UUID

Node Id.

mqtt_host: str

mqtt_port: int = 1883

send_message(msg: Message) → None

start() → None

Connect and start client loop.

stop() → None

test_send(msg_text: str, msg_topic: str) → None

ddf.mqtt_communication.message module

class ddf.mqtt_communication.message.AddNeighborMessage(sender: uuid.UUID, recipient: uuid.UUID, neighbor: uuid.UUID, operator: numpy.ndarray, req_id: uuid.UUID)

Bases: Message

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

neighbor: UUID

operator: ndarray

req_id: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

class ddf.mqtt_communication.message.ControlCommand(*values)

Bases: Enum

ACCEPT = 4

DECLINE = 5

FREEZE_ZERO_BACKPROP = 2

GET_NODE_CONTEXT = 1

GET_NODE_INFO = 0

UPDATE_UNFREEZE_INFORMATION = 3

class ddf.mqtt_communication.message.ControlMessage(sender: uuid.UUID, recipient: uuid.UUID, req_id: uuid.UUID, command: ddf.mqtt_communication.message.ControlCommand)

Bases: Message

command: ControlCommand

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

req_id: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

class ddf.mqtt_communication.message.GradientMessage(sender: uuid.UUID, recipient: uuid.UUID, reference: uuid.UUID, gradient: numpy.ndarray, loss_channel: ddf.ant_link.loss_channel.LossChannel, update_number: int)

Bases: Message

asdict() → dict[str, Any]

Convert the dataclass to a dictionary using dataclass’s asdict method.

This method will on top of dataclass.asdict convert data types for ASN.1 encoding.

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

gradient: ndarray

loss_channel: LossChannel

reference: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

update_number: int

class ddf.mqtt_communication.message.InfoMessage(sender: uuid.UUID, recipient: uuid.UUID, reference: uuid.UUID, mean: numpy.ndarray, covariance: numpy.ndarray, update_number: int)

Bases: Message

covariance: ndarray

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

mean: ndarray

reference: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

update_number: int

class ddf.mqtt_communication.message.Message(sender: uuid.UUID, recipient: uuid.UUID)

Bases: object

asdict() → dict[str, Any]

Convert the dataclass to a dictionary using dataclass’s asdict method.

This method will on top of dataclass.asdict convert data types for ASN.1 encoding.

classmethod fromdict(data: dict[str, Any]) → Message

Create Message dataclass instance from dictionary.

Reverses the process of asdict and converts the dictionary back to a dataclass, converting data types from ASN.1 encoding back to Python types including numpy arrays.

abstractmethod classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

abstractmethod get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

recipient: UUID

sender: UUID

abstractmethod classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

class ddf.mqtt_communication.message.NodeContextMessage(sender: uuid.UUID, recipient: uuid.UUID, fixed_context: str, user_context: str, req_id: uuid.UUID)

Bases: Message

fixed_context: str

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

req_id: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

user_context: str

class ddf.mqtt_communication.message.NodeInfoMessage(sender: uuid.UUID, recipient: uuid.UUID, available_list: list[str], exchange_list: list[str], alias: str, req_id: uuid.UUID)

Bases: Message

alias: str

available_list: list[str]

exchange_list: list[str]

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

req_id: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

class ddf.mqtt_communication.message.ProbeResponseMessage(sender: uuid.UUID, recipient: uuid.UUID, reference: uuid.UUID, gradient: numpy.ndarray, loss_channel: ddf.ant_link.loss_channel.LossChannel)

Bases: Message

classmethod get_asn1_content_choice() → str

Returns the name of the content choice in the ASN.1 specification.

Should be one of info, gradient, nodeinfo, proberesponse, nodecontext, addneighbor, control.

get_topic(sender_is_lossnode: bool, recipient_is_lossnode: bool) → str

Returns the topic to which the message should be sent.

gradient: ndarray

loss_channel: LossChannel

reference: UUID

classmethod translate() → list[tuple[str, str]]

Translation between naming conventions of python and ASN.1.

Returns a list of tuples, where each tuple contains the name of the field in the dataclass and the name of the field in the ASN.1 specification.

ddf.mqtt_communication.message.decode(encoded: bytes) → Message | None

ddf.mqtt_communication.message.encode(message: Message) → bytes

Module contents