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

feat: reject bubble soak/pre-wash/intensive writes on unsupported courses

Browse Source 
Add a validate_fn hook to SwitchDesc, checked in switch.py before dispatch
and surfaced as a ServiceValidationError so an unsupported write shows a
real error in the UI instead of the coordinator's silent log-only rejection.

Wired it into the three course-gated washer switches using their
availability bitmaps (BubbleSoakSet/PreWashAvailableSet/IntensiveAvailableSet),
which line up positionally with editCourseList. Turning a toggle off is
never blocked, and the check fails open whenever the course or bitmap can't
be resolved.

Also fixes a bug in _bool_option_write: it took a `p and 'On' or 'Off'`-style
truthy check, but switch.py always calls it with the string 'On' or 'Off' --
both truthy, so every write landed as 'On' regardless of intent.
This commit is contained in:
Marc Billow
2026-07-22 19:45:57 +00:00
parent e49b61e01e
commit f54daf3608
4 changed files with 155 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/registry/capabilities/washer.py
+57 -17
View File
@@ -17,7 +17,7 @@ from datetime import datetime, timezone
from ..capability import Capability
from ..entities import BinarySensorDesc, SelectDesc, SensorDesc, SwitchDesc
from .laundry import cycle_select, hex_pairs, option_value, replace_in_options
from .laundry import cycle_options, cycle_select, hex_pairs, option_value, replace_in_options
# ---------------------------------------------------------------------------
# Course_XX hex codes. 23 of the codes named in strings.json/translations
@@ -245,25 +245,27 @@ def _dosing_alarm_exists(prefix):
# already used by AiOption and KidsLockBypass in this same array, so
# PreWashSetting/IntensiveSetting are assumed to follow suit.
#
# Each also has a same-shaped '<Prefix>Set'/'<Prefix>AvailableSet' hex-pair
# string that lines up positionally with editCourseList -- e.g. on the combo
# dump, BubbleSoakSet's byte for the active course (Course_1C, position 0)
# is '00' while PreWashAvailableSet/IntensiveAvailableSet are both 'F0',
# suggesting F0/00 is an available/unavailable flag per course. That's not
# exposed here: exists_fn only runs once, against the setup-time snapshot
# (see entity._is_included), so gating on the *current* course would just
# make the entity's existence depend on whatever course happened to be
# selected when Home Assistant started, not on the device's real, static
# capability. Toggling one of these on a course the app would gray it out
# for is untested; treat it like any other write this integration doesn't
# validate against device-side state.
# Each also has a differently-named hex-pair availability field that lines up
# positionally with editCourseList: BubbleSoakSet, PreWashAvailableSet,
# IntensiveAvailableSet. On the reporter's dump (course '30' at position 1 of
# 24), all three read 'F0' at that position and the toggle was writable --
# and the same dump's earlier state (course '1C' at position 0, 'BubbleSoak
# Off') decodes to '00' for that course, matching the app graying the
# control out there. 'F0'/'00' is treated as available/unavailable on that
# evidence. exists_fn (device-level presence) still only runs once, against
# the setup-time snapshot, so it isn't a fit for this per-course check --
# validate_fn runs on every write attempt instead, rejecting an on-write for
# a course whose byte isn't 'F0' with a user-facing error rather than
# silently no-opping against the device.
def _bool_option_write(prefix):
def write(p, rep, href=None):
if p not in ('On', 'Off'):
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, prefix, 'On' if p else 'Off'),
'x.com.samsung.da.options': replace_in_options(opts, prefix, p),
}
return write
@@ -277,6 +279,41 @@ def _bool_option_exists(prefix):
rep.get('x.com.samsung.da.options'), prefix) is not None
_AVAILABILITY_FIELD = {
'BubbleSoak': 'BubbleSoakSet',
'PreWashSetting': 'PreWashAvailableSet',
'IntensiveSetting': 'IntensiveAvailableSet',
}
def _bool_option_validate(prefix, human_name):
"""Reject turning `prefix` on when the selected course's byte in its
availability bitmap isn't 'F0'. Turning off is never blocked. Falls back
to allowing the write whenever the availability data can't be resolved
(unrecognized course, missing/mismatched-length bitmap) rather than
guessing -- a false rejection is worse than an occasional no-op write."""
availability_field = _AVAILABILITY_FIELD[prefix]
def validate(p, rep, resources):
if p != 'On':
return None
opts = rep.get('x.com.samsung.da.options') or []
current = option_value(opts, 'Course')
courses = cycle_options(resources)
if not current or current not in courses:
return None
raw = option_value(opts, availability_field)
if raw is None:
return None
pairs = hex_pairs(raw)
if len(pairs) != len(courses):
return None
if pairs[courses.index(current)] != 'F0':
return f"{human_name} isn't available on the selected cycle."
return None
return validate
WASHER_COURSE = Capability(
href='/course/vs/0',
entities=(
@@ -337,16 +374,19 @@ WASHER_COURSE = Capability(
entity_category='config',
exists_fn=_bool_option_exists('BubbleSoak'),
rep_fn=_bool_option_value('BubbleSoak'),
write_fn=_bool_option_write('BubbleSoak')),
write_fn=_bool_option_write('BubbleSoak'),
validate_fn=_bool_option_validate('BubbleSoak', 'Bubble soak')),
SwitchDesc(key='pre_wash', name='Pre wash', icon='mdi:washing-machine',
entity_category='config',
exists_fn=_bool_option_exists('PreWashSetting'),
rep_fn=_bool_option_value('PreWashSetting'),
write_fn=_bool_option_write('PreWashSetting')),
write_fn=_bool_option_write('PreWashSetting'),
validate_fn=_bool_option_validate('PreWashSetting', 'Pre wash')),
SwitchDesc(key='intensive', name='Intensive', icon='mdi:washing-machine',
entity_category='config',
exists_fn=_bool_option_exists('IntensiveSetting'),
rep_fn=_bool_option_value('IntensiveSetting'),
write_fn=_bool_option_write('IntensiveSetting')),
write_fn=_bool_option_write('IntensiveSetting'),
validate_fn=_bool_option_validate('IntensiveSetting', 'Intensive')),
),
)
custom_components/localthings/registry/entities.py
+9 -1
View File
@@ -2,7 +2,9 @@
Frozen dataclasses so the future native HA component can consume them as
EntityDescription subclasses unchanged. Read transforms live in value_fn;
presence gating in exists_fn; write logic in write_fn on command platforms.
presence gating in exists_fn; write logic in write_fn on command platforms;
pre-write rejection (surfaced to the user, not just logged) in validate_fn
where a description declares one.
"""
from __future__ import annotations
@@ -10,6 +12,11 @@ from dataclasses import dataclass, field
from typing import Any, Callable, Optional
WriteFn = Optional[Callable[[Any, dict], "tuple[list[str], dict] | None"]]
# (payload, rep, resources) -> a human-readable rejection message, or None to
# allow the write. resources is the coordinator's full href->rep snapshot, for
# the same cross-resource lookups exists_fn needs (e.g. reading a sibling
# href's live option list).
ValidateFn = Optional[Callable[[Any, dict, dict], "str | None"]]
def _identity(v: Any) -> Any:
@@ -61,6 +68,7 @@ class SelectDesc(SamsungEntityDescription):
class SwitchDesc(SamsungEntityDescription):
device_class: Optional[str] = None
write_fn: WriteFn = None
validate_fn: ValidateFn = None
@dataclass(frozen=True, kw_only=True)
custom_components/localthings/switch.py
+12 -2
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .registry.entities import SwitchDesc
@@ -38,7 +39,16 @@ class LocalThingsSwitch(LocalThingsEntity, SwitchEntity):
return (self.coordinator.data or {}).get(self._state_key)
async def async_turn_on(self, **kwargs) -> None:
await self.coordinator.async_send_command(self._bound, 'On')
await self._send('On')
async def async_turn_off(self, **kwargs) -> None:
await self.coordinator.async_send_command(self._bound, 'Off')
await self._send('Off')
async def _send(self, payload: str) -> None:
desc: SwitchDesc = self._bound.desc
if desc.validate_fn is not None:
rep = self.coordinator.last_resources.get(self._bound.href) or {}
error = desc.validate_fn(payload, rep, self.coordinator.last_resources)
if error:
raise ServiceValidationError(error)
await self.coordinator.async_send_command(self._bound, payload)
tests/test_washer_capabilities.py
+77 -2
View File
@@ -291,17 +291,92 @@ class TestWashOptionToggles:
assert self._desc(key).rep_fn(rep) is True
def test_write_on_and_off(self):
"""write_fn receives the same 'On'/'Off' string switch.py sends
(not a bool) -- covers a bug where an earlier `'On' if p else 'Off'`
implementation always wrote 'On', since any non-empty string
(including 'Off') is truthy."""
for key in self._keys():
prefix = self._prefix(key)
rep = {'x.com.samsung.da.options': [f'{prefix}_Off', 'GMT_02']}
path, body = self._desc(key).write_fn(True, rep)
path, body = self._desc(key).write_fn('On', rep)
assert path == ['course', 'vs', '0']
assert f'{prefix}_On' in body['x.com.samsung.da.options']
assert 'GMT_02' in body['x.com.samsung.da.options']
rep = {'x.com.samsung.da.options': [f'{prefix}_On']}
path, body = self._desc(key).write_fn(False, rep)
path, body = self._desc(key).write_fn('Off', rep)
assert f'{prefix}_Off' in body['x.com.samsung.da.options']
assert f'{prefix}_On' not in body['x.com.samsung.da.options']
def test_write_rejects_non_on_off_payload(self):
for key in self._keys():
prefix = self._prefix(key)
rep = {'x.com.samsung.da.options': [f'{prefix}_Off']}
assert self._desc(key).write_fn('bogus', rep) is None
# editCourseList and availability bitmaps from the reporter's issue #22
# follow-up dump (WD90T654DBN/S1, course '30' selected, Bubble Soak just
# turned on in the app): 24 courses, course '30' at position 1 reads 'F0'
# (available) on all three bitmaps; course '1C' at position 0 reads '00' on
# BubbleSoakSet (matching the app graying that control out for Eco 40-60).
_EDIT_COURSE_RESOURCES = {
'/wm/editcourse/vs/0': {
'x.com.samsung.da.editCourseList':
'EditCourseList_1C301E26361B1D1F253324322022232F212D272838393729',
},
}
_BUBBLE_SOAK_SET = 'BubbleSoakSet_00F000F000F000F0F0F0F00000F000F0F00000F000000000'
_PRE_WASH_AVAILABLE_SET = 'PreWashAvailableSet_F0F000F0F0F000F0F0F0F00000F0F0F0F00000F000000000'
_INTENSIVE_AVAILABLE_SET = 'IntensiveAvailableSet_F0F000F0F0F000F0F0F0F00000F0F0F0F00000F000000000'
class TestWashOptionToggleValidation:
"""validate_fn rejects turning a toggle on for a course whose byte in
its availability bitmap isn't 'F0', with a user-facing message switch.py
raises as ServiceValidationError -- distinct from write_fn's silent
no-op for a malformed payload."""
@staticmethod
def _desc(key):
return next(e for e in washer.WASHER_COURSE.entities if e.key == key)
def test_allowed_on_a_supported_course(self):
rep = {'x.com.samsung.da.options': ['Course_30', _BUBBLE_SOAK_SET]}
assert self._desc('bubble_soak').validate_fn(
'On', rep, _EDIT_COURSE_RESOURCES) is None
def test_rejected_on_an_unsupported_course(self):
rep = {'x.com.samsung.da.options': ['Course_1C', _BUBBLE_SOAK_SET]}
msg = self._desc('bubble_soak').validate_fn('On', rep, _EDIT_COURSE_RESOURCES)
assert msg == "Bubble soak isn't available on the selected cycle."
def test_pre_wash_and_intensive_use_their_own_availableset_field(self):
rep = {'x.com.samsung.da.options': ['Course_30', _PRE_WASH_AVAILABLE_SET]}
assert self._desc('pre_wash').validate_fn(
'On', rep, _EDIT_COURSE_RESOURCES) is None
rep = {'x.com.samsung.da.options': ['Course_30', _INTENSIVE_AVAILABLE_SET]}
assert self._desc('intensive').validate_fn(
'On', rep, _EDIT_COURSE_RESOURCES) is None
def test_turning_off_is_never_blocked(self):
rep = {'x.com.samsung.da.options': ['Course_1C', _BUBBLE_SOAK_SET]}
assert self._desc('bubble_soak').validate_fn(
'Off', rep, _EDIT_COURSE_RESOURCES) is None
def test_allows_write_when_course_unresolvable(self):
"""No editCourseList, no Course_ token, or a bitmap whose length
doesn't match editCourseList -- in every case, fail open rather than
block a write we can't actually verify."""
desc = self._desc('bubble_soak')
rep = {'x.com.samsung.da.options': ['Course_1C', _BUBBLE_SOAK_SET]}
assert desc.validate_fn('On', rep, {}) is None
rep = {'x.com.samsung.da.options': [_BUBBLE_SOAK_SET]}
assert desc.validate_fn('On', rep, _EDIT_COURSE_RESOURCES) is None
rep = {'x.com.samsung.da.options': ['Course_1C', 'BubbleSoakSet_00F0']}
assert desc.validate_fn('On', rep, _EDIT_COURSE_RESOURCES) is None
class TestFlexWashAndComboFixturesHaveCompleteCoverage: