wait for responses after sending commands

gabrielburnworth · gabrielburnworth · commit c66a54c21aee · 2024-08-31T14:04:25.000-07:00
diff --git a/README.md b/README.md
@@ -159,7 +159,7 @@ sidecar-starter-pack/
 
 | class `BasicCommands()` | Description |
 | :--- | :--- |
-| `wait()` | Pauses FarmBot execution for a certain number of milliseconds. Note: You may need to combine this with a `sleep()` in your sidecar code. |
+| `wait()` | Pauses FarmBot execution for a certain number of milliseconds. |
 | `e_stop()` | Emergency locks (E-stops) the Farmduino microcontroller and resets peripheral pins to OFF. |
 | `unlock()` | Unlocks a locked (E-stopped) device. |
 | `reboot()` | Reboots FarmBot OS and reinitializes the device. |
@@ -172,8 +172,6 @@ sidecar-starter-pack/
 | `connect()` | Establish a persistent connection to send messages via the message broker. |
 | `disconnect()` | Disconnect from the message broker. |
 | `publish()` | Publish messages containing CeleryScript via the message broker. |
-| `on_connect()` | Callback function triggered when a connection to the message broker is successfully established. |
-| `on_message()` | Callback function triggered when a message is received from the message broker. |
 | `start_listen()` | Establish persistent subscription to message broker channels. |
 | `stop_listen()` | End subscription to all message broker channels. |
 | `listen()` | Listen to a message broker channel for the provided duration in seconds. |
diff --git a/farmbot_sidecar_starter_pack/functions/broker.py b/farmbot_sidecar_starter_pack/functions/broker.py
@@ -6,14 +6,13 @@
 #     ├── [BROKER] connect()
 #     ├── [BROKER] disconnect()
 #     ├── [BROKER] publish()
-#     ├── [BROKER] on_connect()
-#     ├── [BROKER] on_message
 #     ├── [BROKER] start_listen()
 #     ├── [BROKER] stop_listen()
 #     └── [BROKER] listen()
 
 import time
 import json
+import uuid
 from datetime import datetime
 import paho.mqtt.client as mqtt
 
@@ -30,12 +29,12 @@ def connect(self):
 
         self.client = mqtt.Client()
         self.client.username_pw_set(
-            username=self.state.token['token']['unencoded']['bot'],
-            password=self.state.token['token']['encoded']
+            username=self.state.token["token"]["unencoded"]["bot"],
+            password=self.state.token["token"]["encoded"]
         )
 
         self.client.connect(
-            self.state.token['token']['unencoded']['mqtt'],
+            self.state.token["token"]["unencoded"]["mqtt"],
             port=1883,
             keepalive=60
         )
@@ -63,7 +62,7 @@ def wrap_message(self, message, priority=None):
         }
 
         if priority is not None:
-            rpc['args']['priority'] = priority
+            rpc["args"]["priority"] = priority
 
         return rpc
 
@@ -73,80 +72,118 @@ def publish(self, message):
         if self.client is None:
             self.connect()
 
-        if message["kind"] != "rpc_request":
-            message = self.wrap_message(message)
+        rpc = message
+        if rpc["kind"] != "rpc_request":
+            rpc = self.wrap_message(rpc)
 
-        device_id_str = self.state.token["token"]["unencoded"]["bot"]
-        topic = f"bot/{device_id_str}/from_clients"
-        if not self.state.dry_run:
-            self.client.publish(topic, payload=json.dumps(message))
-        self.state.print_status(description=f"Publishing to {topic}:")
-        self.state.print_status(endpoint_json=message, update_only=True)
+        if rpc["args"]["label"] == "":
+            rpc["args"]["label"] = uuid.uuid4().hex if not self.state.test_env else "test"
+
+        self.state.print_status(description="Publishing to 'from_clients':")
+        self.state.print_status(endpoint_json=rpc, update_only=True)
         if self.state.dry_run:
             self.state.print_status(description="Sending disabled, message not sent.", update_only=True)
+        else:
+            self.listen("from_device", publish_payload=rpc)
 
-    def on_connect(self, _client, _userdata, _flags, _rc, channel):
-        """Callback function when connection to message broker is successful."""
-
-        self.client.subscribe(
-            f"bot/{self.state.token['token']['unencoded']['bot']}/{channel}")
-
-        self.state.print_status(description=f"Connected to message broker channel {channel}")
-
-    def on_message(self, _client, _userdata, msg, channel):
-        """Callback function when message received from message broker."""
-
-        self.state.last_messages[channel] = json.loads(msg.payload)
+        response = self.state.last_messages.get("from_device")
+        if response is not None:
+            if response["kind"] == "rpc_ok":
+                self.state.print_status(description="Success response received.", update_only=True)
+                self.state.error = None
+            else:
+                self.state.print_status(description="Error response received.", update_only=True)
+                self.state.error = "RPC error response received."
 
-        self.state.print_status(endpoint_json=json.loads(msg.payload), description=f"TOPIC: {msg.topic} ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})\n")
+        self.state.last_published = rpc
 
     def start_listen(self, channel="#"):
         """Establish persistent subscription to message broker channels."""
 
         if self.client is None:
             self.connect()
 
-        def on_connect(client, userdata, flags, rc):
-            """Wrap on_connect to pass channel argument."""
-            self.on_connect(client, userdata, flags, rc, channel)
+        # Set on_message callback
+        def on_message(_client, _userdata, msg):
+            """on_message callback"""
 
-        def on_message(client, userdata, msg):
-            """Wrap on_message to pass channel argument."""
-            self.on_message(client, userdata, msg, channel)
+            self.state.last_messages[channel] = json.loads(msg.payload)
+
+
+            self.state.print_status(description="", update_only=True)
+            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            self.state.print_status(
+                endpoint_json=json.loads(msg.payload),
+                description=f"TOPIC: {msg.topic} ({timestamp})\n")
 
-        self.client.on_connect = on_connect
         self.client.on_message = on_message
 
+        # Subscribe to channel
+        device_id_str = self.state.token["token"]["unencoded"]["bot"]
+        self.client.subscribe(f"bot/{device_id_str}/{channel}")
+        self.state.print_status(description=f"Connected to message broker channel '{channel}'")
+
+        # Start listening
         self.client.loop_start()
-        self.state.print_status(description=f"Now listening to message broker channel {channel}.")
+        self.state.print_status(description=f"Now listening to message broker channel '{channel}'.")
 
     def stop_listen(self):
         """End subscription to all message broker channels."""
 
         self.client.loop_stop()
-        self.client.disconnect()
 
         self.state.print_status(description="Stopped listening to all message broker channels.")
 
-    def listen(self, duration, channel):
+    def listen(self, channel, duration=None, publish_payload=None):
         """Listen to a message broker channel for the provided duration in seconds."""
-        self.state.print_status(description=f"Listening to message broker for {duration} seconds...")
+        # Prepare parameters
+        duration_seconds = duration or self.state.broker_listen_duration
+        message = (publish_payload or {}).get("body", [{}])[0]
+        if message.get("kind") == "wait":
+            duration_seconds += message["args"]["milliseconds"] / 1000
+        publish = publish_payload is not None
+        label = None
+        if publish and publish_payload["args"]["label"] != "":
+            label = publish_payload["args"]["label"]
+
+        # Print status message
+        channel_str = f" channel '{channel}'" if channel != "#" else ""
+        duration_str = f" for {duration_seconds} seconds"
+        label_str = f" for label '{label}'" if label is not None else ""
+        description = f"Listening to message broker{channel_str}{duration_str}{label_str}..."
+        self.state.print_status(description=description)
+
+        # Start listening
         start_time = datetime.now()
         self.start_listen(channel)
         if not self.state.test_env:
             self.state.last_messages[channel] = None
-        while (datetime.now() - start_time).seconds < duration:
+        if publish:
+            time.sleep(0.1) # wait for start_listen to be ready
+            device_id_str = self.state.token["token"]["unencoded"]["bot"]
+            publish_topic =  f"bot/{device_id_str}/from_clients"
+            self.client.publish(publish_topic, payload=json.dumps(publish_payload))
+        self.state.print_status(update_only=True, description="", end="")
+        while (datetime.now() - start_time).seconds < duration_seconds:
             self.state.print_status(update_only=True, description=".", end="")
             time.sleep(0.25)
-            if self.state.last_messages.get(channel) is not None:
+            last_message = self.state.last_messages.get(channel)
+            if last_message is not None:
+                # If a label is provided, verify the label matches
+                if label is not None and last_message["args"]["label"] != label:
+                    self.state.last_messages[channel] = None
+                    continue
                 seconds = (datetime.now() - start_time).seconds
+                self.state.print_status(description="", update_only=True)
                 self.state.print_status(
                     description=f"Message received after {seconds} seconds",
                     update_only=True)
                 break
         if self.state.last_messages.get(channel) is None:
+            self.state.print_status(description="", update_only=True)
             self.state.print_status(
-                description=f"Did not receive message after {duration} seconds",
+                description=f"Did not receive message after {duration_seconds} seconds",
                 update_only=True)
+            self.state.error = "Timed out waiting for RPC response."
 
         self.stop_listen()
diff --git a/farmbot_sidecar_starter_pack/functions/information.py b/farmbot_sidecar_starter_pack/functions/information.py
@@ -148,7 +148,7 @@ def read_status(self):
         }
         self.broker.publish(status_message)
 
-        self.broker.listen(self.state.broker_listen_duration, "status")
+        self.broker.listen("status")
 
         status_tree = self.state.last_messages.get("status")
 
diff --git a/farmbot_sidecar_starter_pack/main.py b/farmbot_sidecar_starter_pack/main.py
@@ -15,7 +15,7 @@
 from .functions.resources import Resources
 from .functions.tools import ToolControls
 
-VERSION = "1.1.1"
+VERSION = "1.2.0"
 
 class Farmbot():
     """Farmbot class."""
@@ -83,9 +83,9 @@ def disconnect_broker(self):
         """Disconnect from the message broker."""
         return self.broker.disconnect()
 
-    def listen(self, duration, channel="#"):
+    def listen(self, channel="#", duration=None):
         """Listen to a message broker channel for the provided duration in seconds."""
-        return self.broker.listen(duration, channel)
+        return self.broker.listen(channel, duration)
 
     # camera.py
 
diff --git a/farmbot_sidecar_starter_pack/state.py b/farmbot_sidecar_starter_pack/state.py
@@ -33,6 +33,7 @@ def __init__(self):
         self.token = None
         self.error = None
         self.last_messages = {}
+        self.last_published = {}
         self.verbosity = 2
         self.broker_listen_duration = 15
         self.test_env = False
@@ -46,7 +47,8 @@ def print_status(self, endpoint_json=None, description=None, update_only=False,
         if depth < self.min_call_stack_depth:
             self.min_call_stack_depth = depth
         top = depth == self.min_call_stack_depth
-        indent = "" if top else " " * 4
+        no_end = end == "" and description != ""
+        indent = "" if (top or no_end) else " " * 4
 
         if self.verbosity >= 2 and not update_only:
             if top:
@@ -56,7 +58,7 @@ def print_status(self, endpoint_json=None, description=None, update_only=False,
         if self.verbosity >= 1:
             if self.verbosity == 1 and not update_only and top:
                 print()
-            if description:
+            if description is not None:
                 print(indent + description, end=end, flush=(end == ""))
             if endpoint_json:
                 json_str = json.dumps(endpoint_json, indent=4)
diff --git a/tests/tests_main.py b/tests/tests_main.py
@@ -30,7 +30,7 @@ def setUp(self):
         self.fb.set_token(MOCK_TOKEN)
         self.fb.set_verbosity(0)
         self.fb.state.test_env = True
-        self.fb.state.broker_listen_duration = 0.3
+        self.fb.state.broker_listen_duration = 0
 
     @patch('requests.post')
     def test_get_token_default_server(self, mock_post):
@@ -577,8 +577,7 @@ def test_listen(self, mock_mqtt):
         '''Test listen command'''
         mock_client = Mock()
         mock_mqtt.return_value = mock_client
-        self.fb.listen(1)
-        mock_client.on_connect('', '', '', '')
+        self.fb.listen()
 
         class MockMessage:
             '''Mock message class'''
@@ -595,7 +594,6 @@ class MockMessage:
         mock_client.subscribe.assert_called_once_with('bot/device_0/#')
         mock_client.loop_start.assert_called()
         mock_client.loop_stop.assert_called()
-        mock_client.disconnect.assert_called()
 
     @patch('paho.mqtt.client.Client')
     def test_listen_clear_last(self, mock_mqtt):
@@ -604,9 +602,18 @@ def test_listen_clear_last(self, mock_mqtt):
         mock_mqtt.return_value = mock_client
         self.fb.state.last_messages = {'#': "message"}
         self.fb.state.test_env = False
-        self.fb.listen(1)
+        self.fb.listen()
         self.assertIsNone(self.fb.state.last_messages['#'])
 
+    @patch('paho.mqtt.client.Client')
+    def test_publish_apply_label(self, mock_mqtt):
+        '''Test publish command: set uuid'''
+        mock_client = Mock()
+        mock_mqtt.return_value = mock_client
+        self.fb.state.test_env = False
+        self.fb.broker.publish({'kind': 'sync', 'args': {}})
+        self.assertNotIn(self.fb.state.last_published.get('args', {}).get('label'), ['test', '', None])
+
     @patch('requests.request')
     @patch('paho.mqtt.client.Client')
     def send_command_test_helper(self, *args, **kwargs):
@@ -617,20 +624,25 @@ def send_command_test_helper(self, *args, **kwargs):
         expected_command = kwargs.get('expected_command')
         extra_rpc_args = kwargs.get('extra_rpc_args')
         mock_api_response = kwargs.get('mock_api_response')
+        error = kwargs.get('error')
         mock_client = Mock()
         mock_mqtt.return_value = mock_client
         mock_response = Mock()
         mock_response.json.return_value = mock_api_response
         mock_response.status_code = 200
         mock_response.text = 'text'
         mock_request.return_value = mock_response
+        self.fb.state.last_messages['from_device'] = {
+            'kind': 'rpc_error' if error else 'rpc_ok',
+            'args': {'label': 'test'},
+        }
         execute_command()
         if expected_command is None:
             mock_client.publish.assert_not_called()
             return
         expected_payload = {
             'kind': 'rpc_request',
-            'args': {'label': '', **extra_rpc_args},
+            'args': {'label': 'test', **extra_rpc_args},
             'body': [expected_command],
         }
         mock_client.username_pw_set.assert_called_once_with(
@@ -644,6 +656,8 @@ def send_command_test_helper(self, *args, **kwargs):
         mock_client.publish.assert_called_once_with(
             'bot/device_0/from_clients',
             payload=json.dumps(expected_payload))
+        if not error:
+            self.assertNotEqual(self.fb.state.error, 'RPC error response received.')
 
     def test_message(self):
         '''Test message command'''
@@ -1752,6 +1766,34 @@ def exec_command():
             extra_rpc_args={},
             mock_api_response=[])
 
+    def test_rpc_error(self):
+        '''Test rpc error handling'''
+        def exec_command():
+            self.fb.wait(100)
+            self.assertEqual(self.fb.state.error, 'RPC error response received.')
+        self.send_command_test_helper(
+            exec_command,
+            error=True,
+            expected_command={
+                'kind': 'wait',
+                'args': {'milliseconds': 100}},
+            extra_rpc_args={},
+            mock_api_response=[])
+
+    def test_rpc_response_timeout(self):
+        '''Test rpc response timeout handling'''
+        def exec_command():
+            self.fb.state.last_messages['from_device'] = {'kind': 'rpc_ok', 'args': {'label': 'wrong label'}}
+            self.fb.wait(100)
+            self.assertEqual(self.fb.state.error, 'Timed out waiting for RPC response.')
+        self.send_command_test_helper(
+            exec_command,
+            expected_command={
+                'kind': 'wait',
+                'args': {'milliseconds': 100}},
+            extra_rpc_args={},
+            mock_api_response=[])
+
     @staticmethod
     def helper_get_print_strings(mock_print):
         '''Test helper to get print call strings.'''

Original file line number	Diff line number	Diff line change
`@@ -148,7 +148,7 @@ def read_status(self):`
`148`	`148`	`}`
`149`	`149`	`self.broker.publish(status_message)`
`150`	`150`
`151`		`- self.broker.listen(self.state.broker_listen_duration, "status")`
	`151`	`+ self.broker.listen("status")`
`152`	`152`
`153`	`153`	`status_tree = self.state.last_messages.get("status")`
`154`	`154`