feat: add runtime capability discovery

Implement discovery module for converting OCF resource metadata into bound
entities, with support for multi-instance device resources and logging of
unknown capabilities.

samsung_appliance/registry/discovery.py

@@ -0,0 +1,52 @@
+"""Runtime discovery: device resources -> bound entities.
+
+A device advertises a set of OCF resources, each carrying one or more `rt`
+values. For every rt present in the registry, emit the capability's entities
+bound to that resource href. Unknown rt is a coverage gap, logged and skipped.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+
+from .capability import Capability
+from .entities import SamsungEntityDescription
+
+
+@dataclass
+class BoundEntity:
+    href: str
+    capability: Capability
+    desc: SamsungEntityDescription
+    instance: str = ''
+
+
+def instance_suffix(href: str) -> str:
+    """'' for the index-0 instance, else '_<n>' from the trailing segment."""
+    tail = href.rstrip('/').rsplit('/', 1)[-1]
+    if tail.isdigit() and tail != '0':
+        return f'_{tail}'
+    return ''
+
+
+def discover(resources: dict[str, dict],
+             registry: dict[str, Capability],
+             log: Optional[Callable[[str], None]] = None) -> list[BoundEntity]:
+    out: list[BoundEntity] = []
+    for href, rep in resources.items():
+        if not isinstance(rep, dict):
+            continue
+        rts = rep.get('rt') or []
+        for rt in rts:
+            cap = registry.get(rt)
+            if cap is None:
+                continue
+            inst = instance_suffix(href)
+            for desc in cap.entities:
+                out.append(BoundEntity(href=href, capability=cap,
+                                       desc=desc, instance=inst))
+        # Coverage logging: a resource none of whose rts are known.
+        if log is not None and not any(rt in registry for rt in rts):
+            for rt in rts:
+                log(f"unknown capability {rt} at {href}")
+    return out

tests/test_discovery.py

@@ -0,0 +1,47 @@
+from samsung_appliance.registry.capability import Capability
+from samsung_appliance.registry.discovery import (
+    BoundEntity, discover, instance_suffix,
+)
+from samsung_appliance.registry.entities import BinarySensorDesc
+
+LOCK = Capability(
+    rt='x.com.samsung.da.kidsLock',
+    entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),),
+)
+REG = {LOCK.rt: LOCK}
+
+
+def test_instance_suffix():
+    assert instance_suffix('/kidslock/vs/0') == ''
+    assert instance_suffix('/door/vs/1') == '_1'
+    assert instance_suffix('/icemaker/vs/2') == '_2'
+
+
+def test_discover_binds_present_capability():
+    resources = {'/kidslock/vs/0': {'rt': ['x.com.samsung.da.kidsLock'],
+                                     'x.com.samsung.da.kidsLock': 'On'}}
+    bound = discover(resources, REG)
+    assert len(bound) == 1
+    assert bound[0].href == '/kidslock/vs/0'
+    assert bound[0].desc.key == 'child_lock'
+    assert bound[0].instance == ''
+
+
+def test_discover_skips_unknown_rt():
+    seen = []
+    resources = {'/mystery/vs/0': {'rt': ['x.com.samsung.da.mystery']}}
+    bound = discover(resources, REG, log=seen.append)
+    assert bound == []
+    assert any('mystery' in m for m in seen)
+
+
+def test_discover_multi_instance_suffixes():
+    cap = Capability(rt='x.com.samsung.da.door',
+                     entities=(BinarySensorDesc(key='door', field='x.com.samsung.da.doorState'),))
+    resources = {
+        '/door/vs/0': {'rt': ['x.com.samsung.da.door']},
+        '/door/vs/1': {'rt': ['x.com.samsung.da.door']},
+    }
+    bound = discover(resources, {cap.rt: cap})
+    insts = sorted(b.instance for b in bound)
+    assert insts == ['', '_1']