Updated gateway operational example and attribute response processing

Author: imbeacon

examples/gateway/operational_example.py

@@ -15,148 +15,162 @@
 import asyncio
 import logging
 import signal
-from datetime import datetime, UTC
-from random import randint, uniform
+from random import uniform, randint
+from time import time
 
 from tb_mqtt_client.common.config_loader import GatewayConfig
+from tb_mqtt_client.common.logging_utils import configure_logging, get_logger
 from tb_mqtt_client.entities.data.attribute_entry import AttributeEntry
-from tb_mqtt_client.entities.data.timeseries_entry import TimeseriesEntry
 from tb_mqtt_client.entities.data.attribute_update import AttributeUpdate
+from tb_mqtt_client.entities.data.requested_attribute_response import RequestedAttributeResponse
+from tb_mqtt_client.entities.data.timeseries_entry import TimeseriesEntry
+from tb_mqtt_client.entities.gateway.gateway_attribute_request import GatewayAttributeRequest
+from tb_mqtt_client.entities.gateway.gateway_rpc_request import GatewayRPCRequest
+from tb_mqtt_client.entities.gateway.gateway_rpc_response import GatewayRPCResponse
 from tb_mqtt_client.service.gateway.client import GatewayClient
-from tb_mqtt_client.common.logging_utils import configure_logging, get_logger
+from tb_mqtt_client.service.gateway.device_session import DeviceSession
 
+# ---- Logging Setup ----
 configure_logging()
 logger = get_logger(__name__)
 logger.setLevel(logging.DEBUG)
 logging.getLogger("tb_mqtt_client").setLevel(logging.DEBUG)
 
+# ---- Constants ----
+GATEWAY_HOST = "localhost"
+GATEWAY_PORT = 1883  # Default MQTT port, change if needed
+GATEWAY_ACCESS_TOKEN = "YOUR_ACCESS_TOKEN"  # Replace with your actual access token
+
+DELAY_BETWEEN_DATA_PUBLISH = 1  # seconds
+
+
+# ---- Callbacks ----
+async def attribute_update_handler(device_session: DeviceSession, update: AttributeUpdate):
+    logger.info("Received attribute update for %s: %s",
+                device_session.device_info.device_name, update.entries)
+
+
+async def requested_attributes_handler(device_session: DeviceSession, response: RequestedAttributeResponse):
+    logger.info("Requested attributes for %s -> client: %r, shared: %r",
+                device_session.device_info.device_name,
+                response.client,
+                response.shared)
+
 
-async def device_attribute_update_callback(update: AttributeUpdate):
-    """
-    Callback function to handle device attribute updates.
-    :param update: The attribute update object.
-    """
-    logger.info("Received attribute update for device %s: %s", update.device, update.attributes)
-
-
-async def device_rpc_request_callback(device_name: str, method: str, params: dict):
-    """
-    Callback function to handle device RPC requests.
-    :param device_name: Name of the device
-    :param method: RPC method
-    :param params: RPC parameters
-    :return: RPC response
-    """
-    logger.info("Received RPC request for device %s: method=%s, params=%s", device_name, method, params)
-    
-    # Example response based on method
-    if method == "getTemperature":
-        return {"temperature": round(uniform(20.0, 30.0), 2)}
-    elif method == "setLedState":
-        state = params.get("state", False)
-        return {"success": True, "state": state}
-    else:
-        return {"error": f"Unsupported method: {method}"}
-
-
-async def device_disconnect_callback(device_name: str):
-    """
-    Callback function to handle device disconnections.
-    :param device_name: Name of the disconnected device
-    """
-    logger.info("Device %s disconnected", device_name)
+async def device_rpc_request_handler(device_session: DeviceSession, rpc_request: GatewayRPCRequest) -> GatewayRPCResponse:
+    logger.info("Received RPC request for %s: %r", device_session.device_info.device_name, rpc_request)
+    response_data = {
+        "status": "success",
+        "echo_method": rpc_request.method,
+        "params": rpc_request.params
+    }
+    return GatewayRPCResponse.build(device_session.device_info.device_name, rpc_request.request_id, response_data)
 
 
+# ---- Main ----
 async def main():
     stop_event = asyncio.Event()
 
     def _shutdown_handler():
         stop_event.set()
+        asyncio.create_task(client.stop())
 
     loop = asyncio.get_running_loop()
     for sig in (signal.SIGINT, signal.SIGTERM):
         try:
             loop.add_signal_handler(sig, _shutdown_handler)
         except NotImplementedError:
-            # Windows compatibility fallback
             signal.signal(sig, lambda *_: _shutdown_handler())
 
+    # ---- Gateway Config ----
     config = GatewayConfig()
-    config.host = "localhost"
-    config.access_token = "YOUR_GATEWAY_ACCESS_TOKEN"
+    config.host = GATEWAY_HOST
+    config.port = GATEWAY_PORT
+    config.access_token = GATEWAY_ACCESS_TOKEN
 
+    global client
     client = GatewayClient(config)
-    client.set_device_attribute_update_callback(device_attribute_update_callback)
-    client.set_device_rpc_request_callback(device_rpc_request_callback)
-    client.set_device_disconnect_callback(device_disconnect_callback)
+
     await client.connect()
+    logger.info("Gateway connected to ThingsBoard.")
+
+    # ---- Register Devices ----
+    devices = [
+        ("Test Device A1", "Test devices"),
+        ("Test Device B1", "Test devices")
+    ]
+    sessions = {}
+
+    for name, profile in devices:
+        session, _ = await client.connect_device(name, profile, wait_for_publish=True)
+        if not session:
+            logger.error("Failed to connect %s", name)
+            continue
+        sessions[name] = session
+        logger.info("Device connected: %s", name)
+
+        # Register callbacks for each device
+        client.device_manager.set_attribute_update_callback(session.device_info.device_id, attribute_update_handler)
+        client.device_manager.set_attribute_response_callback(session.device_info.device_id, requested_attributes_handler)
+        client.device_manager.set_rpc_request_callback(session.device_info.device_id, device_rpc_request_handler)
+
+    # ---- Main loop ----
+    while not stop_event.is_set():
+        iteration_start = time()
 
-    logger.info("Connected to ThingsBoard as gateway.")
+        for device_name, session in sessions.items():
+            logger.info("Publishing data for %s", device_name)
 
-    # Connect devices to the gateway
-    device_names = ["sensor-1", "sensor-2", "actuator-1"]
-    for device_name in device_names:
-        await client.gw_connect_device(device_name)
-        logger.info("Connected device: %s", device_name)
+            # --- Attributes ---
+            raw_attrs = {"firmwareVersion": "2.0.0", "location": "office"}
+            await client.send_device_attributes(session, raw_attrs, wait_for_publish=True)
 
-    while not stop_event.is_set():
-        # Send device attributes
-        for device_name in device_names:
-            # Send device attributes
-            attributes = {
-                "firmwareVersion": "1.0.4",
-                "serialNumber": f"SN-{randint(1000, 9999)}",
-                "deviceType": "sensor" if "sensor" in device_name else "actuator"
-            }
-            await client.gw_send_attributes(device_name, attributes)
-            logger.info("Sent attributes for device %s: %s", device_name, attributes)
-
-            # Send single attribute
-            single_attr = AttributeEntry("lastUpdateTime", datetime.now(UTC).isoformat())
-            await client.gw_send_attributes(device_name, single_attr)
-            logger.info("Sent single attribute for device %s: %s", device_name, single_attr)
-
-            # Send device telemetry
-            if "sensor" in device_name:
-                # For sensor devices
-                telemetry = {
-                    "temperature": round(uniform(20.0, 30.0), 2),
-                    "humidity": round(uniform(40.0, 80.0), 2),
-                    "batteryLevel": randint(1, 100)
-                }
-                await client.gw_send_telemetry(device_name, telemetry)
-                logger.info("Sent telemetry for device %s: %s", device_name, telemetry)
-
-                # Send single telemetry entry
-                single_entry = TimeseriesEntry("signalStrength", randint(-90, -30))
-                await client.gw_send_telemetry(device_name, single_entry)
-                logger.info("Sent single telemetry entry for device %s: %s", device_name, single_entry)
-            else:
-                # For actuator devices
-                telemetry = {
-                    "state": "ON" if randint(0, 1) == 1 else "OFF",
-                    "powerConsumption": round(uniform(0.1, 5.0), 2),
-                    "uptime": randint(1, 1000)
-                }
-                await client.gw_send_telemetry(device_name, telemetry)
-                logger.info("Sent telemetry for device %s: %s", device_name, telemetry)
+            single_attr = AttributeEntry("mode", "auto")
+            await client.send_device_attributes(session, single_attr, wait_for_publish=True)
+
+            multi_attrs = [
+                AttributeEntry("maxTemp", randint(60, 90)),
+                AttributeEntry("calibrated", True)
+            ]
+            await client.send_device_attributes(session, multi_attrs, wait_for_publish=True)
+
+            # --- Telemetry ---
+            raw_ts = {"temperature": round(uniform(20, 30), 2), "humidity": randint(40, 80)}
+            await client.send_device_timeseries(session, raw_ts, wait_for_publish=True)
 
+            list_ts = [
+                {"ts": int(time() * 1000), "values": {"temperature": 25.5}},
+                {"ts": int(time() * 1000) - 1000, "values": {"humidity": 65}}
+            ]
+            await client.send_device_timeseries(session, list_ts, wait_for_publish=True)
+
+            ts_entries = [TimeseriesEntry(f"temp_{i}", randint(0, 50)) for i in range(5)]
+            await client.send_device_timeseries(session, ts_entries, wait_for_publish=True)
+
+            # --- Attribute Request ---
+            attr_req = await GatewayAttributeRequest.build(
+                device_session=session,
+                client_keys=["firmwareVersion", "mode"]
+            )
+            await client.send_device_attributes_request(session, attr_req, wait_for_publish=True)
+
+        # ---- Delay handling ----
         try:
-            await asyncio.wait_for(stop_event.wait(), timeout=5)
+            timeout = DELAY_BETWEEN_DATA_PUBLISH - (time() - iteration_start)
+            if timeout > 0:
+                await asyncio.wait_for(stop_event.wait(), timeout=timeout)
         except asyncio.TimeoutError:
             pass
 
-    # Disconnect devices before shutting down
-    for device_name in device_names:
-        await client.gw_disconnect_device(device_name)
-        logger.info("Disconnected device: %s", device_name)
-
+    # ---- Disconnect devices ----
+    for session in sessions.values():
+        await client.disconnect_device(session, wait_for_publish=True)
     await client.disconnect()
-    logger.info("Disconnected from ThingsBoard.")
+    logger.info("Gateway disconnected cleanly.")
 
 
 if __name__ == "__main__":
     try:
         asyncio.run(main())
     except KeyboardInterrupt:
-        print("Interrupted by user.")
+        logger.info("Interrupted by user.")

tb_mqtt_client/service/gateway/message_adapter.py

@@ -321,10 +321,11 @@ def parse_gateway_requested_attribute_response(self,
                                  and not gateway_attribute_request.client_keys))):
                 # TODO: Skipping case when requested several attributes, but only one is returned, issue on the platform
                 logger.warning("Received gateway attribute response with single key, but multiply keys expected. "
-                               "Request keys: %s, Response keys: %s",
-                               list(*gateway_attribute_request.client_keys, *gateway_attribute_request.shared_keys),
+                               "Request keys: %s, Response value: %r",
+                               gateway_attribute_request.client_keys + gateway_attribute_request.shared_keys,
                                data['value'])
-                return None
+                client = []
+                shared = []
             elif 'value' in data:
                 if not client_keys_empty and len(gateway_attribute_request.client_keys) == 1:
                     client = [AttributeEntry(gateway_attribute_request.client_keys[0], data['value'])]