Merge pull request #4 from fdlamotte/fix-error-handling

Handle error events properly in commands

examples/serial_repeater_status.py

@@ -1,32 +1,81 @@
 #!/usr/bin/python
 
 import asyncio
+import argparse
 
 from meshcore import MeshCore
 from meshcore.events import EventType
 
-PORT = "/dev/tty.usbserial-583A0069501"
-BAUDRATE = 115200
+async def main():
+    # Parse command line arguments
+    parser = argparse.ArgumentParser(description='Get status from a repeater via serial connection')
+    parser.add_argument('-p', '--port', required=True, help='Serial port')
+    parser.add_argument('-b', '--baudrate', type=int, default=115200, help='Baud rate')
+    parser.add_argument('-r', '--repeater', required=True, help='Repeater name')
+    parser.add_argument('-pw', '--password', required=True, help='Password for login')
+    parser.add_argument('-t', '--timeout', type=float, default=10.0, help='Timeout for responses in seconds')
+    args = parser.parse_args()
 
-REPEATER="Orion"
-PASSWORD="floopyboopy"
-
-async def main () :
-    mc = await MeshCore.create_serial(PORT, BAUDRATE)
-    await mc.commands.get_contacts()
-    repeater = mc.get_contact_by_name(REPEATER)
+    # Connect to the device
+    mc = await MeshCore.create_serial(args.port, args.baudrate, debug=True)
     
-    if repeater is None:
-        print(f"Repeater '{REPEATER}' not found in contacts.")
-        return
-    await mc.commands.send_login(repeater, PASSWORD)
-
-    print("Login sent ... awaiting")
-
-    if await mc.wait_for_event(EventType.LOGIN_SUCCESS):
-        print("Logged in success")
-        await mc.commands.send_statusreq(bytes.fromhex(repeater["public_key"]))
-        print("Status request sent ... awaiting")
-        print(await mc.wait_for_event(EventType.STATUS_RESPONSE))
+    try:
+        # Get contacts
+        await mc.ensure_contacts()
+        repeater = mc.get_contact_by_name(args.repeater)
+        
+        if repeater is None:
+            print(f"Repeater '{args.repeater}' not found in contacts.")
+            return
+        
+        # Send login request
+        print(f"Logging in to repeater '{args.repeater}'...")
+        login_event = await mc.commands.send_login(repeater, args.password)
+        
+        if login_event and login_event.get("success") != False:
+            print("Login successful")
+            
+            # Send status request
+            print("Sending status request...")
+            await mc.commands.send_statusreq(repeater)
+            
+            # Wait for status response
+            status_event = await mc.wait_for_event(EventType.STATUS_RESPONSE, timeout=args.timeout)
+            
+            if status_event:
+                # Format status information nicely
+                status = status_event.payload
+                print("\nRepeater Status:")
+                print(f"  Battery: {status.get('bat', 'N/A')}%")
+                print(f"  Uptime: {status.get('uptime', 'N/A')} seconds")
+                print(f"  Last RSSI: {status.get('last_rssi', 'N/A')}")
+                print(f"  Last SNR: {status.get('last_snr', 'N/A')} dB")
+                print(f"  Messages received: {status.get('nb_recv', 'N/A')}")
+                print(f"  Messages sent: {status.get('nb_sent', 'N/A')}")
+                print(f"  Direct messages sent: {status.get('sent_direct', 'N/A')}")
+                print(f"  Flood messages sent: {status.get('sent_flood', 'N/A')}")
+                print(f"  Direct messages received: {status.get('recv_direct', 'N/A')}")
+                print(f"  Flood messages received: {status.get('recv_flood', 'N/A')}")
+                print(f"  Direct duplicates: {status.get('direct_dups', 'N/A')}")
+                print(f"  Flood duplicates: {status.get('flood_dups', 'N/A')}")
+                print(f"  TX queue length: {status.get('tx_queue_len', 'N/A')}")
+                print(f"  Free queue length: {status.get('free_queue_len', 'N/A')}")
+                print(f"  Full events: {status.get('full_evts', 'N/A')}")
+                print(f"  Airtime: {status.get('airtime', 'N/A')}")
+            else:
+                print("Timed out waiting for status response")
+        else:
+            print("Login failed or timed out")
     
-asyncio.run(main())
+    finally:
+        # Always disconnect properly
+        await mc.disconnect()
+        print("Disconnected from device")
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("\nOperation cancelled by user")
+    except Exception as e:
+        print(f"Error: {e}")

src/meshcore/commands.py

@@ -95,11 +95,32 @@ class CommandHandler:
                     expected_events = [expected_events]
                     
                 logger.debug(f"Waiting for events {expected_events}, timeout={timeout}")
+                
+                # Create futures for all expected events
+                futures = []
                 for event_type in expected_events:
-                    # don't apply any filters for now, might change later
-                    event = await self.dispatcher.wait_for_event(event_type, {}, timeout)
+                    future = asyncio.create_task(
+                        self.dispatcher.wait_for_event(event_type, {}, timeout)
+                    )
+                    futures.append(future)
+                
+                # Wait for the first event to complete or all to timeout
+                done, pending = await asyncio.wait(
+                    futures, 
+                    timeout=timeout,
+                    return_when=asyncio.FIRST_COMPLETED
+                )
+                
+                # Cancel all pending futures
+                for future in pending:
+                    future.cancel()
+                
+                # Check if any future completed successfully
+                for future in done:
+                    event = await future
                     if event:
                         return event.payload
+                        
                 return {"success": False, "reason": "no_event_received"}
             except asyncio.TimeoutError:
                 logger.debug(f"Command timed out {data}")

src/meshcore/events.py

@@ -116,10 +116,11 @@ class EventDispatcher:
             event = await self.queue.get()
             logger.debug(f"Dispatching event: {event.type}, {event.payload}, {event.attributes}")
             for subscription in self.subscriptions.copy():
+                logger.debug(f"Checking subscription: {subscription.event_type}, {subscription.attribute_filters}")
                 # Check if event type matches
                 if subscription.event_type is None or subscription.event_type == event.type:
                     # Check if all attribute filters match
-                    if subscription.attribute_filters:
+                    if subscription.attribute_filters and subscription.attribute_filters != {}:
                         # Skip if any filter doesn't match the corresponding event attribute
                         if not all(event.attributes.get(key) == value 
                                 for key, value in subscription.attribute_filters.items()):

tests/unit/test_commands.py

@@ -231,4 +231,24 @@ async def test_send_trace(command_handler, mock_connection):
         path="01,23,45"
     )
     second_call = mock_connection.send.call_args[0][0]
-    assert second_call.startswith(b"\x24")
+    assert second_call.startswith(b"\x24")
+
+async def test_send_with_multiple_expected_events_returns_first_completed(command_handler, mock_connection, mock_dispatcher):
+    # Setup the dispatcher to return an ERROR event
+    error_payload = {"success": False, "reason": "command_failed"}
+    
+    async def simulate_error_event(*args, **kwargs):
+        # Simulate an ERROR event being returned
+        return Event(EventType.ERROR, error_payload)
+    
+    # Patch the wait_for_event method to return our simulated event
+    mock_dispatcher.wait_for_event.side_effect = simulate_error_event
+    
+    # Call send with both OK and ERROR in the expected_events list, with OK first
+    result = await command_handler.send(b"test_command", [EventType.OK, EventType.ERROR])
+    
+    # Verify the command was sent
+    mock_connection.send.assert_called_once_with(b"test_command")
+    
+    # Verify that even though OK was listed first, the ERROR event was returned
+    assert result == error_payload