feat: fill per-device-type registries; add DOOR_GENERIC, OVEN_CONNECTED, dryer caps

- DOOR_GENERIC pattern cap (rt_filter=oic.r.door) in fridge.py with key_fn
- OVEN_CONNECTED (cloud connectivity) added to oven caps
- dryer.py: DRYER_SETTINGS, DRYER_COURSE, JOB_BEGINNING_STATUS, DRYER_DIAGNOSIS
- Dishwasher, refrigerator, oven, dryer DeviceRegistry objects fully populated
- Refrigerator registry includes DOOR_GENERIC as pattern cap
- Oven registry includes OVEN_SETPOINT + OVEN_MODE (safe: no fridge caps in oven registry)
This commit is contained in:
Marc Billow
2026-06-30 16:32:42 -05:00
parent e801af9ab7
commit 1b36ec4a59
9 changed files with 310 additions and 6 deletions
@@ -2,13 +2,15 @@
from typing import Optional
from ._base import DeviceRegistry
from . import dishwasher, refrigerator
from . import dishwasher, dryer, oven, refrigerator
__all__ = ['DeviceRegistry', '_type_key', 'for_device']
_REGISTRY_BY_KEY: dict[str, DeviceRegistry] = {
'dishwasher': dishwasher.REGISTRY,
'dryer': dryer.REGISTRY,
'oven': oven.REGISTRY,
'refrigerator': refrigerator.REGISTRY,
}
@@ -1,14 +1,20 @@
"""Dishwasher device registry."""
from ..capabilities import common, operational
from ..capabilities import common, fridge, laundry, operational
from ._base import DeviceRegistry, _build
# Placeholder: Task 4 will expand with full capability list.
# For now, include common capabilities to verify the import chain.
REGISTRY = DeviceRegistry(
name='dishwasher',
capabilities=_build([
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
common.ALARMS,
common.ENERGY_METER,
common.WATER_METER,
common.WATER_FILTER,
operational.OPERATIONAL_STATE,
laundry.DOOR_LED,
laundry.SOUND_MODE,
fridge.FIRMWARE_UPDATE,
]),
)
@@ -0,0 +1,26 @@
"""Dryer device registry.
Note: dryer devices use port 49155 (not 49154). Pass default_port=49155
when calling build_runtime_descriptor() for this registry.
"""
from ..capabilities import common, dryer, fridge, laundry, operational
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='dryer',
capabilities=_build([
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
common.ALARMS,
common.ENERGY_METER,
operational.OPERATIONAL_STATE,
laundry.DOOR_LED,
laundry.SOUND_MODE,
dryer.DRYER_SETTINGS,
dryer.DRYER_COURSE,
dryer.JOB_BEGINNING_STATUS,
dryer.DRYER_DIAGNOSIS,
fridge.FIRMWARE_UPDATE,
]),
)
@@ -0,0 +1,20 @@
"""Oven device registry."""
from ..capabilities import common, fridge, oven
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='oven',
capabilities=_build([
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
common.ALARMS,
oven.OVEN_CAVITY,
oven.OVEN_SETPOINT,
oven.OVEN_MODE,
oven.OVEN_OPERATIONAL_STATE,
oven.OVEN_DOOR,
oven.OVEN_CONNECTED,
fridge.FIRMWARE_UPDATE,
]),
)
@@ -2,13 +2,31 @@
from ..capabilities import common, fridge
from ._base import DeviceRegistry, _build
# Placeholder: Task 4 will expand with full capability list.
# For now, include common capabilities to verify the import chain.
REGISTRY = DeviceRegistry(
name='refrigerator',
capabilities=_build([
common.POWER,
common.KIDS_LOCK,
common.REMOTE_CONTROL,
common.ALARMS,
common.ENERGY_METER,
common.WATER_FILTER,
fridge.TEMP_FRIDGE_CURRENT,
fridge.TEMP_FREEZER_CURRENT,
fridge.TEMP_FRIDGE_SETPOINT,
fridge.TEMP_FREEZER_SETPOINT,
fridge.DOOR_FRIDGE,
fridge.DOOR_FREEZER,
fridge.DOORS_STATUS,
fridge.ICEMAKER_ONE,
fridge.ICEMAKER_TWO,
fridge.ICEMAKER_STATUS,
fridge.REFRIGERATION,
fridge.AUTOFILL,
fridge.CABINET_LIGHT,
fridge.SABBATH,
fridge.BEVERAGE_ZONE,
fridge.FIRMWARE_UPDATE,
]),
pattern_capabilities=[fridge.DOOR_GENERIC],
)
@@ -0,0 +1,102 @@
"""Capabilities specific to the dryer family (Samsung DV5000T-class).
Resources derived from the old dryer.py descriptor OBSERVE_PATHS and
flatten() implementation:
/washer/vs/0 -> DRYER_SETTINGS (dryLevel, dryTime, dryerType, wrinklePrevent)
/st/dryercourse/vs/0 -> DRYER_COURSE (dryer_mode SelectDesc)
/wm/jobbeginingstatus/vs/0 -> JOB_BEGINNING_STATUS
/diagnosis/vs/0 -> DRYER_DIAGNOSIS
Course table captured 2026-05-29 on a DA_WM_TP2_20_COMMON_DV5000T. Other
dryer models may use a different course table; options=() means HA renders
whatever the device reports, and the write_fn validates against this table.
"""
from ..capability import Capability
from ..entities import SelectDesc, SensorDesc, SwitchDesc
# Course table: hex codes -> human names (Table_03, DV5000T-class).
_COURSE_NAMES = {
0x16: 'Cotton',
0x18: 'Synthetics',
0x19: 'Delicates',
0x1A: 'Wool',
0x1B: 'Bedding',
0x1C: 'Shirts',
0x1D: 'Towels',
0x1E: 'Outdoor',
0x1F: 'Mixed Load',
0x20: 'Iron Dry',
0x23: 'Quick Dry 35',
0x24: 'Cool Air',
0x25: 'Warm Air',
0x27: 'Time Dry',
}
_COURSE_CODE_BY_NAME = {name: code for code, name in _COURSE_NAMES.items()}
def _wrinkle_write(p, rep):
if p not in ('On', 'Off'):
return None
return ['washer', 'vs', '0'], {'x.com.samsung.da.wrinklePrevent': p}
def _course_write(p, rep):
"""Encode a human course name to the Samsung hex-encoded course string."""
code = _COURSE_CODE_BY_NAME.get(p)
if code is None:
return None
return ['st', 'dryercourse', 'vs', '0'], {
'x.com.samsung.da.st.dryerMode': f'Course_{code:02X}',
}
# ---------------------------------------------------------------------------
# Capabilities
# ---------------------------------------------------------------------------
DRYER_SETTINGS = Capability(
href='/washer/vs/0',
poll_tier='warm',
entities=(
SensorDesc(key='dry_level', field='x.com.samsung.da.dryLevel',
name='Dry level', icon='mdi:water-percent'),
SensorDesc(key='dry_time', field='x.com.samsung.da.dryTime',
name='Dry time', icon='mdi:timer'),
SensorDesc(key='dryer_type', field='x.com.samsung.da.dryerType',
name='Dryer type', icon='mdi:tumble-dryer'),
SwitchDesc(key='wrinkle_prevent', field='x.com.samsung.da.wrinklePrevent',
name='Wrinkle prevent', icon='mdi:iron',
value_fn=lambda v: v == 'On',
write_fn=_wrinkle_write),
),
)
DRYER_COURSE = Capability(
href='/st/dryercourse/vs/0',
poll_tier='warm',
entities=(
SelectDesc(key='dryer_mode', field='x.com.samsung.da.st.dryerMode',
name='Dryer mode', icon='mdi:tumble-dryer',
options=(), write_fn=_course_write),
),
)
JOB_BEGINNING_STATUS = Capability(
href='/wm/jobbeginingstatus/vs/0',
poll_tier='warm',
entities=(
SensorDesc(key='job_beginning_status',
field='x.com.samsung.da.jobBeginingStatus',
name='Job beginning status',
entity_category='diagnostic'),
),
)
DRYER_DIAGNOSIS = Capability(
href='/diagnosis/vs/0',
poll_tier='warm',
entities=(
SensorDesc(key='diagnosis', field='x.com.samsung.da.diagnosisStart',
name='Diagnosis', entity_category='diagnostic'),
),
)
@@ -340,3 +340,25 @@ FIRMWARE_UPDATE = Capability(
),
),
)
# ---------------------------------------------------------------------------
# Generic door pattern capability (href=None — use as pattern_cap only)
# ---------------------------------------------------------------------------
def _door_key(href: str) -> str:
# /door/cooler/0 -> door_cooler_open ; /door/wine/0 -> door_wine_open
segs = [s for s in href.strip('/').split('/') if s and not s.isdigit()]
return '_'.join(segs) + '_open'
DOOR_GENERIC = Capability(
href=None,
rt_filter='oic.r.door',
poll_tier='hot',
key_fn=_door_key,
entities=(
BinarySensorDesc(key='door_open', field='openState',
name=None, device_class='door',
value_fn=lambda v: v == 'Open'),
),
)
@@ -330,6 +330,17 @@ OVEN_DOOR = Capability(
),
)
OVEN_CONNECTED = Capability(
href='/connected/vs/0',
poll_tier='warm',
entities=(
BinarySensorDesc(key='cloud_connected', field='x.com.samsung.da.connected',
name='Cloud connected', device_class='connectivity',
entity_category='diagnostic',
value_fn=lambda v: v == 'On'),
),
)
OVEN_MODE = Capability(
href='/mode/vs/0',
poll_tier='warm',
+97
View File
@@ -0,0 +1,97 @@
"""Tests for device-type registries and DOOR_GENERIC pattern capability."""
import json
from pathlib import Path
import pytest
from samsung_appliance.registry.adapter import _key
from samsung_appliance.registry.by_type import dishwasher, refrigerator
from samsung_appliance.registry.capabilities import fridge
from samsung_appliance.registry.discovery import discover
GOLDEN = Path(__file__).parent / 'fixtures' / 'golden'
DUMPS = Path(__file__).resolve().parent.parent / 'local-tools' / 'dumps'
def _load_resources(ip: str) -> dict[str, dict]:
data = json.loads((DUMPS / f'{ip}.json').read_text())
return {k: v for k, v in data['resources'].items() if isinstance(v, dict)}
# ---------------------------------------------------------------------------
# Integration: dishwasher and refrigerator registries
# ---------------------------------------------------------------------------
def test_dishwasher_registry_discovers_expected_keys():
"""Discover over dishwasher dump and verify golden state_keys are a subset."""
resources = _load_resources('10.0.0.129')
reg = dishwasher.REGISTRY
bound = discover(resources, reg.capabilities, reg.pattern_capabilities)
actual_keys = {_key(b) for b in bound}
golden = json.loads((GOLDEN / 'dishwasher.json').read_text())
golden_keys = set(golden['state_keys'])
missing = golden_keys - actual_keys
assert not missing, (
f"Golden keys not found in dishwasher discover output:\n {sorted(missing)}"
)
def test_refrigerator_registry_discovers_expected_keys():
"""Discover over refrigerator dump and verify golden state_keys are a subset."""
resources = _load_resources('10.0.0.254')
reg = refrigerator.REGISTRY
bound = discover(resources, reg.capabilities, reg.pattern_capabilities)
actual_keys = {_key(b) for b in bound}
golden = json.loads((GOLDEN / 'refrigerator.json').read_text())
golden_keys = set(golden['state_keys'])
missing = golden_keys - actual_keys
assert not missing, (
f"Golden keys not found in refrigerator discover output:\n {sorted(missing)}"
)
# ---------------------------------------------------------------------------
# DOOR_GENERIC pattern capability unit tests
# ---------------------------------------------------------------------------
def test_door_generic_binds_unknown_door_href():
"""/door/wine/0 with oic.r.door rt binds via DOOR_GENERIC with key door_wine_open."""
resources = {
'/door/wine/0': {'rt': ['oic.r.door'], 'openState': 'Open'},
}
bound = discover(resources, {}, [fridge.DOOR_GENERIC])
assert len(bound) == 1
b = bound[0]
assert _key(b) == 'door_wine_open', f"Expected 'door_wine_open', got {_key(b)!r}"
def test_door_generic_does_not_bind_doors_aggregate():
"""/doors/vs/0 with Samsung-proprietary rt is not bound by DOOR_GENERIC."""
resources = {
'/doors/vs/0': {
'rt': ['x.com.samsung.da.doors'],
'x.com.samsung.da.items': [],
},
}
bound = discover(resources, {}, [fridge.DOOR_GENERIC])
assert bound == [], f"Expected no bindings, got {bound}"
def test_door_generic_does_not_bind_known_door():
"""/door/cooler/0 is claimed by DOOR_FRIDGE exact cap; DOOR_GENERIC must not also bind it."""
resources = {
'/door/cooler/0': {'rt': ['oic.r.door'], 'openState': 'Closed'},
}
reg = refrigerator.REGISTRY
bound = discover(resources, reg.capabilities, [fridge.DOOR_GENERIC])
# Should be bound only by DOOR_FRIDGE (exact cap), not DOOR_GENERIC
cooler_bindings = [b for b in bound if b.href == '/door/cooler/0']
assert len(cooler_bindings) > 0, "Expected at least one binding for /door/cooler/0"
for b in cooler_bindings:
assert b.capability is fridge.DOOR_FRIDGE, (
f"Expected capability DOOR_FRIDGE, got {b.capability!r}"
)
# key_override should be None when bound via exact cap
assert b.key_override is None, (
f"Expected no key_override (exact cap binding), got {b.key_override!r}"
)