Add Combus module PGM control (MODULE_PGM_ADDRESSES)

   Implements control of PGMs on Paradox expansion modules.  (PGM4, zone extenders with PGM outputs, etc.) using the 0xA4 command class. Paradox handles "Module" PGMs differently to "Panel" PGMs which are already supported in pai.

   Protocol:
   - TX: A4 17 00 <module_addr> 00 00 <pgm1> <pgm2> <pgm3> <pgm4> 00*12 <cs>
   - RX: A2 07 00 <module_addr> 00 00 <cs>
   - Module address = Babyware enrollment slot index (1-254)
   - No RAM status block; state tracked optimistically after each command

   Changes:
   - parsers.py: PerformModulePGMAction + PerformModulePGMActionResponse structs,
     MODULE_PGM_OUTPUTS_PER_MODULE constant
   - panel.py (evo): parse_message dispatch for 0xA4/0xA, control_module_pgm_outputs()
   - panel.py (base): abstract control_module_pgm_outputs stub
   - config.py: MODULE_PGM_ADDRESSES dict config (address -> pgm_count)
   - paradox.py: _init_module_pgms(), extended control_output() for module_pgm container
   - helpers.py: module_pgm topic mapping
   - homeassistant.py: _publish_module_pgm_configs() for HA autodiscovery
   - switch.py: ModulePGMSwitch with on_override/off_override payloads
   - factory.py: make_module_pgm_switch()
   - test_module_pgm_action.py: 16 tests covering build and parse against captured frames

Author: CamW

paradox/config.py

@@ -57,6 +57,7 @@ class Config:
         "KEEP_ALIVE_INTERVAL": 10,  # Interval between status updates
         "IO_TIMEOUT": 0.5,  # Timeout for IO operations
         "LIMITS": {},  # By default all zones will be monitored
+        "MODULE_PGM_ADDRESSES": ({}, dict, None),  # Map of bus module address -> pgm count, e.g. {4: 4}
         "LABEL_ENCODING": "paradox-en",  # Encoding to use when decoding labels. paradox-* or https://docs.python.org/3/library/codecs.html#standard-encodings
         "LABEL_REFRESH_INTERVAL": (
             15 * 60,

paradox/hardware/evo/panel.py

@@ -128,6 +128,8 @@ def parse_message(
                     return parsers.PerformPartitionAction.parse(message)
                 elif message[0] == 0xD0:
                     return parsers.PerformZoneAction.parse(message)
+                elif message[0] == 0xA4:
+                    return parsers.PerformModulePGMAction.parse(message)
             else:
                 if message[0] >> 4 == 0x7:
                     return parsers.ErrorMessage.parse(message)
@@ -139,6 +141,8 @@ def parse_message(
                     return parsers.PerformActionResponse.parse(message)
                 elif message[0] >> 4 == 0xD:
                     return parsers.PerformZoneActionResponse.parse(message)
+                elif message[0] >> 4 == 0xA:
+                    return parsers.PerformModulePGMActionResponse.parse(message)
                 # elif message[0] == 0x50 and message[2] == 0x80:
                 #     return PanelStatus.parse(message)
                 # elif message[0] == 0x50 and message[2] < 0x80:
@@ -304,6 +308,32 @@ async def control_outputs(self, outputs, command) -> bool:
             logger.info('PGM command: "%s" failed' % command)
         return reply is not None
 
+    async def control_module_pgm_outputs(self, module_address: int, pgm_index: int, command: str) -> bool:
+        """
+        Control PGM module outputs
+        :param int module_address: bus address of the PGM module
+        :param int pgm_index: 1-4 index of the PGM output
+        :param str command: textual command
+        :return: True if accepted
+        """
+        pgm_commands = ["release"] * parsers.MODULE_PGM_OUTPUTS_PER_MODULE
+        pgm_commands[pgm_index - 1] = command
+
+        args = {"module_address": module_address, "pgm_commands": pgm_commands}
+        try:
+            reply = await self.core.send_wait(
+                parsers.PerformModulePGMAction, args, reply_expected=0xA
+            )
+        except MappingError:
+            logger.error('Module PGM command: "%s" is not supported' % command)
+            return False
+
+        if reply:
+            logger.info('Module PGM command: "%s" succeeded' % command)
+        else:
+            logger.info('Module PGM command: "%s" failed' % command)
+        return reply is not None
+
     async def control_doors(self, doors, command) -> bool:
         """
         Control Doors

paradox/hardware/evo/parsers.py

@@ -823,6 +823,42 @@ def _parse(self, stream, context, path):
     "checksum" / PacketChecksum(Bytes(1)),
 )
 
+MODULE_PGM_OUTPUTS_PER_MODULE = 4
+
+PerformModulePGMAction = Struct(
+    "fields"
+    / RawCopy(
+        Struct(
+            "po" / Struct("command" / Const(0xA4, Int8ub)),
+            "packet_length" / PacketLength(Int8ub),
+            "_not_used0" / Padding(1),
+            "module_address" / Default(Int8ub, 0),
+            "_not_used1" / Padding(2),
+            "pgm_commands" / Array(MODULE_PGM_OUTPUTS_PER_MODULE, Default(_PGMCommandEnum, "release")),
+            "_not_used2" / Padding(12),
+        )
+    ),
+    "checksum" / PacketChecksum(Bytes(1)),
+)
+
+PerformModulePGMActionResponse = Struct(
+    "fields"
+    / RawCopy(
+        Struct(
+            "po"
+            / BitStruct(
+                "command" / Const(0xA, Nibble),
+                "_not_used" / Nibble,
+            ),
+            "packet_length" / PacketLength(Int8ub),
+            "_not_used0" / Padding(1),
+            "module_address" / Int8ub,
+            "_not_used1" / Padding(2),
+        )
+    ),
+    "checksum" / PacketChecksum(Bytes(1)),
+)
+
 _PGMBroadcastCommandEnum = Enum(
     Int8ub,
     no_change=0,

paradox/hardware/panel.py

@@ -357,6 +357,10 @@ def control_partitions(self, partitions, command) -> bool:
     def control_outputs(self, outputs, command) -> bool:
         raise NotImplementedError("override control_outputs in a subclass")
 
+    @abstractmethod
+    def control_module_pgm_outputs(self, module_address, pgm_index, command) -> bool:
+        raise NotImplementedError("override control_module_pgm_outputs in a subclass")
+
     @abstractmethod
     def control_doors(self, doors, command) -> bool:
         raise NotImplementedError("override control_doors in a subclass")

paradox/interfaces/mqtt/entities/factory.py

@@ -2,7 +2,7 @@
 from paradox.interfaces.mqtt.entities.binary_sensors import ZoneStatusBinarySensor, \
     SystemBinarySensor, PartitionBinarySensor
 from paradox.interfaces.mqtt.entities.sensor import PAIStatusSensor, SystemStatusSensor, ZoneNumericSensor
-from paradox.interfaces.mqtt.entities.switch import ZoneBypassSwitch, PGMSwitch
+from paradox.interfaces.mqtt.entities.switch import ZoneBypassSwitch, PGMSwitch, ModulePGMSwitch
 
 
 class MQTTAutodiscoveryEntityFactory:
@@ -34,6 +34,9 @@ def make_zone_status_numeric_sensor(self, zone, status):
     def make_pgm_switch(self, pgm):
         return PGMSwitch(pgm, self.device, self.availability_topic)
 
+    def make_module_pgm_switch(self, module_pgm):
+        return ModulePGMSwitch(module_pgm, self.device, self.availability_topic)
+
     def make_system_status(self, system_key, status):
         if system_key == 'troubles':
             return SystemBinarySensor(system_key, status, self.device, self.availability_topic)

paradox/interfaces/mqtt/entities/switch.py

@@ -47,3 +47,23 @@ def __init__(self, pgm, device, availability_topic: str):
         self.property = "on"
 
         self.pai_entity_type = "pgm"
+
+
+class ModulePGMSwitch(Switch):
+    def __init__(self, module_pgm, device, availability_topic: str):
+        super().__init__(device, availability_topic)
+        self.module_pgm = module_pgm
+
+        self.key = sanitize_key(module_pgm["key"])
+        self.label = module_pgm["label"]
+        self.property = "on"
+
+        self.pai_entity_type = "pgm"
+
+    def serialize(self):
+        config = super().serialize()
+        config.update(dict(
+            payload_on="on_override",
+            payload_off="off_override",
+        ))
+        return config

paradox/interfaces/mqtt/helpers.py

@@ -5,6 +5,7 @@
     zone=cfg.MQTT_ZONE_TOPIC,
     output=cfg.MQTT_OUTPUT_TOPIC,
     pgm=cfg.MQTT_OUTPUT_TOPIC,
+    module_pgm=cfg.MQTT_OUTPUT_TOPIC,
     repeater=cfg.MQTT_REPEATER_TOPIC,
     bus=cfg.MQTT_BUS_TOPIC,
     module=cfg.MQTT_MODULE_TOPIC,

paradox/interfaces/mqtt/homeassistant.py

@@ -25,6 +25,7 @@ def __init__(self, alarm):
         self.partitions = {}
         self.zones = {}
         self.pgms = {}
+        self.module_pgms = {}
 
         self.entity_factory = MQTTAutodiscoveryEntityFactory(
             self.mqtt.availability_topic
@@ -69,6 +70,7 @@ def _handle_labels_loaded(self, data):
 
         self.zones = data.get("zone", {})
         self.pgms = data.get("pgm", {})
+        self.module_pgms = data.get("module_pgm", {})
 
     def _publish_when_ready(self, panel: DetectedPanel, status):
         self.entity_factory.set_device(Device(panel))
@@ -80,6 +82,9 @@ def _publish_when_ready(self, panel: DetectedPanel, status):
             self._publish_zone_configs(status["zone"])
         if "pgm" in status:
             self._publish_pgm_configs(status["pgm"])
+        module_pgms = dict(self.alarm.storage.get_container("module_pgm"))
+        if module_pgms:
+            self._publish_module_pgm_configs(module_pgms)
         if "system" in status:
             self._publish_system_property_configs(status["system"])
 
@@ -162,6 +167,11 @@ def _publish_pgm_configs(self, pgm_statuses):
             pgm_switch_config = self.entity_factory.make_pgm_switch(pgm)
             self._publish_config(pgm_switch_config)
 
+    def _publish_module_pgm_configs(self, module_pgms):
+        for module_pgm_key, module_pgm in module_pgms.items():
+            module_pgm_switch_config = self.entity_factory.make_module_pgm_switch(module_pgm)
+            self._publish_config(module_pgm_switch_config)
+
     def _publish_system_property_configs(self, system_statuses):
         for system_key, system_status in system_statuses.items():
             for property_name in system_status:

paradox/paradox.py

@@ -216,6 +216,7 @@ async def full_connect(self) -> bool:
 
             logger.info("Loading data from panel memory")
             await self.panel.load_memory()
+            self._init_module_pgms()
 
             logger.info("Running")
             self.run_state = RunState.RUN
@@ -489,33 +490,61 @@ async def control_partition(self, partition: str, command: str) -> bool:
 
         return accepted
 
+    def _init_module_pgms(self):
+        for addr, pgm_count in cfg.MODULE_PGM_ADDRESSES.items():
+            for pgm_index in range(1, pgm_count + 1):
+                key = f"module{addr}_pgm{pgm_index}"
+                self.storage.get_container("module_pgm")[key] = {
+                    "id": pgm_index,
+                    "key": key,
+                    "label": f"Module {addr} PGM {pgm_index}",
+                    "module_address": addr,
+                    "pgm_index": pgm_index,
+                }
+                self.storage.update_container_object("module_pgm", key, {"on": False})
+
     async def control_output(self, output, command) -> bool:
         command = command.lower()
         logger.debug(f"Control Output: {output} - {command}")
 
         outputs_selected = self.storage.get_container("pgm").select(output)
-
-        # Not Found
-        if len(outputs_selected) == 0:
-            logger.error("No outputs selected")
-            return False
-
-        # Apply state changes
-        accepted = False
-        try:
-            accepted = await self.panel.control_outputs(outputs_selected, command)
-        except NotImplementedError:
-            logger.error("control_output is not implemented for this alarm type")
-        except asyncio.CancelledError:
-            logger.error("control_output canceled")
-        except asyncio.TimeoutError:
-            logger.error("control_output timeout")
-        # Apply state changes
-
-        # Refresh status
-        self.request_status_refresh()  # Trigger status update
-
-        return accepted
+        if outputs_selected:
+            accepted = False
+            try:
+                accepted = await self.panel.control_outputs(outputs_selected, command)
+            except NotImplementedError:
+                logger.error("control_output is not implemented for this alarm type")
+            except asyncio.CancelledError:
+                logger.error("control_output canceled")
+            except asyncio.TimeoutError:
+                logger.error("control_output timeout")
+            self.request_status_refresh()
+            return accepted
+
+        module_pgm_selected = self.storage.get_container("module_pgm").select(output)
+        if module_pgm_selected:
+            accepted = False
+            module_pgm_container = self.storage.get_container("module_pgm")
+            for key in module_pgm_selected:
+                out = module_pgm_container[key]
+                try:
+                    accepted = await self.panel.control_module_pgm_outputs(
+                        out["module_address"], out["pgm_index"], command
+                    )
+                except NotImplementedError:
+                    logger.error("control_module_pgm_outputs is not implemented for this alarm type")
+                except asyncio.CancelledError:
+                    logger.error("control_output canceled")
+                except asyncio.TimeoutError:
+                    logger.error("control_output timeout")
+                if accepted:
+                    is_on = command in ("on", "on_override")
+                    self.storage.update_container_object("module_pgm", out["key"], {"on": is_on})
+            self.request_status_refresh()
+            return accepted
+
+        logger.error("No outputs selected")
+        return False
 
     async def send_panic(self, partition_id, panic_type, user_id) -> bool:
         logger.debug(