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

Merge pull request #70 from vmvarga/feat/wd7000b-fridge-fixes

Browse Source 
feat: laundry firmware-flag gating, fridge vendor temp writes, course…
This commit is contained in:
Marc Billow
2026-07-26 21:25:26 -05:00
committed by GitHub
parent aff7647460 2fd8ff8260
commit 9b09f1aee3
9 changed files with 347 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/coordinator.py
+18 -4
View File
@@ -22,7 +22,11 @@ from smartthings_local.ocf.state_cache import StateCache
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device, for_device_by_model, for_device_by_resources
from .registry.capabilities.common import merge_options_field, remote_control_enabled
from .registry.capabilities.common import (
merge_options_field,
remote_control_enabled,
remote_control_required_for_write,
)
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -531,7 +535,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
opted this device out of it via CONF_BYPASS_REMOTE_CONTROL (issue
#54: some devices accept certain writes, e.g. a washer's default
dosing levels, even while reporting remote control off, so the
block's assumption doesn't hold for every model)."""
block's assumption doesn't hold for every model), or the laundry
firmware flag isModelSettingWithoutSC declares settings writable
without Smart Control (cycle start/pause/stop on /operational/state
still require it)."""
desc = bound_entity.desc
write_fn = getattr(desc, 'write_fn', None)
if write_fn is None:
@@ -540,7 +547,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
rep = self._cache.get(href or '') or {}
resources = self._cache.snapshot()
bypass_remote_control = self._entry.options.get(CONF_BYPASS_REMOTE_CONTROL, False)
if not bypass_remote_control and not remote_control_enabled(resources):
if (
not bypass_remote_control
and remote_control_required_for_write(resources, href or '')
and not remote_control_enabled(resources)
):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="remote_control_disabled",
@@ -553,7 +564,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
translation_domain=DOMAIN,
translation_key=error,
)
result = write_fn(payload, rep, href)
try:
result = write_fn(payload, rep, href, resources)
except TypeError:
result = write_fn(payload, rep, href)
if result is None:
self._log.warning("write_fn rejected payload %r for %s", payload, href)
return
custom_components/localthings/registry/by_type/refrigerator.py
+1 -1
View File
@@ -33,7 +33,7 @@ REGISTRY = DeviceRegistry(
]),
pattern_capabilities=[
fridge.TEMP_CURRENT_GENERIC,
fridge.TEMP_SETPOINT_GENERIC,
fridge.TEMP_SETPOINT,
fridge.ICEMAKER_GENERIC,
fridge.DOOR_GENERIC,
],
custom_components/localthings/registry/capabilities/common.py
+85 -1
View File
@@ -59,9 +59,21 @@ def _ml_to_l(v):
def _active_alarm_codes(items):
"""Join active alarm codes; skip retained rows Samsung leaves as Deleted.
Laundry boards keep a Deleted ErrorCode row in /alarms/vs/0 after the
condition clears (see WD7000B diagnostics). Surface only live alarms so
HA doesn't stick on a stale ErrorCode. Range-hood has its own stricter
helper that also drops ErrorCode_OFF.
"""
if not items or not isinstance(items, list):
return 'none'
codes = [i.get('x.com.samsung.da.code') for i in items if i.get('x.com.samsung.da.code')]
codes = [
i.get('x.com.samsung.da.code')
for i in items
if i.get('x.com.samsung.da.code')
and str(i.get('x.com.samsung.da.state', '')).lower() != 'deleted'
]
return ', '.join(codes) if codes else 'none'
@@ -92,6 +104,52 @@ def merge_options_field(cached, new_tokens):
return merged
# /wm/setinfo/vs/0 -- laundry-family firmware capability flags. Present on
# washers, dryers, and dishwashers; absent on fridge/oven/AC. Static for the
# life of a given board, so reading them from the /device/0 seed (no dedicated
# poll_tier) is enough.
_SETINFO_HREF = '/wm/setinfo/vs/0'
_POWER_ON_OFF_FIELD = 'x.com.samsung.da.isModelSettingPowerOnOff'
_WITHOUT_SC_FIELD = 'x.com.samsung.da.isModelSettingWithoutSC'
def model_allows_power_on_off(resources: dict) -> bool:
"""True unless firmware explicitly declares remote power on/off unsupported.
`/wm/setinfo/vs/0`.`isModelSettingPowerOnOff` is `"false"` on many laundry
boards (washers/dryers): `/power/0` and `/power/vs/0` still report state,
but CoAP writes are ignored. Absent setinfo (non-laundry families) keeps
the writable switch -- current behavior.
"""
setinfo = resources.get(_SETINFO_HREF)
if setinfo is None:
return True
flag = setinfo.get(_POWER_ON_OFF_FIELD)
if flag is None:
return True
return str(flag).lower() != 'false'
def model_setting_without_sc(resources: dict) -> bool:
"""True when firmware declares settings writable without Smart Control.
`/wm/setinfo/vs/0`.`isModelSettingWithoutSC` is `"true"` on washers/dryers
that accept temperature/spin/cycle-option writes while remote control is
off. Cycle start/pause/stop still need Smart Control on those boards --
the flag name is settings-specific, not a blanket remote-control bypass.
"""
setinfo = resources.get(_SETINFO_HREF) or {}
return str(setinfo.get(_WITHOUT_SC_FIELD, '')).lower() == 'true'
def _power_switch_exists(rep, resources):
return model_allows_power_on_off(resources)
def _power_sensor_exists(rep, resources):
return not model_allows_power_on_off(resources)
def sensor_item_value(items, sensor_type, index=0):
"""Pull one reading out of a `/sensors/vs/0`-style items[] list -- each
item is `{type, value: [...]}`; `index` picks which slot of a possibly
@@ -128,10 +186,17 @@ def sensor_item_value(items, sensor_type, index=0):
POWER_GENERIC = Capability(
href='/power/0',
entities=(
# Writable when firmware allows remote power; otherwise a read-only
# binary_sensor with the same key keeps HA state without a dead switch.
SwitchDesc(key='power_switch', field='value',
value_fn=lambda v: bool(v),
exists_fn=_power_switch_exists,
write_fn=lambda p, rep, href=None: (
['power', '0'], {'value': p == 'On'})),
BinarySensorDesc(key='power_switch', field='value',
device_class='power',
value_fn=lambda v: bool(v),
exists_fn=_power_sensor_exists),
),
)
@@ -141,9 +206,14 @@ POWER_VS_FALLBACK = Capability(
entities=(
SwitchDesc(key='power_switch', field='x.com.samsung.da.power',
value_fn=lambda v: v == 'On',
exists_fn=_power_switch_exists,
write_fn=lambda p, rep, href=None: (
['power', 'vs', '0'],
{'x.com.samsung.da.power': 'On' if p == 'On' else 'Off'})),
BinarySensorDesc(key='power_switch', field='x.com.samsung.da.power',
device_class='power',
value_fn=lambda v: v == 'On',
exists_fn=_power_sensor_exists),
),
)
@@ -192,6 +262,20 @@ def remote_control_enabled(resources: dict) -> bool:
return True
def remote_control_required_for_write(resources: dict, bound_href: str) -> bool:
"""Whether a write to bound_href should be gated on Smart Control.
When isModelSettingWithoutSC is true, laundry firmware accepts settings
writes (wash temp, spin, course options, buzzer, ...) with remote
control off, but cycle start/pause/stop on /operational/state still
need Smart Control. Absent that flag, keep the historical blanket gate.
"""
if not model_setting_without_sc(resources):
return True
href = bound_href or ''
return href.startswith('/operational/state')
REMOTE_CONTROL_GENERIC = Capability(
href='/remotectrl/0',
poll_tier='warm',
custom_components/localthings/registry/capabilities/fridge.py
+33 -5
View File
@@ -62,7 +62,38 @@ TEMP_CURRENT_GENERIC = Capability(
),
)
TEMP_SETPOINT_GENERIC = Capability(
def _temp_setpoint_write(p, rep, href=None, resources=None):
"""Write temperature — prefer vendor /temperatures/vs/0 when available,
fall back to direct OCF /temperature/desired/ write otherwise.
Samsung fridges expose both OCF-standard /temperature/desired/* and vendor
/temperatures/vs/0. On some models only the vendor path commits the change;
on others both work. Using the vendor path when present is always correct.
Item IDs follow the Samsung convention: "0" = Freezer, "1" = Fridge/Cooler.
"""
if not href:
return None
if resources and '/temperatures/vs/0' in resources:
if '/cooler/' in href:
item_id = '1'
elif '/freezer/' in href:
item_id = '0'
else:
return None
return (
['temperatures', 'vs', '0'],
{'x.com.samsung.da.items': [
{'x.com.samsung.da.id': item_id,
'x.com.samsung.da.desired': str(int(round(float(p))))}
]}
)
return (
[s for s in href.strip('/').split('/') if s],
{'temperature': int(round(float(p)))}
)
TEMP_SETPOINT = Capability(
href=None,
href_prefix='/temperature/desired/',
strip_prefix_in_key=True,
@@ -73,10 +104,7 @@ TEMP_SETPOINT_GENERIC = Capability(
use_instance_name=True, device_class='temperature', unit_fn=_temp_unit,
native_min=-20.0, native_max=50.0,
range_field='range', entity_category='config',
write_fn=lambda p, rep, href=None: (
[s for s in href.strip('/').split('/') if s],
{'temperature': int(round(float(p)))}
) if href else None),
write_fn=_temp_setpoint_write),
),
)
custom_components/localthings/translations/en.json
+10 -3
View File
@@ -55,6 +55,9 @@
"power_state": {
"name": "Power state"
},
"power_switch": {
"name": "Power"
},
"rapid_freezing": {
"name": "Rapid freezing"
},
@@ -379,16 +382,20 @@
"27": "Rinse+Spin",
"28": "Drain/Spin",
"29": "Drum Clean+",
"2a": "Jeans",
"2b": "AI Wash",
"2d": "Silent Wash",
"2e": "Baby Care",
"2f": "Activewear",
"30": "Cloudy Day",
"32": "Shirts",
"33": "Bedding",
"34": "Mixed",
"36": "Wash+Dry",
"37": "Air Wash",
"38": "Cotton Dry",
"39": "Synthetics Dry",
"3a": "Drum Clean",
"53": "Heavy Duty",
"55": "Activewear",
"57": "Delicate",
@@ -398,11 +405,11 @@
"7c": "Whites",
"7d": "Bedding/Waterproof",
"7e": "Self-Clean",
"7f": "Wool/Delicate",
"86": "Deep Wash",
"87": "Download",
"8f": "Intense Cold",
"96": "Less Microfiber",
"2b": "AI Wash",
"2a": "Jeans"
"96": "Less Microfiber"
}
},
"washer_dry_level": {
custom_components/localthings/translations/nl.json
+10 -3
View File
@@ -55,6 +55,9 @@
"power_state": {
"name": "Voedingsstatus"
},
"power_switch": {
"name": "Voeding"
},
"rapid_freezing": {
"name": "Snelvriezen"
},
@@ -379,16 +382,20 @@
"27": "Spoelen+centrifugeren",
"28": "Afpompen/centrifugeren",
"29": "Drum Clean+",
"2a": "Spijkerbroek",
"2b": "AI Wash",
"2d": "Stille was",
"2e": "Babyverzorging",
"2f": "Sportkleding",
"30": "Bewolkte dag",
"32": "Overhemden",
"33": "Beddengoed",
"34": "Gemengd",
"36": "Wassen+drogen",
"37": "Air Wash",
"38": "Katoen drogen",
"39": "Synthetisch drogen",
"3a": "Trommel reinigen",
"53": "Intensief",
"55": "Sportkleding",
"57": "Fijn",
@@ -398,11 +405,11 @@
"7c": "Witte was",
"7d": "Beddengoed/Waterdicht",
"7e": "Zelfreinigend",
"7f": "Wol/Fijn",
"86": "Diep wassen",
"87": "Downloaden",
"8f": "Intensief koud",
"96": "Minder microvezels",
"2b": "AI Wash",
"2a": "Spijkerbroek"
"96": "Minder microvezels"
}
},
"washer_dry_level": {
tests/localthings/test_coordinator.py
+125
View File
@@ -16,6 +16,7 @@ from custom_components.localthings.const import (
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capabilities.common import (
remote_control_enabled,
remote_control_required_for_write,
)
from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL, PUSH_HEALTH_WINDOW_S
@@ -880,6 +881,38 @@ class TestRemoteControlEnabled:
assert remote_control_enabled({}) is True
class TestRemoteControlRequiredForWrite:
"""isModelSettingWithoutSC auto-exempts settings writes, not cycle control."""
def test_without_flag_always_requires(self):
resources = {}
assert remote_control_required_for_write(resources, '/washer/vs/0') is True
assert remote_control_required_for_write(
resources, '/operational/state/vs/0') is True
def test_without_sc_exempts_settings_keeps_operational(self):
resources = {
'/wm/setinfo/vs/0': {
'x.com.samsung.da.isModelSettingWithoutSC': 'true',
},
}
assert remote_control_required_for_write(resources, '/washer/vs/0') is False
assert remote_control_required_for_write(resources, '/course/vs/0') is False
assert remote_control_required_for_write(resources, '/buzzersound/vs/0') is False
assert remote_control_required_for_write(
resources, '/operational/state/vs/0') is True
assert remote_control_required_for_write(
resources, '/operational/state/0') is True
def test_without_sc_false_still_requires(self):
resources = {
'/wm/setinfo/vs/0': {
'x.com.samsung.da.isModelSettingWithoutSC': 'false',
},
}
assert remote_control_required_for_write(resources, '/washer/vs/0') is True
async def test_send_command_blocked_when_remote_control_disabled(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
@@ -1024,3 +1057,95 @@ async def test_send_command_bypasses_remote_control_when_option_enabled(
await coordinator.async_send_command(bound, 5)
assert coordinator._cache.get('/some/path') == {'value': 5}
async def test_send_command_settings_allowed_when_without_sc(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""Laundry boards with isModelSettingWithoutSC=true accept settings
writes while remote control is off -- no options-flow bypass needed."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import SelectDesc
fake = mock_coordinator_observe_session
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
coordinator._cache.apply_rep(
'/remotectrl/vs/0',
{'x.com.samsung.da.remoteControlEnabled': 'false'},
source='test',
)
coordinator._cache.apply_rep(
'/wm/setinfo/vs/0',
{'x.com.samsung.da.isModelSettingWithoutSC': 'true'},
source='test',
)
def _write_fn(payload, rep, href=None):
return (['washer', 'vs', '0'], {'x.com.samsung.da.spinLevel': payload})
desc = SelectDesc(key='spin_speed', field='x.com.samsung.da.spinLevel',
write_fn=_write_fn)
bound = BoundEntity(
href='/washer/vs/0', capability=coordinator.bound[0].capability, desc=desc,
)
with patch.object(fake, 'subscribe'):
fake.post = lambda *a, **k: (0x44, b'')
await coordinator.async_send_command(bound, '1000')
assert coordinator._cache.get('/washer/vs/0') == {
'x.com.samsung.da.spinLevel': '1000',
}
async def test_send_command_operational_still_blocked_when_without_sc(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""Start/Pause/Stop stay gated on Smart Control even when settings
are allowed without it -- isModelSettingWithoutSC is settings-only."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import ButtonDesc
fake = mock_coordinator_observe_session
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
coordinator._cache.apply_rep(
'/remotectrl/vs/0',
{'x.com.samsung.da.remoteControlEnabled': 'false'},
source='test',
)
coordinator._cache.apply_rep(
'/wm/setinfo/vs/0',
{'x.com.samsung.da.isModelSettingWithoutSC': 'true'},
source='test',
)
def _write_fn(payload, rep, href=None):
return (['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': payload})
desc = ButtonDesc(key='start', payload='Run', write_fn=_write_fn)
bound = BoundEntity(
href='/operational/state/vs/0',
capability=coordinator.bound[0].capability,
desc=desc,
)
posted = False
def _post(*a, **k):
nonlocal posted
posted = True
return (0x44, b'')
with patch.object(fake, 'subscribe'):
fake.post = _post
with pytest.raises(ServiceValidationError) as exc_info:
await coordinator.async_send_command(bound, 'Run')
assert exc_info.value.translation_domain == DOMAIN
assert exc_info.value.translation_key == 'remote_control_disabled'
assert posted is False
tests/test_common_capabilities.py
+63 -2
View File
@@ -1,5 +1,7 @@
from custom_components.localthings.registry.capabilities import common
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.entities import BinarySensorDesc, SwitchDesc
from tests.conftest import _load_device
def _reg():
@@ -65,7 +67,7 @@ class TestMergeOptionsField:
class TestPowerFallback:
def test_generic_href_read_write(self):
assert common.POWER_GENERIC.href == '/power/0'
desc = common.POWER_GENERIC.entities[0]
desc = next(e for e in common.POWER_GENERIC.entities if isinstance(e, SwitchDesc))
assert desc.value_fn(True) is True
assert desc.value_fn(False) is False
path, body = desc.write_fn('On', {})
@@ -80,13 +82,72 @@ class TestPowerFallback:
{}, {'/power/0': {}, '/power/vs/0': {}}) is False
def test_vs_fallback_read_write(self):
desc = common.POWER_VS_FALLBACK.entities[0]
desc = next(e for e in common.POWER_VS_FALLBACK.entities if isinstance(e, SwitchDesc))
assert desc.value_fn('On') is True
assert desc.value_fn('Off') is False
path, body = desc.write_fn('On', {})
assert path == ['power', 'vs', '0']
assert body == {'x.com.samsung.da.power': 'On'}
def test_power_switch_hidden_when_firmware_disallows(self):
switch = next(e for e in common.POWER_GENERIC.entities if isinstance(e, SwitchDesc))
sensor = next(e for e in common.POWER_GENERIC.entities if isinstance(e, BinarySensorDesc))
resources = {
'/power/0': {'value': True},
'/wm/setinfo/vs/0': {'x.com.samsung.da.isModelSettingPowerOnOff': 'false'},
}
assert switch.exists_fn(resources['/power/0'], resources) is False
assert sensor.exists_fn(resources['/power/0'], resources) is True
def test_power_switch_writable_when_firmware_allows(self):
switch = next(e for e in common.POWER_GENERIC.entities if isinstance(e, SwitchDesc))
sensor = next(e for e in common.POWER_GENERIC.entities if isinstance(e, BinarySensorDesc))
resources = {
'/power/0': {'value': True},
'/wm/setinfo/vs/0': {'x.com.samsung.da.isModelSettingPowerOnOff': 'true'},
}
assert switch.exists_fn(resources['/power/0'], resources) is True
assert sensor.exists_fn(resources['/power/0'], resources) is False
def test_power_switch_default_writable_without_setinfo(self):
switch = next(e for e in common.POWER_GENERIC.entities if isinstance(e, SwitchDesc))
sensor = next(e for e in common.POWER_GENERIC.entities if isinstance(e, BinarySensorDesc))
resources = {'/power/0': {'value': True}}
assert switch.exists_fn(resources['/power/0'], resources) is True
assert sensor.exists_fn(resources['/power/0'], resources) is False
class TestWmSetinfoFlags:
def test_washer_fixture_carries_setinfo_from_device0(self):
"""/wm/setinfo/vs/0 lands in the seed snapshot -- no dedicated capability."""
resources = _load_device('washer')
assert '/wm/setinfo/vs/0' in resources
assert common.model_allows_power_on_off(resources) is False
assert common.model_setting_without_sc(resources) is True
def test_dishwasher_allows_power_on_off(self):
resources = _load_device('dishwasher')
assert common.model_allows_power_on_off(resources) is True
assert common.model_setting_without_sc(resources) is False
def test_deleted_alarms_filtered(self):
assert common._active_alarm_codes([
{
'x.com.samsung.da.code': 'ErrorCode',
'x.com.samsung.da.state': 'Deleted',
},
]) == 'none'
assert common._active_alarm_codes([
{
'x.com.samsung.da.code': 'LE',
'x.com.samsung.da.state': 'Triggered',
},
{
'x.com.samsung.da.code': 'ErrorCode',
'x.com.samsung.da.state': 'Deleted',
},
]) == 'LE'
class TestKidsLockFallback:
def test_generic_read_write(self):
tests/test_fridge_capabilities.py
+2 -2
View File
@@ -20,9 +20,9 @@ class TestTempCurrentGeneric:
assert desc.unit_fn({'temperature': 5.0}) == '°F'
class TestTempSetpointGeneric:
class TestTempSetpoint:
def test_unit_reads_celsius(self):
desc = fridge.TEMP_SETPOINT_GENERIC.entities[0]
desc = fridge.TEMP_SETPOINT.entities[0]
assert desc.unit_fn({'temperature': -19.0, 'units': 'C'}) == '°C'