bhetherman/localthings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update discover() for pattern caps, match_fn, rt_filter gating

Browse Source 
Registry changes from dict[str,Capability] to dict[str,list[Capability]].
discover() gains pattern_caps parameter (fallback for unmatched hrefs)
and gates each candidate on rt_filter and match_fn before binding.
Duplicate hrefs in registry allowed only if all caps set a discriminator.
This commit is contained in:
Marc Billow
2026-06-30 16:32:42 -05:00
parent 211fcd1324
commit b10cfad194
7 changed files with 176 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
samsung_appliance/registry/discovery.py
+44 -12
View File
@@ -7,7 +7,7 @@ Unknown hrefs are a coverage gap, logged at debug and skipped.
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Optional
from typing import Callable, Iterable, Optional
from .capability import Capability
from .entities import SamsungEntityDescription
@@ -30,20 +30,52 @@ def instance_suffix(href: str) -> str:
return ''
def discover(resources: dict[str, dict],
registry: dict[str, 'Capability'],
log: Optional[Callable[[str], None]] = None) -> list[BoundEntity]:
def discover(
resources: dict[str, dict],
registry: dict[str, list[Capability]],
pattern_caps: Iterable[Capability] = (),
log: Optional[Callable[[str], None]] = None,
) -> list[BoundEntity]:
out: list[BoundEntity] = []
bound_hrefs: set[str] = set()
for href, rep in resources.items():
if not isinstance(rep, dict):
continue
cap = registry.get(href)
if cap is None:
if log is not None:
log(f"unknown resource {href}")
rts = rep.get('rt') or ()
caps = registry.get(href) or []
matched = False
for cap in caps:
if cap.rt_filter is not None and cap.rt_filter not in rts:
continue
if cap.match_fn is not None and not cap.match_fn(rep, resources):
continue
inst = instance_suffix(href)
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap,
desc=desc, instance=inst))
matched = True
if matched:
bound_hrefs.add(href)
continue
inst = instance_suffix(href)
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap,
desc=desc, instance=inst))
# Pattern cap fallback — first matching pattern wins
for cap in pattern_caps:
if cap.rt_filter is not None and cap.rt_filter not in rts:
continue
if cap.match_fn is not None and not cap.match_fn(rep, resources):
continue
inst = instance_suffix(href)
key = cap.key_fn(href) if cap.key_fn else None
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap, desc=desc,
instance=inst, key_override=key))
matched = True
break
if not matched and log is not None:
log(f"unknown resource {href}")
return out
samsung_appliance/registry/registry.py
+28 -9
View File
@@ -1,27 +1,46 @@
"""Global CAPABILITIES registry: href -> Capability.
"""Global CAPABILITIES registry: href -> list[Capability].
Built from the full list of Capability objects in the capabilities package.
Consumed by discover() at connection time to bind device resources to entities.
Raises ValueError at import if any href appears in multiple capabilities.
Raises ValueError at import if any href group contains an unfiltered cap
alongside other caps (i.e., a cap with neither rt_filter nor match_fn set
in a group with multiple caps).
"""
from .capabilities import ALL
from .capability import Capability
def _build() -> dict[str, Capability]:
"""Build the registry, raising ValueError if any href is duplicated.
def _build() -> dict[str, list[Capability]]:
"""Build the registry, raising ValueError for invalid duplicate hrefs.
Duplicates are allowed only when every cap in the group has at least one
of rt_filter or match_fn set. An unfiltered cap sharing an href with
another cap would be ambiguous.
Skips capabilities with href=None (pattern capabilities, handled elsewhere).
"""
out: dict[str, Capability] = {}
out: dict[str, list[Capability]] = {}
for cap in ALL:
if cap.href is None:
continue
if cap.href in out:
raise ValueError(f"duplicate capability href: {cap.href}")
out[cap.href] = cap
if cap.href not in out:
out[cap.href] = [cap]
else:
out[cap.href].append(cap)
# Validate: every group with >1 cap must have all caps filtered
for href, caps in out.items():
if len(caps) > 1:
unfiltered = [c for c in caps
if c.rt_filter is None and c.match_fn is None]
if unfiltered:
raise ValueError(
f"duplicate capability href {href!r} has unfiltered cap(s); "
f"each cap sharing an href must set rt_filter or match_fn"
)
return out
CAPABILITIES: dict[str, Capability] = _build()
CAPABILITIES: dict[str, list[Capability]] = _build()
tests/test_adapter.py
+2 -2
View File
@@ -6,12 +6,12 @@ from samsung_appliance.registry.discovery import discover
def _bound(resources):
reg = {c.href: c for c in (common.KIDS_LOCK, common.ENERGY_METER)}
reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER)}
return discover(resources, reg)
def _bound_with_power(resources):
reg = {c.href: c for c in (common.KIDS_LOCK, common.ENERGY_METER, common.POWER)}
reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER, common.POWER)}
return discover(resources, reg)
tests/test_common_capabilities.py
+1 -1
View File
@@ -3,7 +3,7 @@ from samsung_appliance.registry.discovery import discover
def _reg():
return {c.href: c for c in (
return {c.href: [c] for c in (
common.KIDS_LOCK, common.REMOTE_CONTROL, common.POWER,
common.ALARMS, common.ENERGY_METER, common.WATER_METER,
common.WATER_FILTER,
tests/test_discovery.py
+95 -3
View File
@@ -8,7 +8,7 @@ LOCK = Capability(
href='/kidslock/vs/0',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),),
)
REG = {LOCK.href: LOCK}
REG = {LOCK.href: [LOCK]}
def test_instance_suffix():
@@ -41,7 +41,7 @@ def test_discover_multi_instance_suffixes():
'/door/vs/0': {'x.com.samsung.da.doorState': 'Open'},
'/door/vs/1': {'x.com.samsung.da.doorState': 'Closed'},
}
reg = {'/door/vs/0': cap, '/door/vs/1': cap}
reg = {'/door/vs/0': [cap], '/door/vs/1': [cap]}
bound = discover(resources, reg)
insts = sorted(b.instance for b in bound)
assert insts == ['', '_1']
@@ -55,6 +55,98 @@ def test_discover_logs_unregistered_href():
'/kidslock/vs/0': {'x.com.samsung.da.kidsLock': 'On'},
'/mystery/vs/0': {'x.com.samsung.da.mystery': 'x'},
}
bound = discover(resources, {'/kidslock/vs/0': cap}, log=seen.append)
bound = discover(resources, {'/kidslock/vs/0': [cap]}, log=seen.append)
assert len(bound) == 1
assert any('mystery' in m for m in seen)
# ---------------------------------------------------------------------------
# New tests for Task 2: pattern caps, rt_filter, match_fn
# ---------------------------------------------------------------------------
def test_discover_pattern_cap_binds_unmatched_href():
"""Pattern cap with rt_filter binds unmatched hrefs; exact-href cap does not
steal unmatched hrefs."""
cooler_cap = Capability(
href='/door/cooler/0',
entities=(BinarySensorDesc(key='cooler_door', field='x.com.samsung.da.doorState'),),
)
door_pattern = Capability(
href=None,
rt_filter='oic.r.door',
key_fn=lambda href: href.split('/')[-2],
entities=(BinarySensorDesc(key='door', field='x.com.samsung.da.doorState'),),
)
resources = {
'/door/cooler/0': {'x.com.samsung.da.doorState': 'Closed', 'rt': ['oic.r.door']},
'/door/wine/0': {'x.com.samsung.da.doorState': 'Open', 'rt': ['oic.r.door']},
}
reg = {'/door/cooler/0': [cooler_cap]}
bound = discover(resources, reg, pattern_caps=[door_pattern])
hrefs = [b.href for b in bound]
# exact cap claims /door/cooler/0
assert '/door/cooler/0' in hrefs
# pattern cap claims /door/wine/0 (unmatched by registry)
assert '/door/wine/0' in hrefs
# /door/cooler/0 is not also claimed by the pattern cap
cooler_bindings = [b for b in bound if b.href == '/door/cooler/0']
assert all(b.capability is cooler_cap for b in cooler_bindings)
def test_discover_pattern_cap_skips_already_bound_href():
"""Pattern cap must not bind an href already claimed by an exact-href cap."""
exact_cap = Capability(
href='/door/cooler/0',
entities=(BinarySensorDesc(key='cooler_door', field='x.com.samsung.da.doorState'),),
)
pattern = Capability(
href=None,
rt_filter='oic.r.door',
entities=(BinarySensorDesc(key='generic_door', field='x.com.samsung.da.doorState'),),
)
resources = {'/door/cooler/0': {'x.com.samsung.da.doorState': 'Closed', 'rt': ['oic.r.door']}}
reg = {'/door/cooler/0': [exact_cap]}
bound = discover(resources, reg, pattern_caps=[pattern])
# exactly one binding for /door/cooler/0 — the exact cap, not the pattern
assert len(bound) == 1
assert bound[0].capability is exact_cap
def test_discover_match_fn_filters_wrong_device():
"""A cap with match_fn must not bind when its condition is not met."""
oven_cap = Capability(
href='/oven/vs/0',
match_fn=lambda r, rs: '/oven/vs/0' in rs,
entities=(BinarySensorDesc(key='oven_status', field='x.com.samsung.da.state'),),
)
# resources does NOT include /oven/vs/0 — match_fn should fail
resources = {'/oven/vs/0': {'x.com.samsung.da.state': 'Ready'}}
reg = {'/oven/vs/0': [oven_cap]}
# Without /oven/vs/0 present as a key in resources, match_fn should be False
resources_no_oven = {'/other/vs/0': {'x.com.samsung.da.state': 'Ready'}}
reg_no_oven = {}
# Put oven cap on a different href to test match_fn rejection
oven_cap2 = Capability(
href='/other/vs/0',
match_fn=lambda r, rs: '/oven/vs/0' in rs,
entities=(BinarySensorDesc(key='oven_status', field='x.com.samsung.da.state'),),
)
bound = discover(resources_no_oven, {'/other/vs/0': [oven_cap2]})
# match_fn returns False → no bindings
assert bound == []
def test_discover_rt_filter_gates_binding():
"""Cap with rt_filter must not bind a rep whose rt list does not match."""
oven_mode_cap = Capability(
href='/mode/vs/0',
rt_filter='x.com.samsung.da.ovenMode',
entities=(BinarySensorDesc(key='oven_mode', field='x.com.samsung.da.mode'),),
)
# rep has a different rt → should not bind
resources = {'/mode/vs/0': {'rt': ['x.com.samsung.da.mode'], 'x.com.samsung.da.mode': 'Bake'}}
bound = discover(resources, {'/mode/vs/0': [oven_mode_cap]})
assert bound == []
tests/test_oven_capabilities.py
+2 -2
View File
@@ -158,7 +158,7 @@ def test_cook_time_rejects_out_of_range():
def test_adapter_sets_cycle_active_field_for_oven():
"""build_runtime_descriptor must set cycle_active_field when oven keys present."""
reg = {c.href: c for c in (oven.OVEN_OPERATIONAL_STATE, oven.OVEN_SETPOINT)}
reg = {c.href: [c] for c in (oven.OVEN_OPERATIONAL_STATE, oven.OVEN_SETPOINT)}
resources = {
'/operational/state/vs/0': {'x.com.samsung.da.state': 'Run'},
'/temperatures/vs/0': {'x.com.samsung.da.items': [
@@ -177,7 +177,7 @@ def test_adapter_sets_cycle_active_field_for_oven():
def test_adapter_no_cycle_active_without_oven_keys():
"""Non-oven appliances must not get cycle_active_field."""
from samsung_appliance.registry.capabilities import common
reg = {c.href: c for c in (common.KIDS_LOCK,)}
reg = {c.href: [c] for c in (common.KIDS_LOCK,)}
resources = {'/kidslock/vs/0': {'x.com.samsung.da.kidsLock': 'Ready'}}
bound = discover(resources, reg)
rd = build_runtime_descriptor(
tests/test_registry.py
+4 -3
View File
@@ -3,9 +3,10 @@ from samsung_appliance.registry.registry import CAPABILITIES
def test_registry_is_keyed_by_href():
"""Registry keys should match capability hrefs."""
for href, cap in CAPABILITIES.items():
assert cap.href == href
"""Registry keys should match capability hrefs for all caps in the group."""
for href, caps in CAPABILITIES.items():
for cap in caps:
assert cap.href == href
def test_registry_has_no_duplicate_href():