feat: commands implemented with tests

Author: sidey79

signalduino/commands.py

@@ -97,6 +97,24 @@ def set_manchester_min_bit_length(self, length: int) -> str:
         """Set MC Min Bit Length (CSmcmbl=<val>)."""
         return self._send(f"CSmcmbl={length}", expect_response=True, timeout=2.0, response_pattern=None)
 
+    def set_message_type_enabled(self, message_type: str, enabled: bool) -> None:
+        """
+        Enable/disable reception for message types (C<FLAG><TYPE>).
+
+        Args:
+            message_type: One of 'MS', 'MU', 'MC' (or other 2-letter codes, e.g. 'MN').
+                          The second character is used as the type char in the command.
+            enabled: True to enable (E), False to disable (D).
+        """
+        if not message_type or len(message_type) != 2:
+             raise ValueError(f"Invalid message_type: {message_type}. Must be a 2-character string (e.g., 'MS').")
+
+        # The command structure seems to be C<E/D><S/U/C/N>, where <S/U/C/N> is the second char of message_type
+        cmd_char = message_type # 'S', 'U', 'C', 'N', etc.
+        flag_char = "E" if enabled else "D"
+        command = f"C{flag_char}{cmd_char}"
+        self._send(command, expect_response=False, timeout=0, response_pattern=None)
+
     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}"
@@ -122,11 +140,30 @@ def read_eeprom_block(self, address: int) -> str:
         addr_hex = f"{address:02X}"
         return self._send(f"r{addr_hex}n", expect_response=True, timeout=2.0, response_pattern=None)
 
-    def set_patable(self, value: int) -> str:
+    def set_patable(self, value: str | int) -> str:
         """Write PA Table (x<val>)."""
-        val_hex = f"{value:02X}"
+        if isinstance(value, int):
+            val_hex = f"{value:02X}"
+        else:
+            # Assume it's an already formatted hex string (e.g. 'C0')
+            val_hex = value
         return self._send(f"x{val_hex}", expect_response=True, timeout=2.0, response_pattern=None)
 
+    def set_bwidth(self, value: int) -> str:
+        """Set CC1101 Bandwidth (C10<val>)."""
+        val_str = str(value)
+        return self._send(f"C10{val_str}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def set_rampl(self, value: int) -> str:
+        """Set CC1101 PA_TABLE/ramp length (W1D<val>)."""
+        val_str = str(value)
+        return self._send(f"W1D{val_str}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def set_sens(self, value: int) -> str:
+        """Set CC1101 sensitivity/MCSM0 (W1F<val>)."""
+        val_str = str(value)
+        return self._send(f"W1F{val_str}", expect_response=True, timeout=2.0, response_pattern=None)
+
     # --- Send Commands ---
     # These typically don't expect a response, or the response is just an echo/OK which might be hard to sync with async rx
 
@@ -145,3 +182,10 @@ def send_raw(self, params: str) -> 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)
+
+    def send_message(self, message: str) -> None:
+        """
+        Sends a pre-encoded message (P..., S..., e.g. from an FHEM set command).
+        This command is sent without any additional prefix.
+        """
+        self._send(message, expect_response=False, timeout=0, response_pattern=None)

tests/test_set_commands.py

@@ -17,8 +17,18 @@ def mock_transport():
 def controller(mock_transport):
     """Fixture for a SignalduinoController with a mocked transport."""
     ctrl = SignalduinoController(transport=mock_transport)
+
+    def mock_put(queued_command):
+        # Simulate an immediate response for commands that expect one.
+        # This is necessary because we mock the internal thread queue.
+        if queued_command.expect_response and queued_command.on_response:
+            # For Set-Commands, the response is often an echo of the command itself or 'OK'.
+            # We use the command payload as the response.
+            queued_command.on_response(queued_command.payload)
+
     # We don't want to test the full threading model here, so we mock the queue
     ctrl._write_queue = MagicMock()
+    ctrl._write_queue.put.side_effect = mock_put
     return ctrl
 
 
@@ -38,17 +48,17 @@ def test_send_raw_command(controller):
 @pytest.mark.parametrize(
     "message_type, enabled, expected_command",
     [
-        ("MS", True, "CES"),
-        ("MS", False, "CDS"),
-        ("MU", True, "CEU"),
-        ("MU", False, "CDU"),
-        ("MC", True, "CEC"),
-        ("MC", False, "CDC"),
+        ("MS", True, "CEMS"),
+        ("MS", False, "CDMS"),
+        ("MU", True, "CEMU"),
+        ("MU", False, "CDMU"),
+        ("MC", True, "CEMC"),
+        ("MC", False, "CDMC"),
     ],
 )
 def test_set_message_type_enabled(controller, message_type, enabled, expected_command):
     """Test enabling and disabling message types."""
-    controller.set_message_type_enabled(message_type, enabled)
+    controller.commands.set_message_type_enabled(message_type, enabled)
 
     controller._write_queue.put.assert_called_once()
     queued_command = controller._write_queue.put.call_args[0][0]
@@ -66,7 +76,7 @@ def test_set_message_type_enabled(controller, message_type, enabled, expected_co
 )
 def test_cc1101_commands(controller, method_name, value, expected_command_prefix):
     """Test various CC1101 set commands."""
-    method = getattr(controller, method_name)
+    method = getattr(controller.commands, method_name)
     method(value)
 
     controller._write_queue.put.assert_called_once()
@@ -77,7 +87,7 @@ def test_cc1101_commands(controller, method_name, value, expected_command_prefix
 def test_send_message(controller):
     """Test sending a pre-encoded message."""
     message = "P3#is11111000000F#R6"
-    controller.send_message(message)
+    controller.commands.send_message(message)
 
     controller._write_queue.put.assert_called_once()
     queued_command = controller._write_queue.put.call_args[0][0]