Feature/mqtt connector

README.md

@@ -248,6 +248,79 @@ post-callbacks.
 These callbacks enable operations to be executed **before** or **after** the method invocation,
 allowing for user-defined interactions with the data model.
 
+## Connectors
+
+Connectors keep the public node API protocol-agnostic. A variable can still be
+read with `read()`, written with `write()`, and subscribed to regardless of the
+remote backend.
+
+### OPC UA
+
+OPC UA variables map to OPC UA nodes. Reads and writes are performed on the
+OPC UA nodes, and method calls are performed on the OPC UA method nodes.
+
+```yaml
+connectors:
+  - !!OpcUaConnector
+    name: "opc_ua_server"
+    ip: "127.0.0.1"
+    port: 4840
+
+root:
+  !!FolderNode
+  name: "Objects"
+  connector_name: "opc_ua_server"
+  children:
+    - !!NumericalVariableNode
+      name: "Temperature"
+      default_value: 20.0
+    - !!BooleanVariableNode
+      name: "StartCommand"
+      default_value: false
+      remote_resource_spec:
+        !!OpcUaRemoteResourceSpec
+        node_id: "ns=2;s=StartCommand"
+```
+
+### MQTT
+
+MQTT variables map to MQTT topics. Reads return the last received payload for
+the subscription topic, writes publish the value to the publish topic,
+and method calls are not supported by the MQTT connector yet.
+
+```yaml
+connectors:
+  - !!MqttConnector
+    name: "mqtt_broker"
+    ip: "127.0.0.1"
+    port: 1883
+    topic_prefix: "machines/boiler-1"
+    payload_codec: "string"
+
+root:
+  !!FolderNode
+  name: "Objects"
+  connector_name: "mqtt_broker"
+  children:
+    - !!NumericalVariableNode
+      name: "Temperature"
+      default_value: 20.0
+    - !!BooleanVariableNode
+      name: "StartCommand"
+      default_value: false
+      remote_resource_spec:
+        !!MqttRemoteResourceSpec
+        subscribe_topic: "plant/line-1/start/state"
+        publish_topic: "plant/line-1/start/set"
+```
+
+Supported MQTT payload codecs are `string`, `json`, and `msgpack`. The default
+`string` codec encodes strings, integers, floats, and booleans as UTF-8 scalar
+payloads and uses the data model node type when deserializing subscribed values.
+Python code can also pass `payload_serializer` and `payload_deserializer` to
+`MqttConnector` for custom payload formats. Topic wildcards are not supported
+in remote resource specs.
+
 ## Protocol Manager
 
 The protocol manager is a component that acts as an interface between the data model

examples/mqtt/basic_mqtt.yml

@@ -0,0 +1,27 @@
+name: "boiler"
+machine_category: "boiler_cat"
+machine_type: "boiler_type"
+machine_model: "boiler_model"
+description: "Boiler data model with MQTT-backed variables"
+connectors:
+  - !!MqttConnector
+    name: "mqtt_broker"
+    ip: "127.0.0.1"
+    port: 1883
+    topic_prefix: "machines/boiler-1"
+    payload_codec: "string"
+root:
+  !!FolderNode
+  name: "Objects"
+  connector_name: "mqtt_broker"
+  children:
+    - !!NumericalVariableNode
+      name: "Temperature"
+      default_value: 20.0
+    - !!BooleanVariableNode
+      name: "StartCommand"
+      default_value: false
+      remote_resource_spec:
+        !!MqttRemoteResourceSpec
+        subscribe_topic: "plant/line-1/start/state"
+        publish_topic: "plant/line-1/start/set"

examples/opcua/connect.py

@@ -18,6 +18,7 @@
     AbstractConnector,
     SubscriptionArguments,
 )
+from machine_data_model.nodes.connectors.remote_resource import RemoteResource
 from machine_data_model.nodes.method_node import MethodNode
 from machine_data_model.nodes.subscription.variable_subscription import (
     DataChangeSubscription,
@@ -70,12 +71,13 @@ def read_and_write_variable_node_using_connector(
     connector: AbstractConnector, node_path: str
 ):
     """Uses the connector to read and write the variable."""
+    resource = RemoteResource(node_path)
     print(f"Reading node '{node_path}'...")
-    current_value = connector.read_node_value(node_path)
+    current_value = connector.read_node_value(resource)
     print("current value:", current_value)
-    connector.write_node_value(node_path, current_value + 10)
+    connector.write_node_value(resource, current_value + 10)
     print("wrote the previous value + 10")
-    new_value = connector.read_node_value(node_path)
+    new_value = connector.read_node_value(resource)
     print(f"Read node '{node_path}' again, its current value is:", new_value)
 
 
@@ -147,7 +149,9 @@ def subscribe_to_variable_node_using_connector(
     connector: AbstractConnector, node_path: str
 ):
     """Subscribe to node changes using the connector."""
-    connector.subscribe_to_node_changes(node_path, my_remote_callback)
+    connector.subscribe_to_node_changes(
+        RemoteResource(node_path), my_remote_callback
+    )
 
 
 def main() -> None:

machine_data_model/builder/data_model_builder.py

@@ -29,6 +29,10 @@
 from machine_data_model.nodes.composite_method.composite_method_node import (
     CompositeMethodNode,
 )
+from machine_data_model.nodes.connectors.mqtt import (
+    MqttConnector,
+    MqttRemoteResourceSpec,
+)
 from machine_data_model.nodes.connectors.opcua.opcua_connector import (
     OpcuaConnector,
     OpcuaRemoteResourceSpec,
@@ -640,6 +644,50 @@ def _get_opcua_remote_resource_spec(
     return OpcuaRemoteResourceSpec(**kwargs)
 
 
+def _get_mqtt_connector_node(
+    loader: yaml.FullLoader, node: yaml.MappingNode
+) -> MqttConnector:
+    """Construct an MQTT Connector from a yaml node."""
+    data = loader.construct_mapping(node, deep=True)
+    default_kwargs = {
+        "name": None,
+        "ip": "127.0.0.1",
+        "ip_env_var": None,
+        "port": 1883,
+        "port_env_var": None,
+        "username": None,
+        "username_env_var": None,
+        "password": None,
+        "password_env_var": None,
+        "client_id": None,
+        "topic_prefix": None,
+        "keepalive": 60,
+        "qos": 0,
+        "retain": False,
+        "payload_codec": "string",
+    }
+    kwargs = _build_kwargs(data, default_kwargs)
+    return MqttConnector(**kwargs)
+
+
+def _get_mqtt_remote_resource_spec(
+    loader: yaml.FullLoader, node: yaml.MappingNode
+) -> MqttRemoteResourceSpec:
+    """Construct an MQTT remote resource spec from a yaml node."""
+    data = loader.construct_mapping(node, deep=True)
+    default_kwargs = {
+        "remote_path": None,
+        "topic": None,
+        "topic_prefix": None,
+        "publish_topic": None,
+        "subscribe_topic": None,
+        "qos": None,
+        "retain": None,
+    }
+    kwargs = _build_kwargs(data, default_kwargs)
+    return MqttRemoteResourceSpec(**kwargs)
+
+
 def _register_yaml_constructors() -> None:
     """Register all YAML constructors for data model building."""
     constructors = {
@@ -661,6 +709,8 @@ def _register_yaml_constructors() -> None:
         WaitRemoteEventNode: _get_wait_remote_event_node,
         OpcuaConnector: _get_opcua_connector_node,
         OpcuaRemoteResourceSpec: _get_opcua_remote_resource_spec,
+        MqttConnector: _get_mqtt_connector_node,
+        MqttRemoteResourceSpec: _get_mqtt_remote_resource_spec,
     }
 
     for node, constructor in constructors.items():

machine_data_model/builder/data_model_dumper.py

@@ -27,6 +27,10 @@
 from machine_data_model.nodes.composite_method.composite_method_node import (
     CompositeMethodNode,
 )
+from machine_data_model.nodes.connectors.mqtt import (
+    MqttConnector,
+    MqttRemoteResourceSpec,
+)
 from machine_data_model.nodes.connectors.opcua.opcua_connector import (
     OpcuaConnector,
     OpcuaRemoteResourceSpec,
@@ -89,14 +93,7 @@ def _opcua_connector_representer(
     """
     connector_dict: dict[str, Any] = {"name": connector.name}
 
-    if connector.ip_env_var:
-        connector_dict["ip_env_var"] = connector.ip_env_var
-    else:
-        connector_dict["ip"] = connector.ip
-    if connector.port_env_var:
-        connector_dict["port_env_var"] = connector.port_env_var
-    else:
-        connector_dict["port"] = connector.port
+    connector_dict.update(connector.address_to_dict())
     if connector.security_policy:
         connector_dict["security_policy"] = connector.security_policy
     if connector.client_app_uri != connector.get_default_client_app_uri():
@@ -147,6 +144,47 @@ def _get_opcua_remote_resource_spec_representer(
     )
 
 
+def _mqtt_connector_representer(
+    dumper: yaml.Dumper, connector: MqttConnector
+) -> yaml.nodes.MappingNode:
+    """Represent an MqttConnector as a YAML mapping node."""
+    connector_dict: dict[str, Any] = {"name": connector.name}
+
+    connector_dict.update(connector.address_to_dict())
+    connector_dict.update(connector.auth_to_dict())
+    if connector.client_id:
+        connector_dict["client_id"] = connector.client_id
+    if connector.topic_prefix:
+        connector_dict["topic_prefix"] = connector.topic_prefix
+    if connector.keepalive != 60:
+        connector_dict["keepalive"] = connector.keepalive
+    if connector.qos != 0:
+        connector_dict["qos"] = connector.qos
+    if connector.retain:
+        connector_dict["retain"] = connector.retain
+    if connector.payload_codec != "string":
+        connector_dict["payload_codec"] = connector.payload_codec
+
+    return dumper.represent_mapping(
+        "tag:yaml.org,2002:MqttConnector", connector_dict
+    )
+
+
+def _get_mqtt_remote_resource_spec_representer(
+    dumper: yaml.Dumper, spec: MqttRemoteResourceSpec
+) -> yaml.nodes.MappingNode:
+    """Represent an MqttRemoteResourceSpec as a YAML mapping node."""
+    remote_resource_spec: dict[str, Any] = {}
+    for key, value in spec.to_dict().items():
+        if value is not None:
+            remote_resource_spec[key] = value
+
+    return dumper.represent_mapping(
+        "tag:yaml.org,2002:MqttRemoteResourceSpec",
+        remote_resource_spec,
+    )
+
+
 def _get_data_model_node_common_fields(node: DataModelNode) -> dict[str, Any]:
     """Get common fields for variable nodes.
 
@@ -623,6 +661,8 @@ def _register_representers() -> None:
         DataModel: _data_model_representer,
         OpcuaConnector: _opcua_connector_representer,
         OpcuaRemoteResourceSpec: _get_opcua_remote_resource_spec_representer,
+        MqttConnector: _mqtt_connector_representer,
+        MqttRemoteResourceSpec: _get_mqtt_remote_resource_spec_representer,
         FolderNode: _folder_node_representer,
         NumericalVariableNode: _numerical_variable_node_representer,
         BooleanVariableNode: _boolean_variable_node_representer,