Review improvements

Author: CamW

paradox/hardware/evo/panel.py

@@ -316,7 +316,8 @@ async def control_module_pgm_outputs(self, module_address: int, pgm_index: int,
         :param str command: textual command
         :return: True if accepted
         """
-        pgm_commands = ["release"] * parsers.MODULE_PGM_OUTPUTS_PER_MODULE
+        assert 1 <= pgm_index <= parsers.MODULE_PGM_PACKET_SLOTS, "pgm_index must be between 1 and %d" % parsers.MODULE_PGM_PACKET_SLOTS
+        pgm_commands = ["release"] * parsers.MODULE_PGM_PACKET_SLOTS
         pgm_commands[pgm_index - 1] = command
 
         args = {"module_address": module_address, "pgm_commands": pgm_commands}

paradox/hardware/evo/parsers.py

@@ -823,7 +823,9 @@ def _parse(self, stream, context, path):
     "checksum" / PacketChecksum(Bytes(1)),
 )
 
-MODULE_PGM_OUTPUTS_PER_MODULE = 4
+# Fixed number of PGM command slots in the 0xA4 packet — a protocol constant,
+# not the number of PGMs a specific module exposes (which is configured separately).
+MODULE_PGM_PACKET_SLOTS = 4
 
 PerformModulePGMAction = Struct(
     "fields"
@@ -834,7 +836,7 @@ def _parse(self, stream, context, path):
             "_not_used0" / Padding(1),
             "module_address" / Default(Int8ub, 0),
             "_not_used1" / Padding(2),
-            "pgm_commands" / Array(MODULE_PGM_OUTPUTS_PER_MODULE, Default(_PGMCommandEnum, "release")),
+            "pgm_commands" / Array(MODULE_PGM_PACKET_SLOTS, Default(_PGMCommandEnum, "release")),
             "_not_used2" / Padding(12),
         )
     ),

tests/test_paradox.py

@@ -39,3 +39,71 @@ async def test_control_doors(mocker):
 
     assert await alarm.control_door("Door 1", "unlock")
     alarm.panel.control_doors.assert_called_once_with([1], "unlock")
+
+
+def _add_module_pgm(alarm, module_address, pgm_index):
+    key = f"module{module_address}_pgm{pgm_index}"
+    alarm.storage.get_container("module_pgm")[key] = {
+        "id": pgm_index,
+        "key": key,
+        "label": f"Module {module_address} PGM {pgm_index}",
+        "module_address": module_address,
+        "pgm_index": pgm_index,
+    }
+    alarm.storage.update_container_object("module_pgm", key, {"on": False})
+    return key
+
+
+@pytest.mark.asyncio
+async def test_control_output_regular_pgm_routes_to_control_outputs(mocker):
+    """Regular PGM uses control_outputs, not control_module_pgm_outputs."""
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+    alarm.panel.control_outputs = AsyncMock(return_value=True)
+    alarm.panel.control_module_pgm_outputs = AsyncMock(return_value=True)
+
+    alarm.storage.get_container("pgm").deep_merge({1: {"id": 1, "key": "PGM 1"}})
+
+    assert await alarm.control_output("PGM 1", "on")
+    alarm.panel.control_outputs.assert_called_once()
+    alarm.panel.control_module_pgm_outputs.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_control_output_module_pgm_routes_to_control_module_pgm_outputs(mocker):
+    """Module PGM uses control_module_pgm_outputs with the correct address and index."""
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+    alarm.panel.control_outputs = AsyncMock(return_value=True)
+    alarm.panel.control_module_pgm_outputs = AsyncMock(return_value=True)
+
+    _add_module_pgm(alarm, module_address=4, pgm_index=2)
+
+    assert await alarm.control_output("module4_pgm2", "on_override")
+    alarm.panel.control_module_pgm_outputs.assert_called_once_with(4, 2, "on_override")
+    alarm.panel.control_outputs.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_control_output_module_pgm_optimistic_state(mocker):
+    """Accepted on_override sets on=True; off_override sets on=False."""
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+    alarm.panel.control_module_pgm_outputs = AsyncMock(return_value=True)
+
+    key = _add_module_pgm(alarm, module_address=4, pgm_index=1)
+
+    await alarm.control_output(key, "on_override")
+    assert alarm.storage.get_container("module_pgm")[key]["on"] is True
+
+    await alarm.control_output(key, "off_override")
+    assert alarm.storage.get_container("module_pgm")[key]["on"] is False
+
+
+@pytest.mark.asyncio
+async def test_control_output_no_match_returns_false(mocker):
+    """Returns False when the output key matches neither pgm nor module_pgm."""
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+
+    assert await alarm.control_output("nonexistent", "on") is False