HA: discover programs over the wire + diagnostic sensor

Coordinator's _discover_programs is no longer a placeholder. It now
drives client.iter_programs() (v2 path) or the v1 adapter's forward
to OmniClientV1.iter_programs (v1 path), populating OmniData.programs
with decoded Program records keyed by slot. Errors are logged and
swallowed so partial enumeration doesn't break entry setup —
programs are non-critical telemetry.

OmniData.programs is now dict[int, Program] rather than the
ProgramProperties dict that was an empty placeholder. The
ProgramProperties dataclass remains in models.py for the Properties
opcode reply path; only the coordinator's value type changed.

New OmniProgramsSensor on the sensor platform: a single diagnostic
entity per panel whose state is the count of defined programs and
whose 'programs' attribute lists each program's slot, type name,
and schedule fields. Easy to consume from automations and the
developer-states UI.

Mock fixture seeds three programs (TIMED+TIMED+EVENT at slots 12 /
42 / 99). New integration test verifies the sensor enumerates them
in slot-ascending order with the expected per-record fields.

Full suite: 494 passed, 1 skipped (fixture-gated).

custom_components/omni_pca/coordinator.py

@@ -62,7 +62,6 @@ from omni_pca.models import (
     AreaStatus,
     ButtonProperties,
     ObjectType,
-    ProgramProperties,
     SystemInformation,
     SystemStatus,
     ThermostatProperties,
@@ -73,6 +72,7 @@ from omni_pca.models import (
     ZoneStatus,
 )
 from omni_pca.opcodes import OmniLink2MessageType
+from omni_pca.programs import Program
 
 from .const import (
     DOMAIN,
@@ -108,7 +108,7 @@ class OmniData:
     areas: dict[int, AreaProperties] = field(default_factory=dict)
     thermostats: dict[int, ThermostatProperties] = field(default_factory=dict)
     buttons: dict[int, ButtonProperties] = field(default_factory=dict)
-    programs: dict[int, ProgramProperties] = field(default_factory=dict)
+    programs: dict[int, Program] = field(default_factory=dict)
 
     zone_status: dict[int, ZoneStatus] = field(default_factory=dict)
     unit_status: dict[int, UnitStatus] = field(default_factory=dict)
@@ -382,15 +382,32 @@ class OmniDataUpdateCoordinator(DataUpdateCoordinator[OmniData]):
 
     async def _discover_programs(
         self, client: OmniClient
-    ) -> dict[int, ProgramProperties]:
+    ) -> dict[int, Program]:
-        # Programs aren't reachable via the Properties opcode (the C# side
+        """Enumerate defined panel programs via the appropriate wire path.
-        # uses a separate request/reply pair), so we just return an empty
+
-        # dict. We keep the field on OmniData so Phase B can plug in real
+        * v2 (TCP): ``client.iter_programs()`` drives UploadProgram with
-        # discovery the moment the library exposes it. AMBIGUITY: the spec
+          request_reason=1 ("next defined after slot"), stopping at EOD.
-        # asks for "named programs" — there's no on-the-wire path for that
+        * v1 (UDP): the adapter forwards to OmniClientV1.iter_programs(),
-        # in v1.0 of omni_pca, so an empty mapping is the honest answer.
+          which is a bare UploadPrograms stream ack-walked to EOD.
-        _ = client, ProgramProperties
+
-        return {}
+        Both surfaces yield :class:`omni_pca.programs.Program` and skip
+        empty slots, so the resulting dict only carries defined programs.
+        Errors during enumeration are logged and swallowed — programs
+        are a non-critical part of discovery, so a partial list is better
+        than blocking the entry setup.
+        """
+        out: dict[int, Program] = {}
+        try:
+            async for prog in client.iter_programs():
+                if prog.slot is not None:
+                    out[prog.slot] = prog
+        except (OmniConnectionError, RequestTimeoutError):
+            raise
+        except Exception:
+            LOGGER.debug(
+                "program enumeration interrupted (kept %d)", len(out), exc_info=True
+            )
+        return out
 
     async def _walk_properties(
         self,

custom_components/omni_pca/sensor.py

@@ -69,6 +69,7 @@ async def async_setup_entry(
 
     entities.append(OmniSystemModelSensor(coordinator))
     entities.append(OmniLastEventSensor(coordinator))
+    entities.append(OmniProgramsSensor(coordinator))
 
     async_add_entities(entities)
 
@@ -261,3 +262,55 @@ class OmniLastEventSensor(
             if hasattr(ev, key):
                 result[key] = getattr(ev, key)
         return result
+
+
+class OmniProgramsSensor(
+    CoordinatorEntity[OmniDataUpdateCoordinator], SensorEntity
+):
+    """Diagnostic sensor exposing the panel's automation programs.
+
+    State value is the count of defined programs. The ``programs``
+    attribute carries a list of per-program summaries — a stable,
+    JSON-serializable view automations and template sensors can read.
+    """
+
+    _attr_has_entity_name = True
+    _attr_entity_category = EntityCategory.DIAGNOSTIC
+    _attr_state_class = SensorStateClass.MEASUREMENT
+
+    def __init__(self, coordinator: OmniDataUpdateCoordinator) -> None:
+        super().__init__(coordinator)
+        self._attr_unique_id = f"{coordinator.unique_id}-programs"
+        self._attr_name = "Panel Programs"
+        self._attr_device_info = coordinator.device_info
+
+    @property
+    def native_value(self) -> int:
+        return len(self.coordinator.data.programs)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        from omni_pca.programs import ProgramType
+
+        summaries: list[dict[str, Any]] = []
+        for slot in sorted(self.coordinator.data.programs):
+            p = self.coordinator.data.programs[slot]
+            try:
+                type_name = ProgramType(p.prog_type).name
+            except ValueError:
+                type_name = f"UNKNOWN({p.prog_type})"
+            summaries.append(
+                {
+                    "slot": slot,
+                    "type": type_name,
+                    "cmd": p.cmd,
+                    "par": p.par,
+                    "pr2": p.pr2,
+                    "month": p.month,
+                    "day": p.day,
+                    "days": p.days,
+                    "hour": p.hour,
+                    "minute": p.minute,
+                }
+            )
+        return {"programs": summaries}

src/omni_pca/v1/adapter.py

@@ -172,6 +172,16 @@ class OmniClientV1Adapter:
     async def list_message_names(self) -> dict[int, str]:
         return (await self._ensure_names()).get(7, {})  # NameType.MESSAGE
 
+    # ---- programs ------------------------------------------------------
+
+    def iter_programs(self):
+        """Forward to OmniClientV1.iter_programs (streaming UploadPrograms).
+
+        Same async-iterator shape as :meth:`OmniClient.iter_programs` so the
+        coordinator does not need a transport branch.
+        """
+        return self._client.iter_programs()
+
     # ---- properties synthesis ------------------------------------------
 
     async def get_object_properties(

tests/ha_integration/conftest.py

@@ -64,6 +64,26 @@ def _short_scan_interval(monkeypatch: pytest.MonkeyPatch) -> None:
 @pytest.fixture
 def populated_state() -> MockState:
     """A lightly-populated mock state covering every entity platform."""
+    from omni_pca.programs import Days, Program, ProgramType
+    programs = {
+        slot: prog.encode_wire_bytes()
+        for slot, prog in {
+            12: Program(
+                slot=12, prog_type=int(ProgramType.TIMED),
+                cmd=3, hour=6, minute=0,
+                days=int(Days.MONDAY | Days.FRIDAY),
+            ),
+            42: Program(
+                slot=42, prog_type=int(ProgramType.TIMED),
+                cmd=4, hour=22, minute=30,
+                days=int(Days.SUNDAY),
+            ),
+            99: Program(
+                slot=99, prog_type=int(ProgramType.EVENT),
+                cmd=5, month=5, day=12,
+            ),
+        }.items()
+    }
     return MockState(
         zones={
             1: MockZoneState(name="FRONT_DOOR"),
@@ -84,6 +104,7 @@ def populated_state() -> MockState:
             1: MockButtonState(name="GOOD_MORNING"),
         },
         user_codes={1: 1234},
+        programs=programs,
     )
 
 

tests/ha_integration/test_setup.py

@@ -78,6 +78,35 @@ async def test_button_entity_for_panel_button(
     assert len(states) == 1
 
 
+async def test_programs_sensor_reflects_seeded_panel(
+    hass: HomeAssistant, configured_panel
+) -> None:
+    """The diagnostic Panel Programs sensor enumerates discovered programs.
+
+    Mock seed has 3 programs at slots 12 / 42 / 99 (see
+    :func:`populated_state`); after discovery the coordinator stashes them
+    on ``OmniData.programs`` and the sensor surfaces a count + per-slot
+    summary attribute.
+    """
+    programs_states = [
+        s for s in hass.states.async_all("sensor")
+        if s.entity_id.endswith("_panel_programs") or "panel_programs" in s.entity_id
+    ]
+    assert len(programs_states) == 1, (
+        f"expected one panel_programs sensor, got {[s.entity_id for s in programs_states]}"
+    )
+    s = programs_states[0]
+    assert int(s.state) == 3
+    summaries = s.attributes["programs"]
+    assert [p["slot"] for p in summaries] == [12, 42, 99]
+    # Spot-check the per-record fields landed in summary form.
+    by_slot = {p["slot"]: p for p in summaries}
+    assert by_slot[12]["type"] == "TIMED"
+    assert by_slot[12]["hour"] == 6 and by_slot[12]["minute"] == 0
+    assert by_slot[99]["type"] == "EVENT"
+    assert by_slot[99]["month"] == 5 and by_slot[99]["day"] == 12
+
+
 async def test_event_entity_per_panel(
     hass: HomeAssistant, configured_panel
 ) -> None: