feat: additional commands and more robust ones

Author: sidey79

signalduino/commands.py

@@ -26,28 +26,34 @@ def __init__(self, send_command_func: Callable[[str, bool, float, Optional[Patte
 
     def get_version(self, timeout: float = 2.0) -> str:
         """Query firmware version (V)."""
-        pattern = re.compile(r"V\s.*SIGNAL(?:duino|ESP|STM).*", re.IGNORECASE)
+        pattern = re.compile(r"V\s.*SIGNAL(?:duino|ESP|STM).*(?:\s\d\d:\d\d:\d\d)", re.IGNORECASE)
         return self._send("V", expect_response=True, timeout=timeout, response_pattern=pattern)
 
     def get_help(self) -> str:
         """Show help (?)."""
+        # This is for internal use/legacy. The MQTT 'cmds' command uses a specific pattern.
         return self._send("?", expect_response=True, timeout=2.0, response_pattern=None)
 
+    def get_cmds(self) -> str:
+        """Show help/commands (?). Used for MQTT 'cmds' command."""
+        pattern = re.compile(r".*")
+        return self._send("?", expect_response=True, timeout=2.0, response_pattern=pattern)
+    
     def get_free_ram(self) -> str:
         """Query free RAM (R)."""
         # Response is typically a number (bytes)
-        pattern = re.compile(r"^\d+$")
+        pattern = re.compile(r"^[0-9]+")
         return self._send("R", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def get_uptime(self) -> str:
         """Query uptime in seconds (t)."""
         # Response is a number (seconds)
-        pattern = re.compile(r"^\d+$")
+        pattern = re.compile(r"^[0-9]+")
         return self._send("t", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def ping(self) -> str:
         """Ping device (P)."""
-        return self._send("P", expect_response=True, timeout=2.0, response_pattern=re.compile(r"OK"))
+        return self._send("P", expect_response=True, timeout=2.0, response_pattern=re.compile(r"^OK$"))
 
     def get_cc1101_status(self) -> str:
         """Query CC1101 status (s)."""
@@ -70,7 +76,7 @@ def factory_reset(self) -> str:
     def get_config(self) -> str:
         """Read configuration (CG)."""
         # Response format: MS=1;MU=1;...
-        pattern = re.compile(r"^MS=.*")
+        pattern = re.compile(r"^M[S|N]=.*")
         return self._send("CG", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def set_decoder_state(self, decoder: str, enabled: bool) -> None:
@@ -121,10 +127,24 @@ def set_message_type_enabled(self, message_type: str, enabled: bool) -> None:
         command = f"C{flag_char}{cmd_char}"
         self._send(command, expect_response=False, timeout=0, response_pattern=None)
 
+    def get_ccconf(self) -> str:
+        """Query CC1101 configuration (C0DnF)."""
+        # Response format: C0Dnn=[A-F0-9a-f]+ (e.g., C0D11=0F)
+        pattern = re.compile(r"C0Dn11=[A-F0-9a-f]+")
+        return self._send("C0DnF", expect_response=True, timeout=2.0, response_pattern=pattern)
+
+    def get_ccpatable(self) -> str:
+        """Query CC1101 PA Table (C3E)."""
+        # Response format: C3E = ...
+        pattern = re.compile(r"^C3E\s=\s.*")
+        return self._send("C3E", expect_response=True, timeout=2.0, response_pattern=pattern)
+
     def read_cc1101_register(self, register: int) -> str:
         """Read CC1101 register (C<reg>). Register is int, sent as 2-digit hex."""
         reg_hex = f"{register:02X}"
-        return self._send(f"C{reg_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response format: Cnn = vv or ccreg 00: ...
+        pattern = re.compile(r"^(?:C[A-Fa-f0-9]{2}\s=\s[0-9A-Fa-f]+$|ccreg 00:)")
+        return self._send(f"C{reg_hex}", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def write_register(self, register: int, value: int) -> str:
         """Write to EEPROM/CC1101 register (W<reg><val>)."""
@@ -189,6 +209,13 @@ def send_raw(self, params: str) -> None:
         """Send Raw (SR...). params should be the full string after SR."""
         self._send(f"SR{params}", expect_response=False, timeout=0, response_pattern=None)
 
+    def send_raw_message(self, message: str) -> str:
+        """Send the raw message/command directly as payload. Expects a response."""
+        # The 'rawmsg' MQTT command sends the content of the payload directly as a command.
+        # It is assumed that it will get a response which is why we expect one.
+        # No specific pattern can be given here, rely on the default response matchers.
+        return self._send(message, expect_response=True, timeout=2.0, response_pattern=None)
+
     def send_xfsk(self, params: str) -> None:
         """Send xFSK (SN...). params should be the full string after SN."""
         self._send(f"SN{params}", expect_response=False, timeout=0, response_pattern=None)

signalduino/controller.py

@@ -235,7 +235,11 @@ def _parser_loop(self) -> None:
 
                 line_data = raw_line.strip()
 
-                if self._handle_as_command_response(line_data):
+                # Messages starting with \x02 (STX) are sensor data and should never be treated as command responses.
+                # They are passed directly to the parser.
+                if line_data.startswith("\x02"):
+                    pass # Skip _handle_as_command_response and go to parsing
+                elif self._handle_as_command_response(line_data):
                     continue
 
                 if line_data.startswith("XQ") or line_data.startswith("XR"):
@@ -472,17 +476,41 @@ def _handle_mqtt_command(self, command: str, payload: str) -> None:
             self.logger.warning("Cannot handle MQTT command; publisher not connected.")
             return
 
+        # Mapping von MQTT-Befehl zu einer Methode (ohne Args) oder einer Lambda-Funktion (mit Args)
         command_mapping = {
             "version": self.commands.get_version,
-            "help": self.commands.get_help,
-            "free_ram": self.commands.get_free_ram,
+            "freeram": self.commands.get_free_ram,
             "uptime": self.commands.get_uptime,
+            # "help" wird durch "cmds" ersetzt, da der Serial Command "?" ignoriert werden sollte.
+            "cmds": self.commands.get_cmds, # Sendet Serial Command '?' mit Regex '.*'
+            "ping": self.commands.ping,
+            "config": self.commands.get_config,
+            "ccconf": self.commands.get_ccconf,
+            "ccpatable": self.commands.get_ccpatable,
+            "ccreg": lambda p: self.commands.read_cc1101_register(int(p, 16)),
+            "rawmsg": lambda p: self.commands.send_raw_message(p),
         }
 
+        # Der Befehl '?' soll ignoriert werden, aber 'cmds' wurde als Ersatz eingeführt.
+        if command == "help":
+            self.logger.warning("Ignoring deprecated 'help' MQTT command (use 'cmds').")
+            self.mqtt_publisher.publish_simple(f"error/{command}", "Deprecated command. Use 'cmds'.")
+            return
+        
         if command in command_mapping:
             try:
                 # Execute the corresponding command method
-                response = command_mapping[command]()
+                if command in ["ccreg", "rawmsg"]:
+                    # Befehle, die den Payload als Argument benötigen
+                    if not payload:
+                        self.logger.error("Command '%s' requires a payload argument.", command)
+                        self.mqtt_publisher.publish_simple(f"error/{command}", "Missing payload argument.")
+                        return
+                    
+                    response = command_mapping[command](payload)
+                else:
+                    # Befehle ohne Argumente
+                    response = command_mapping[command]()
 
                 self.logger.info("Got response for %s: %s", command, response)
 

tests/test_controller.py

@@ -80,7 +80,7 @@ def test_send_command_with_response(mock_transport, mock_parser):
     def write_line_side_effect(payload):
         # When the controller writes "V", simulate the device responding.
         if payload == "V":
-            response_q.put("V 3.5.0-dev SIGNALduino\n")
+            response_q.put("V 3.5.0-dev SIGNALduino - compiled at Mar 10 2017 22:54:50\n")
 
     def readline_side_effect():
         # Simulate blocking read that gets a value after write_line is called.
@@ -115,7 +115,7 @@ def test_send_command_with_interleaved_message(mock_transport, mock_parser):
     # The irrelevant message (e.g., an asynchronous received signal)
     interleaved_message = "MU;P0=353;P1=-184;D=0123456789;CP=1;SP=0;R=248;\n"
     # The expected command response
-    command_response = "V 3.5.0-dev SIGNALduino\n"
+    command_response = "V 3.5.0-dev SIGNALduino - compiled at Mar 10 2017 22:54:50\n"
 
     def write_line_side_effect(payload):
         # When the controller writes "V", simulate the device responding with
@@ -141,8 +141,9 @@ def readline_side_effect():
 
     controller = SignalduinoController(transport=mock_transport, parser=mock_parser)
     controller.connect()
+    time.sleep(0.2) # Give threads time to start
     try:
-        response = controller.commands.get_version(timeout=1)
+        response = controller.commands.get_version(timeout=2.0)
         mock_transport.write_line.assert_called_with("V")
 
         # 1. Verify that the correct command response was received by send_command
@@ -156,7 +157,7 @@ def readline_side_effect():
         mock_parser.parse_line.assert_called_with(interleaved_message.strip())
 
         # Give the parser thread a moment to process the message
-        time.sleep(0.1)
+        time.sleep(0.2)
 
     finally:
         controller.disconnect()
@@ -258,4 +259,58 @@ def side_effect(*args, **kwargs):
     finally:
         signalduino.controller.SDUINO_INIT_WAIT = original_wait
         signalduino.controller.SDUINO_INIT_WAIT_XQ = original_wait_xq
+        controller.disconnect()
+
+def test_stx_message_bypasses_command_response(mock_transport, mock_parser):
+    """
+    Test that messages starting with STX (\x02) are NOT treated as command responses,
+    even if the command's regex (like .* for cmds) would match them.
+    They should be passed directly to the parser.
+    """
+    # Queue for responses
+    response_q = queue.Queue()
+    
+    # STX message (Sensor data)
+    stx_message = "\x02SomeSensorData\x03\n"
+    # Expected response for 'cmds' (?)
+    cmd_response = "V X t R C S U P G r W x E Z\n"
+    
+    def write_line_side_effect(payload):
+        if payload == "?":
+            # Simulate STX message followed by real response
+            response_q.put(stx_message)
+            response_q.put(cmd_response)
+            
+    def readline_side_effect():
+        try:
+            return response_q.get(timeout=0.5)
+        except queue.Empty:
+            return None
+            
+    mock_transport.write_line.side_effect = write_line_side_effect
+    mock_transport.readline.side_effect = readline_side_effect
+    
+    # Mock parser to verify STX message is parsed
+    mock_parser.parse_line = Mock(wraps=mock_parser.parse_line)
+    
+    controller = SignalduinoController(transport=mock_transport, parser=mock_parser)
+    controller.connect()
+    time.sleep(0.2)
+    
+    try:
+        # get_cmds uses pattern r".*", which would normally match the STX message
+        # if we didn't have the special handling in the controller.
+        response = controller.commands.get_cmds()
+        
+        # Verify we got the correct response, not the STX message
+        assert response is not None
+        assert response.strip() == cmd_response.strip()
+        
+        # Verify STX message was sent to parser
+        mock_parser.parse_line.assert_any_call(stx_message.strip())
+        
+        # Give parser thread some time
+        time.sleep(0.2)
+        
+    finally:
         controller.disconnect()