This module imports the Kortex MQTT Transport, which permits high-level communication with Link arms.
- class MqttTransport(enable_logger=False)¶
Kortex Transport which communicates with Kortex devices over MQTT (high level API).
The normal usage is to create the transport, then pass it to the
RouterClientas it is created, then connect the transport to the Kortex device by calling the
connect()method. When the program reaches its end, you can call
disconnect()or simply let the destructor run, as it cleanly disconnects the transport itself.
transport = MqttTransport() router = RouterClient(transport) transport.connect("192.168.1.10", 1883) # Body of the code ... transport.disconnect() # optional
Constructor for an MqttTransport.
No connection information is passed to this method, as the connection only happens on a call to the “connect” method.
enable_logger (bool, optional) – If True, enables DEBUG logging level in the transport. Defaults to False.
- MQTT_MAX_PAYLOAD_SIZE = 268435455¶
- connect(host: str, port: int = 1883, timeoutSeconds: int = 10)¶
ITransportClientbase class for description.
Connects to a Kortex device through MQTT. An optional timeout can be specified to raise a ConnectionError if the connection is not established after a given period of time.
host (str) – Address of the Kortex device to connect to.
port (int, optional) – Port of the Kortex device to connect to. Defaults to DEFAULT_MQTT_BROKER_PORT.
timeoutSeconds (int, optional) – Timeout for the connection, in seconds. Defaults to DEFAULT_MQTT_TIMEOUT_SEC.
ConnectionError – The transport could not connect to the Kortex device within the ‘timeoutSeconds’ duration.