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

refactor: key capabilities on href instead of rt

Browse Source 
OCF rt is a schema category shared by many unrelated resources.
The stable unique resource address is the href. Update Capability,
discovery, common capabilities, tests, and plan accordingly.
This commit is contained in:
Marc Billow
2026-06-30 16:32:37 -05:00
parent bdae3cc82d
commit 0499c702d7
6 changed files with 62 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
samsung_appliance/registry/capabilities/common.py
+21 -26
View File
@@ -1,21 +1,17 @@
"""Capabilities shared across multiple appliance families.
rt strings verified against live device dumps:
/kidslock/vs/0 -> x.com.samsung.da.operation
/remotectrl/vs/0 -> x.com.samsung.da.configuration
/power/vs/0 -> x.com.samsung.da.operation (shares rt with kidslock)
/alarms/vs/0 -> x.com.samsung.da.alarms
/energy/consumption/vs/0 -> x.com.samsung.da.energyconsumption
/water/consumption/vs/0 -> x.com.samsung.da.waterconsumption
/filter/waterfilter/vs/0 -> x.com.samsung.da.filter.water
NOTE: KIDS_LOCK and POWER share the same rt (x.com.samsung.da.operation).
When building a registry dict keyed by rt, only one can be present.
They target different hrefs (/kidslock/vs/0 vs /power/vs/0) with different
field keys — at runtime the absent field returns None and is filtered.
Each capability is keyed on the stable OCF resource href (not rt), verified
against live device dumps:
/kidslock/vs/0 -> x.com.samsung.da.kidsLock
/remotectrl/vs/0 -> x.com.samsung.da.remoteControlEnabled
/power/vs/0 -> x.com.samsung.da.power
/alarms/vs/0 -> x.com.samsung.da.items
/energy/consumption/vs/0 -> x.com.samsung.da.instantaneousPower / cumulativePower
/water/consumption/vs/0 -> x.com.samsung.da.cumulativeWater
/filter/waterfilter/vs/0 -> x.com.samsung.da.filterUsage / filterStatus
"""
from ..capability import Capability
from ..entities import BinarySensorDesc, SensorDesc
from ..entities import BinarySensorDesc, SensorDesc, SwitchDesc
def _num(v):
@@ -47,7 +43,7 @@ def _last_alarm_code(items):
KIDS_LOCK = Capability(
rt='x.com.samsung.da.operation',
href='/kidslock/vs/0',
entities=(
BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock',
name='Child lock', device_class='lock',
@@ -56,7 +52,7 @@ KIDS_LOCK = Capability(
)
REMOTE_CONTROL = Capability(
rt='x.com.samsung.da.configuration',
href='/remotectrl/vs/0',
entities=(
BinarySensorDesc(key='remote_control',
field='x.com.samsung.da.remoteControlEnabled',
@@ -65,19 +61,18 @@ REMOTE_CONTROL = Capability(
),
)
# POWER shares rt with KIDS_LOCK (x.com.samsung.da.operation).
# Do not include both in the same registry dict simultaneously.
POWER = Capability(
rt='x.com.samsung.da.operation',
href='/power/vs/0',
entities=(
BinarySensorDesc(key='power_state', field='x.com.samsung.da.power',
name='Power', device_class='power',
value_fn=lambda v: v == 'On'),
SwitchDesc(key='power_switch', field='x.com.samsung.da.power',
name='Power',
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep: (['/power/vs/0'], {'x.com.samsung.da.power': 'On' if p else 'Off'})),
),
)
ALARMS = Capability(
rt='x.com.samsung.da.alarms',
href='/alarms/vs/0',
entities=(
SensorDesc(key='alarm_code', field='x.com.samsung.da.items',
name='Alarm code', icon='mdi:alert',
@@ -86,7 +81,7 @@ ALARMS = Capability(
)
ENERGY_METER = Capability(
rt='x.com.samsung.da.energyconsumption',
href='/energy/consumption/vs/0',
entities=(
SensorDesc(key='power_watts', field='x.com.samsung.da.instantaneousPower',
name='Power', device_class='power', state_class='measurement',
@@ -98,7 +93,7 @@ ENERGY_METER = Capability(
)
WATER_METER = Capability(
rt='x.com.samsung.da.waterconsumption',
href='/water/consumption/vs/0',
entities=(
SensorDesc(key='water_liters', field='x.com.samsung.da.cumulativeWater',
name='Water consumption', device_class='water',
@@ -108,7 +103,7 @@ WATER_METER = Capability(
)
WATER_FILTER = Capability(
rt='x.com.samsung.da.filter.water',
href='/filter/waterfilter/vs/0',
entities=(
SensorDesc(key='filter_usage', field='x.com.samsung.da.filterUsage',
name='Filter usage', unit='%', state_class='measurement',
samsung_appliance/registry/capability.py
+2 -2
View File
@@ -1,4 +1,4 @@
"""A Capability binds one OCF resource type (rt) to the entities it produces."""
"""A Capability binds one OCF resource href to the entities it produces."""
from __future__ import annotations
from dataclasses import dataclass
@@ -9,7 +9,7 @@ from .entities import SamsungEntityDescription
@dataclass(frozen=True, kw_only=True)
class Capability:
rt: str
href: str
entities: tuple[SamsungEntityDescription, ...]
poll_tier: str = 'warm' # 'hot' | 'warm' | 'cold'
observe: bool = True
samsung_appliance/registry/discovery.py
+13 -15
View File
@@ -1,8 +1,8 @@
"""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.
A device advertises a set of OCF resources keyed by href. For every href
present in the registry, emit the capability's entities bound to that resource.
Unknown hrefs are a coverage gap, logged at debug and skipped.
"""
from __future__ import annotations
@@ -30,21 +30,19 @@ def instance_suffix(href: str) -> str:
def discover(resources: dict[str, dict],
registry: dict[str, Capability],
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:
if log is not None:
log(f"unknown capability {rt} at {href}")
continue
inst = instance_suffix(href)
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap,
desc=desc, instance=inst))
cap = registry.get(href)
if cap is None:
if log is not None:
log(f"unknown resource {href}")
continue
inst = instance_suffix(href)
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap,
desc=desc, instance=inst))
return out
tests/test_capability.py
+3 -3
View File
@@ -3,7 +3,7 @@ from samsung_appliance.registry.entities import BinarySensorDesc
def test_capability_defaults():
c = Capability(rt='x.com.samsung.da.kidsLock',
c = Capability(href='/kidslock/vs/0',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),))
assert c.poll_tier == 'warm'
assert c.observe is True
@@ -12,9 +12,9 @@ def test_capability_defaults():
def test_capability_is_frozen():
c = Capability(rt='x', entities=())
c = Capability(href='/kidslock/vs/0', entities=())
try:
c.rt = 'y'
c.href = '/other/vs/0'
except Exception:
return
raise AssertionError("expected frozen dataclass")
tests/test_common_capabilities.py
+4 -3
View File
@@ -3,9 +3,10 @@ from samsung_appliance.registry.discovery import discover
def _reg():
return {c.rt: c for c in (
common.KIDS_LOCK, common.REMOTE_CONTROL, common.ALARMS,
common.ENERGY_METER, common.WATER_METER, common.WATER_FILTER,
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
+19 -17
View File
@@ -5,10 +5,10 @@ from samsung_appliance.registry.discovery import (
from samsung_appliance.registry.entities import BinarySensorDesc
LOCK = Capability(
rt='x.com.samsung.da.kidsLock',
href='/kidslock/vs/0',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),),
)
REG = {LOCK.rt: LOCK}
REG = {LOCK.href: LOCK}
def test_instance_suffix():
@@ -18,8 +18,7 @@ def test_instance_suffix():
def test_discover_binds_present_capability():
resources = {'/kidslock/vs/0': {'rt': ['x.com.samsung.da.kidsLock'],
'x.com.samsung.da.kidsLock': 'On'}}
resources = {'/kidslock/vs/0': {'x.com.samsung.da.kidsLock': 'On'}}
bound = discover(resources, REG)
assert len(bound) == 1
assert bound[0].href == '/kidslock/vs/0'
@@ -27,32 +26,35 @@ def test_discover_binds_present_capability():
assert bound[0].instance == ''
def test_discover_skips_unknown_rt():
def test_discover_skips_unknown_href():
seen = []
resources = {'/mystery/vs/0': {'rt': ['x.com.samsung.da.mystery']}}
resources = {'/mystery/vs/0': {'x.com.samsung.da.mystery': 'x'}}
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',
cap = Capability(href='/door/vs/0',
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']},
'/door/vs/0': {'x.com.samsung.da.doorState': 'Open'},
'/door/vs/1': {'x.com.samsung.da.doorState': 'Closed'},
}
bound = discover(resources, {cap.rt: 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']
def test_discover_logs_only_unknown_rt_in_mixed_resource():
def test_discover_logs_unregistered_href():
seen = []
cap = Capability(rt='x.com.samsung.da.kidsLock',
cap = Capability(href='/kidslock/vs/0',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),))
resources = {'/kidslock/vs/0': {'rt': ['x.com.samsung.da.kidsLock', 'x.com.samsung.da.mystery']}}
bound = discover(resources, {cap.rt: cap}, log=seen.append)
assert len(bound) == 1 # known rt still produces entity
assert any('mystery' in m for m in seen) # unknown rt is logged
assert not any('kidsLock' in m for m in seen) # known rt is NOT logged
resources = {
'/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)
assert len(bound) == 1
assert any('mystery' in m for m in seen)