Compare commits

...
Author SHA1 Message Date
Marc Billow ea986fe8e5 refactor: adopt cleaner write pattern for options arrays 2026-07-24 15:05:26 -05:00
Marc Billow e47a71689d Merge pull request #71 from splitsec2/fix/switch-off-inversion
fix(fridge): switch turn-off silently sends 'On' (truthy-string guard)
2026-07-24 13:50:53 -05:00
Rob Martin 9cf1e46bf6 fix(fridge): stop switch turn-off silently sending 'On'
Six fridge SwitchDesc write_fns built their payload with
`'On' if p else 'Off'`. The switch platform passes the literal
string 'Off' on turn-off, which is truthy, so the guard always
produced 'On' -- turning these switches off silently re-sent On
and they could never be turned off:

  - ICEMAKER_NIGHTTIME (ice.night.status)
  - STATUS_LOCK helper (devicecontrol + device.sound)
  - DEFROST_DELAY (delayDefrost)
  - WELCOME_LIGHTING (status)
  - CABINET_LIGHT dim (light.dimming.status)
  - ICEMAKER_STATUS_FALLBACK (iceMaker)

Compare `p == 'On'` instead, matching the pattern the other
capability files already use. Adds a regression test asserting
every affected write_fn sends 'Off' on 'Off' and 'On' on 'On'.
2026-07-24 12:11:53 -06:00
Marc Billow 7276ff65a8 Merge pull request #67 from mbillow/claude/arbitrary-href-writes-debug-5wjtm3
feat: add debug options panel for arbitrary resource-href writes (#54)
2026-07-24 10:17:35 -05:00
Marc Billow c731ecefe5 feat: add debug options panel for arbitrary resource-href writes (#54)
Power users can now pick a resource href from a live dropdown, view its
current value, and POST a minimal patch straight to the device -- to pin
down device-specific write behavior without waiting on a new release.
Bypasses the remote-control block and all write_fn/validate_fn logic by
design; the existing remote-control settings toggle moves behind the same
options-flow menu.
2026-07-24 14:02:15 +00:00
Marc Billow 77cc875608 Merge pull request #64 from mbillow/claude/course-list-supportedoptions-fallback
feat: derive washer/dryer/dishwasher cycle list from supportedOptions when editCourseList is empty
2026-07-24 01:18:02 -05:00
Marc Billow ae711038ca Merge pull request #63 from mbillow/claude/issue-56-followup
Drop unexplained Blooming_* diagnostic, confirm remaining air purifier fields (issue #56)
2026-07-24 01:17:33 -05:00
Marc Billow 3b3dd54372 Simplify table-scoped translation key; fix stale resolution; correct docstring
Simplification (feedback: this was overcomplicated): drop the
validated_table gate entirely. cycle_select's table_href now just builds
the translation key directly from whatever course table the device
reports (washer_cycle + Table_02 -> washer_cycle_table_02) instead of
comparing against a hardcoded known-good value and falling back to no key
on any mismatch. A table we haven't shipped translations for yet (e.g.
FlexWash's Table_00) still gets a key built for it -- Home Assistant's own
missing-translation handling takes it from there, the same graceful
fallback already relied on for any individual untranslated code within an
existing table. Adding a newly-confirmed table later is just new
strings.json entries, no code change.

Independent (Opus) review of the prior version caught two real issues,
fixed here regardless of the simplification above:

- translation_key was resolved once at entity construction from whatever
  coordinator.last_resources held at that moment. Discovery can run while
  a sibling resource is still an empty stub (documented precedent: see
  _is_included), so a callable translation_key could permanently bake in
  a stale value for the entity's lifetime. Moved resolution into a
  translation_key property override (Entity.translation_key is a property
  upstream, not a plain attribute), re-evaluated against live coordinator
  data on every access, matching how options/current_option already work.

- The supportedOptions fallback's "smallest passing K wins" docstring
  claimed every larger passing K is an exact multiple of the true one.
  False: the shipped dishwasher fixture has passing K=7 (true) alongside
  10, 14, and 35, none of which are multiples of 7 -- position 0 always
  lands on the same real course code regardless of K, which alone
  satisfies the current-course guard for several unrelated splits.
  Corrected the reasoning to what's actually true (an empirically-matched
  heuristic across six real dumps, not a proof) and added a regression
  test locking in the real dishwasher case so this isn't silently lost.
2026-07-24 05:55:04 +00:00
Marc Billow b2c32a90b6 Scope washer/dryer cycle translations to the device's own course table
Course codes on the shared /course/vs/0 contract aren't guaranteed
consistent across board generations: washer/combo devices report course
table Table_02, dryer devices Table_03 (x.com.samsung.da.st.courseTable,
previously fully ignored), and every code in washer_cycle/dryer_cycle was
confirmed exclusively against those. FlexWash's older DA_WM_A51 board
reports Table_00 instead -- applying the same translations there risked
showing a wrong name for any code that happens to numerically collide
between tables, not just an untranslated one.

SelectDesc.translation_key can now be a callable (resources -> key or
None), mirroring the existing pattern for `options`. laundry.cycle_select
gains optional table_href/validated_table params: when given, the renamed
washer_cycle_table_02/dryer_cycle_table_03 keys only apply when the
device's own course table matches exactly -- a different table, or no
table id at all, gets no translation_key (raw code display) rather than
a guess. dishwasher's call site is unchanged (static key, unconditional):
no equivalent table-id resource exists in any dump seen, and no evidence
its course codes vary by table the way washer/dryer's do.

entity.py and select.py resolve a callable translation_key once (via
coordinator.last_resources) and reuse that resolved value everywhere
_display() needs it, rather than re-checking the raw descriptor field.
2026-07-24 05:42:12 +00:00
Marc Billow b2c0517115 feat: derive washer/dryer/dishwasher cycle list from supportedOptions when editCourseList is empty
Some DA_WM_TP1/TP2-class boards populate /wm/editcourse/vs/0 without ever
filling in editCourseList itself (issue #1), so the Cycle select never gets
created even though the device clearly has one (confirmed via SmartThings
app screenshots and a currently-selected course).

/course/vs/0's own x.com.samsung.da.supportedOptions turns out to already
carry the course list, just undocumented: a 1-hex-nibble header followed by
one fixed-width record per course, self-indexed by a course-code first byte
rather than positional like editCourseList. Confirmed against six
independent real-world dumps pulled from open and closed GitHub issues.

cycle_options() now falls back to deriving this when editCourseList is
empty, gated on two checks: the derived codes must all be distinct, and
must include whatever course is currently selected. Larger multiples of
the true record width trivially re-pass both checks too (they're just a
sparser sampling of the same table), so the smallest passing width wins
rather than requiring one unambiguous match.

Also fires on the washer_flexwash fixture, newly creating a Cycle select
there -- unconfirmed against any ground truth for that device (a different,
older board generation with no editCourseList and no screenshots to check
against), flagged for follow-up discussion rather than silently accepted.
2026-07-24 05:06:39 +00:00
Marc Billow 39dcf6a7b8 Drop unexplained Blooming_* diagnostic, confirm remaining air purifier fields (issue #56)
Per the five running-state diagnostics dumps (Auto/Sleep/Low/Medium/High)
gathered in the issue thread:
- Blooming_* has no corresponding SmartThings app setting, so it's dropped
  entirely rather than kept as an unexplained diagnostic.
- Comode_* reads 'Off' on all five, ruling out the original guess that it
  was the fan-speed selector -- still exposed read-only, purpose unconfirmed.
- OptionCode_60282 and the missing humidity sensor are confirmed correct as
  already modeled.
- /airflow's speed doesn't map monotonically to the five settings and the
  dumps were all captured within one ~30s poll cycle of each other, so it
  stays read-only pending a cleaner, time-spaced capture.

FilterProgress is untouched here: an earlier pass on this issue read the
thread as confirming 100 means "fresh" and renamed the sensor to filter_life
to match, but that reading was backwards -- the reporter clarified 100
means fully used and needs replacing, which is what filter_progress (the
already-shipped name) already implies. That rename was caught before
merging and is not part of this change.
2026-07-24 04:20:42 +00:00
Marc Billow ed768d7257 Merge pull request #62 from mbillow/claude/issue-52-quick-fix
Recognize A-CAWW-TP2-20-COMMON system air conditioners (issue #52)
2026-07-23 23:08:32 -05:00
Marc Billow efe87381b7 Bump version to 0.11.1 2026-07-24 04:06:31 +00:00
Marc Billow 8e25e6dc71 Recognize A-CAWW-TP2-20-COMMON system air conditioners (issue #52)
New '-CAWW-' modelNum token for multi-indoor-unit commercial AC
installs -- these report no oneUiVersion, same as the other RAC/PRAC
boards. Once routed to the existing airconditioner registry, every
resource in the reporter's dump already binds except one new
SAC-specific installation-topology blob, now ignored.
2026-07-24 04:06:31 +00:00
36 changed files with 1809 additions and 166 deletions
+125 -5
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import datetime
import json
import logging
import re
import selectors
@@ -15,6 +16,10 @@ from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.selector import (
ObjectSelector,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
TextSelectorConfig,
TextSelectorType,
@@ -396,22 +401,44 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
class LocalThingsOptionsFlow(config_entries.OptionsFlow):
"""Per-device override of the remote-control-off write block (issue
#54). The block exists because most devices reject writes outright
while remote control is off and a clear error beats a silent
"""Per-device options: the remote-control-off write-block override
(issue #54) plus a debug panel for writing an arbitrary body to an
arbitrary resource href, so a user can pin down device-specific write
behavior without waiting on a new release.
The remote-control override exists because most devices reject writes
outright while remote control is off and a clear error beats a silent
device-side rejection -- but not every model actually enforces that,
so this lets a user who's confirmed their device accepts writes anyway
turn the block off for just that device rather than it being
hardcoded on for everyone."""
hardcoded on for everyone. The debug panel goes further: it bypasses
that block (and every write_fn/validate_fn) entirely, sending exactly
the body the user types to whatever href they pick.
"""
def __init__(self) -> None:
self._debug_href: str = ""
self._debug_result: tuple[int, dict] | None = None
def _coordinator(self):
return self.hass.data.get(DOMAIN, {}).get(self.config_entry.entry_id)
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
return self.async_show_menu(
step_id="init",
menu_options=["settings", "debug_write"],
)
async def async_step_settings(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
if user_input is not None:
return self.async_create_entry(data=user_input)
return self.async_show_form(
step_id="init",
step_id="settings",
data_schema=vol.Schema({
vol.Required(
CONF_BYPASS_REMOTE_CONTROL,
@@ -421,3 +448,96 @@ class LocalThingsOptionsFlow(config_entries.OptionsFlow):
): bool,
}),
)
async def async_step_debug_write(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
coord = self._coordinator()
if coord is None:
return self.async_abort(reason="not_loaded")
if user_input is not None:
self._debug_href = user_input["href"]
return await self.async_step_debug_edit()
hrefs = sorted(coord.last_resources.keys())
return self.async_show_form(
step_id="debug_write",
data_schema=vol.Schema({
vol.Required("href"): SelectSelector(SelectSelectorConfig(
options=hrefs,
custom_value=True,
mode=SelectSelectorMode.DROPDOWN,
)),
}),
)
def _show_debug_edit_form(
self, href: str, current: dict, errors: dict[str, str], payload,
) -> FlowResult:
return self.async_show_form(
step_id="debug_edit",
data_schema=vol.Schema({
vol.Required(
"payload", default=(payload if payload is not None else {}),
): ObjectSelector(),
}),
errors=errors,
description_placeholders={
"href": href,
"current_value": (
json.dumps(current, indent=2, ensure_ascii=False)
if current else "(no cached value)"
),
},
)
async def async_step_debug_edit(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
coord = self._coordinator()
if coord is None:
return self.async_abort(reason="not_loaded")
href = self._debug_href
current = coord.resource(href)
if user_input is not None:
payload = user_input.get("payload") or {}
if not isinstance(payload, dict) or not payload:
return self._show_debug_edit_form(
href, current, {"payload": "empty_payload"}, payload
)
try:
code, new_rep = await coord.async_raw_write(href, payload)
except Exception: # noqa: BLE001 - surfaced to the user below
_LOGGER.exception("debug raw write failed for %s", href)
return self._show_debug_edit_form(
href, current, {"base": "write_failed"}, payload
)
self._debug_result = (code, new_rep)
return await self.async_step_debug_result()
return self._show_debug_edit_form(href, current, {}, None)
async def async_step_debug_result(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
code, new_rep = self._debug_result or (0, {})
return self.async_show_menu(
step_id="debug_result",
menu_options=["debug_write", "finish"],
description_placeholders={
"code": f"{code >> 5}.{code & 0x1f:02d} ({code:#04x})",
"new_value": (
json.dumps(new_rep, indent=2, ensure_ascii=False)
if new_rep else "(no value returned)"
),
},
)
async def async_step_finish(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
# Close the flow without altering saved options.
return self.async_create_entry(data=dict(self.config_entry.options))
+74 -2
View File
@@ -22,7 +22,7 @@ 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 remote_control_enabled
from .registry.capabilities.common import merge_options_field, remote_control_enabled
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -612,7 +612,24 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# introduced its own races around overlapping writes to the same
# href. Simpler and safer to just hold the guard for the full,
# generously-sized window and let it expire on its own.
self._observe.apply(write_href, body, source='optimistic')
# write_fn bodies that touch x.com.samsung.da.options carry only the
# changed token(s) now (issue #54: confirmed sufficient on the wire --
# the device merges by prefix itself), not the whole packed array.
# observe.apply()'s field-level {**cached, **rep} merge doesn't know
# that -- handed the bare token list, it would replace the cached
# field outright and wipe every sibling option for the rest of the
# settle window. Pre-merge it here the same way the device does, so
# the optimistic cache entry stays complete; the minimal `body` below
# is still exactly what goes out over the wire.
optimistic_body = body
new_options = body.get('x.com.samsung.da.options')
if isinstance(new_options, list):
cached_options = (self._cache.get(write_href) or {}).get('x.com.samsung.da.options')
optimistic_body = {
**body,
'x.com.samsung.da.options': merge_options_field(cached_options, new_options),
}
self._observe.apply(write_href, optimistic_body, source='optimistic')
self._observe.mark_write_pending(
write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
)
@@ -630,3 +647,58 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._log.error("command failed for %s: %s", write_href, e)
else:
await self.async_request_refresh()
# ------------------------------------------------------------------
# Debug raw write (issue #54): a power-user escape hatch for the
# options-flow debug panel, letting a user POST an arbitrary partial
# body to an arbitrary href to pin down device-specific write behavior
# without waiting on a new release. Deliberately bypasses the
# remote-control block and every write_fn/validate_fn above -- that's
# the whole point, so use with care.
# ------------------------------------------------------------------
def _raw_write_blocking(self, path_segs: list[str], body: dict, href: str) -> tuple[int, dict]:
"""Debug primitive: POST an arbitrary patch, then read the href
back for ground truth. Blocking -- runs in executor."""
if self._session is None:
self._connect_session()
sess = self._session
if sess is None:
raise RuntimeError("no session")
code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S)
self._log.warning("DEBUG raw write POST %s %r → code %#04x", href, body, code)
new_rep: dict = {}
try:
sess.pace()
rcode, payload = sess.get(path_segs, timeout=10.0)
if rcode == 0x45 and payload:
rep = cbor2.loads(payload)
if isinstance(rep, dict):
self._observe.apply(href, rep, source='poll')
new_rep = rep
except Exception as e:
self._log.debug("raw write follow-up read failed: %s", e)
return code, new_rep
async def async_raw_write(self, href: str, body: dict) -> tuple[int, dict]:
"""Debug-only arbitrary write (issue #54). Bypasses the
remote-control block and all write_fn/validate_fn logic; sends
`body` verbatim as a partial-rep PATCH to `href`. Returns
(coap_code, new_rep) where new_rep is the href's value read back
right after the write. Used by the options-flow debug panel to
help users pin down device-specific write behavior without a new
release."""
if not isinstance(body, dict) or not body:
raise ServiceValidationError("Debug write payload must be a non-empty object.")
path_segs = [s for s in href.strip('/').split('/') if s]
if not path_segs:
raise ServiceValidationError("A resource href is required.")
norm_href = '/' + '/'.join(path_segs)
async with self._session_lock:
code, new_rep = await self.hass.async_add_executor_job(
self._raw_write_blocking, path_segs, body, norm_href
)
# Hasten a full summary poll so entities on other resources catch
# up too -- a debug write can affect siblings, not just its href.
await self.async_request_refresh()
return code, new_rep
+19 -1
View File
@@ -66,12 +66,30 @@ class LocalThingsEntity(CoordinatorEntity[LocalThingsCoordinator]):
self._attr_name = f"{bound.instance_name} {_derive_name(bound.desc.key)}".strip()
else:
self._attr_name = _derive_name(self._state_key)
self._attr_translation_key = bound.desc.translation_key
self._attr_icon = bound.desc.icon
raw_cat = bound.desc.entity_category
self._attr_entity_category = EntityCategory(raw_cat) if raw_cat else None
self._attr_entity_registry_enabled_default = bound.desc.enabled_default
@property
def translation_key(self) -> str | None:
"""Override Entity.translation_key (a property upstream, not a
plain attribute) so a callable descriptor -- e.g.
laundry.cycle_select's table-id-gated resolver -- is re-evaluated
against live coordinator data on every access, not resolved once
at construction time.
Discovery runs on the first /device/0 poll, which the entity
registry already documents can hand a sibling resource an empty
stub rep before it's actually been fetched (see _is_included's
docstring) -- a static one-time resolution here would risk baking
in a permanent None (no translation) for the entity's whole
lifetime if that stub hadn't populated yet, even once the real
value arrives on a later poll.
"""
tk = self._bound.desc.translation_key
return tk(self.coordinator.last_resources) if callable(tk) else tk
@property
def device_info(self) -> DeviceInfo:
return self.coordinator.device_info
+1 -1
View File
@@ -12,5 +12,5 @@
"pyOpenSSL>=23.0",
"smartthings-local>=0.1.0"
],
"version": "0.11.0"
"version": "0.11.1"
}
@@ -113,6 +113,15 @@ def for_device_by_model(model_num: str, description: str) -> Optional[DeviceRegi
# 'P' sits between the underscore and 'RAC' in that token).
if key is None and '_RAC_' in (model_num or ''):
key = 'airconditioner'
# System air conditioners (multi-indoor-unit commercial installs, e.g.
# A-CAWW-TP2-20-COMMON, issue #52) report no oneUiVersion either and
# carry the '-CAWW-' board-family token instead of '_RAC_'/'_PRAC_'.
# Same TP1X/TP2X-class resource surface as the room-AC models above
# (confirmed by the issue #52 dump binding cleanly against the existing
# airconditioner registry once routed here), plus one new SAC-specific
# resource (see airconditioner.py's _AC_IGNORED).
if key is None and '-CAWW-' in (model_num or '').upper():
key = 'airconditioner'
# Air purifiers (e.g. ARTIK051_TVTL_18K, issue #56) report no
# oneUiVersion either, and carry the '_TVTL_' board-family token.
if key is None and '_TVTL_' in (model_num or ''):
@@ -1,6 +1,5 @@
"""Capabilities for the Samsung ARTIK051_TVTL-class air purifier family
(model AX60R5080WD/SE, issue #56 -- verified against two independent
diagnostics dumps of the same internal model).
(model AX60R5080WD/SE, issue #56).
Power, kids-lock, remote-control, alarms, and the energy meter are the shared
common.py capabilities (this family exposes the standard /power/0+/power/vs/0
@@ -8,37 +7,44 @@ pair and /alarms/vs/0, /energy/consumption/vs/0). /diagnosis/vs/0 reuses
dishwasher.DIAGNOSIS -- identical field/write contract
(x.com.samsung.da.diagnosisStart, 'Ready' on both dumps).
Two things are deliberately left as raw, unwritable diagnostic sensors rather
than modeled as real controls, per the "don't guess" rule:
/mode/vs/0's x.com.samsung.da.options array packs multiple independent
'<Prefix>_<value>' flags into one list -- the same packed-list contract
laundry.py's option_value/option_write already model for /course/vs/0's
options[] (reused directly below, just against this family's own href). Per
issue #56's follow-up (five diagnostics dumps captured with the physical unit
set to Auto/Sleep/Low/Medium/High):
Light_On / Light_Off -- a plain on/off flag; MODE below models it as a
real switch, RMW-replacing just that one entry.
Comode_Off -- read 'Off' on *every* one of the five dumps,
including High/Low/Medium/Auto -- confirms this
is NOT the fan-speed selector (ruling out the
original guess); exposed read-only since its
actual purpose is still unconfirmed.
OptionCode_60282 -- confirmed opaque/not user-facing in the
SmartThings app; not modeled (same treatment as
range_hood's OptionCode_* token on the same
href).
Blooming_* -- confirmed to have no corresponding SmartThings
app setting; dropped entirely rather than kept
as an unexplained diagnostic (it did track 1:1
with Sleep mode across the five dumps -- 0 in
Sleep, 6 otherwise -- so it's plausibly an
automatic side effect of sleep mode, e.g. a
display-dimming level, but that's still a guess).
/airflow/0, /airflow/vs/0 -- OCF-standard + vendor pair for fan speed/
direction, both zeroed/'Off' on every dump seen (device was off in both).
No supportedSpeed/supportedModes list is present anywhere in either dump
to confirm the valid range, so a write-capable fan/select isn't safe to
build yet -- see the issue #56 request for a running-state dump.
/mode/vs/0's x.com.samsung.da.options array packs multiple independent
'<Prefix>_<value>' flags into one list -- the same packed-list/RMW
contract laundry.py's option_value/replace_in_options already model for
/course/vs/0's options[] (reused directly below, just against this
family's own href). Of the tokens seen:
Light_On / Light_Off -- read as a plain on/off flag; MODE below
models it as a real switch, RMW-
replacing just that one list entry.
Comode_Off -- never seen non-'Off' on these dumps;
likely the fan operating mode the issue
describes (Auto/Sleep/1/2/3), but
unconfirmed -- exposed read-only.
Blooming_0 / Blooming_6 -- meaning unconfirmed; exposed read-only.
OptionCode_60282 -- opaque, unchanged across both dumps;
not modeled (same treatment as
range_hood's OptionCode_* token on the
same href).
/airflow/0 and /airflow/vs/0's `speed` still isn't modeled as a real
fan-speed control: across the same five dumps it read 0 for both Auto *and*
High, and 3 for Low/Medium *and* Sleep -- not a monotonic mapping to any
selectable level, and the dumps were all captured within about three minutes
of each other (only one poll cycle apart at this integration's 30s summary
interval), so the values may not have settled after each change before the
diagnostics snapshot was taken. Exposed read-only pending a confirmed,
stable capture -- see the issue #56 discussion for what's needed.
"""
from ..capability import Capability
from ..entities import BinarySensorDesc, SensorDesc, SwitchDesc
from .common import int_or_none, sensor_item_value
from .laundry import bool_option_exists, bool_option_value, option_value, replace_in_options
from .laundry import bool_option_exists, bool_option_value, option_value, option_write
_AIR_QUALITY_SENSORS = (
('dust', 'Dust', 'mdi:blur', 'Dust'),
@@ -68,10 +74,12 @@ def _consumable_state(items, name):
return None
# FilterProgress is a raw 0-100 percentage in both dumps (100 and 62); which
# end of that scale means "replace me" isn't confirmed from the dump alone,
# so the entity is named after the raw field rather than asserting a
# direction (see issue #56 follow-up questions).
# FilterProgress is a 0-100 percentage counting up as the filter wears --
# confirmed via issue #56: the SmartThings app shows "Filter needs changing"
# once this reaches 100, so 100 means fully used, not "brand new." Named
# after the raw field (matching the AC/range_hood filterUsage convention,
# which counts the same direction) rather than "filter life," which would
# imply the opposite direction.
FILTER = Capability(
href='/consumable/vs/0',
poll_tier='cold',
@@ -127,9 +135,14 @@ AIRFLOW_VS_FALLBACK = Capability(
def _light_write(payload, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
# option_write's single-token write is confirmed on a washer's
# /course/vs/0 (issue #54), NOT independently on this family's
# /mode/vs/0 -- extrapolated on the assumption the same vendor field
# merges the same way everywhere. If some unit replaces the field
# outright instead, this would drop Comode/OptionCode alongside it on
# the next light toggle; revisit if a real device report surfaces that.
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, 'Light', payload),
'x.com.samsung.da.options': option_write('Light', payload),
}
@@ -142,15 +155,12 @@ MODE = Capability(
rep_fn=bool_option_value('Light'),
exists_fn=bool_option_exists('Light'),
write_fn=_light_write),
# Read-only pending issue #56 follow-up -- see module docstring.
# Read-only -- confirmed NOT the fan-speed selector (see module
# docstring), actual purpose still unconfirmed.
SensorDesc(key='operating_mode', name='Operating mode', icon='mdi:fan',
entity_category='diagnostic',
rep_fn=lambda rep: option_value(rep.get('x.com.samsung.da.options'), 'Comode'),
exists_fn=bool_option_exists('Comode')),
SensorDesc(key='blooming_level', name='Blooming level', icon='mdi:flower',
entity_category='diagnostic',
rep_fn=lambda rep: option_value(rep.get('x.com.samsung.da.options'), 'Blooming'),
exists_fn=bool_option_exists('Blooming')),
),
)
@@ -243,6 +243,11 @@ _AC_IGNORED = [
'/remotetemperature/vs/0', # external temp-sensor feed (unset on this unit)
'/reserverulesets/vs/0', # opaque hex-encoded schedule reservation blob
'/welcome/temperature/vs/0', # welcome-cooling plumbing
# System-AC-only (multi-indoor-unit commercial installs, e.g.
# A-CAWW-TP2-20-COMMON, issue #52): opaque hex-encoded installation
# topology -- indoor/outdoor unit pairing, per-unit serials, MCU info.
# Commissioning-time plumbing, not user-actionable appliance state.
'/sac/installationinfo/vs/0',
]
# Built as bare no-entity caps; folded into the AC registry (not global).
@@ -65,6 +65,33 @@ def _active_alarm_codes(items):
return ', '.join(codes) if codes else 'none'
def merge_options_field(cached, new_tokens):
"""Merge freshly-written `<Prefix>_<Value>` tokens into a cached
x.com.samsung.da.options[]-style array the same way the device itself
merges them: match by prefix, replace if present, append if not.
Confirmed on real hardware (issue #54) that a write only needs to carry
the changed token(s), not the whole array -- see laundry.option_write /
oven._option_write for the write side. This is the read side of that
same fact: coordinator.async_send_command uses it to keep the
optimistic cache entry for the written href complete (every sibling
option still present) during the write-settle window, since the wire
body it applies straight to the cache no longer carries them."""
merged = list(cached or [])
for token in new_tokens or ():
if not isinstance(token, str) or '_' not in token:
continue
prefix = token.split('_', 1)[0]
replaced = False
for i, o in enumerate(merged):
if isinstance(o, str) and o.startswith(prefix + '_'):
merged[i] = token
replaced = True
if not replaced:
merged.append(token)
return merged
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
@@ -48,7 +48,8 @@ DRYER_SETTINGS = Capability(
DRYER_COURSE = Capability(
href='/course/vs/0',
entities=(
cycle_select(translation_key='dryer_cycle', icon='mdi:tumble-dryer'),
cycle_select(translation_key='dryer_cycle', icon='mdi:tumble-dryer',
table_href='/st/dryercourse/vs/0'),
),
)
@@ -92,7 +92,7 @@ ICEMAKER_NIGHTTIME = Capability(
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['icemaker', 'nighttime', 'vs', '0'],
{'ice.night.status': 'On' if p else 'Off'})),
{'ice.night.status': 'On' if p == 'On' else 'Off'})),
),
)
@@ -168,7 +168,7 @@ DOOR_ALERT = Capability(
def _status_lock_write(field):
return lambda p, rep, href=None: (
['status', 'lock', 'vs', '0'],
{field: 'On' if p else 'Off'}
{field: 'On' if p == 'On' else 'Off'}
)
@@ -212,7 +212,7 @@ DEFROST_DELAY = Capability(
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['defrost', 'delay', 'vs', '0'],
{'x.com.samsung.da.delayDefrost': 'On' if p else 'Off'})),
{'x.com.samsung.da.delayDefrost': 'On' if p == 'On' else 'Off'})),
),
)
@@ -300,7 +300,7 @@ WELCOME_LIGHTING = Capability(
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['proximity', 'vs', '0'],
{'status': 'On' if p else 'Off'})),
{'status': 'On' if p == 'On' else 'Off'})),
),
)
@@ -440,7 +440,7 @@ CABINET_LIGHT = Capability(
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['cabinet', 'light', 'total', 'vs', '0'],
{'light.dimming.status': 'On' if p else 'Off'})),
{'light.dimming.status': 'On' if p == 'On' else 'Off'})),
),
)
@@ -669,7 +669,7 @@ ICEMAKER_STATUS_FALLBACK = Capability(
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['icemaker', 'status', 'vs', '0'],
{'x.com.samsung.da.iceMaker': 'On' if p else 'Off'})),
{'x.com.samsung.da.iceMaker': 'On' if p == 'On' else 'Off'})),
),
)
@@ -146,9 +146,13 @@ BUZZER_SOUND = Capability(
# Cycle selection over /course/vs/0.
#
# The selected course and every other user-tunable option ride in the
# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens;
# a write is a read-modify-write of that whole array (cycle_write). The set of
# *selectable* courses is not hardcoded -- it's read live from
# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens.
# Confirmed on real hardware (issue #54): a write only needs to carry the one
# changed token -- `{'x.com.samsung.da.options': ['SoftenerLevelCtrl_2']}` --
# the device matches by prefix, evicts the stale token, and merges the result
# into the array itself. No read-modify-write of the whole array needed (see
# option_write). The set of *selectable* courses is not hardcoded -- it's read
# live from
# x.com.samsung.da.editCourseList on /wm/editcourse/vs/0 (cycle_options), so we
# never show a course a given model doesn't have or hide one it does. Course
# codes are uppercase hex; display names live in translations under
@@ -157,6 +161,11 @@ BUZZER_SOUND = Capability(
# washer.py's course comment has the byte-level evidence for why the options[]
# MostUsed_* entry is *not* a trustworthy second source.
#
# Some boards populate /wm/editcourse/vs/0 without ever filling in
# editCourseList itself (issue #1) -- cycle_options() falls back to deriving
# the same list from /course/vs/0's own supportedOptions in that case; see
# _course_codes_from_supported_options for the byte-level evidence.
#
# Shared verbatim by washer, dishwasher, and dryer -- all DA_WM_-family boards
# expose the same /course/vs/0 options contract.
# ---------------------------------------------------------------------------
@@ -176,7 +185,10 @@ def parse_edit_course_list(raw):
def cycle_options(resources):
rep = resources.get('/wm/editcourse/vs/0') or {}
return parse_edit_course_list(rep.get('x.com.samsung.da.editCourseList'))
codes = parse_edit_course_list(rep.get('x.com.samsung.da.editCourseList'))
if codes:
return codes
return _course_codes_from_supported_options(resources.get('/course/vs/0') or {})
def option_value(options, prefix):
@@ -187,29 +199,129 @@ def option_value(options, prefix):
return None
def replace_in_options(options, prefix, new_value):
return [f"{prefix}_{new_value}" if isinstance(o, str) and o.startswith(prefix + '_') else o
for o in options]
def _course_codes_from_supported_options(course_rep):
"""Fallback for an empty/missing editCourseList: derive the selectable
course list from /course/vs/0's own x.com.samsung.da.supportedOptions
instead (issue #1: some DA_WM_TP1/TP2-class boards populate the
/wm/editcourse/vs/0 href but never fill in editCourseList itself).
supportedOptions is a 1-hex-nibble header followed by one fixed-width
record per selectable course, self-indexed rather than positional --
the first byte of every record is that course's own hex code, just in
the firmware's own internal order, not editCourseList's. Confirmed
against six independent real-world washer/dryer/dishwasher dumps: every
one divides evenly into `header + N * K bytes` with fully unique first
bytes across all N records, at the record's true byte width. (What the
rest of each record encodes is still unconfirmed -- this only uses the
course-code byte.)
Two guards, deliberately conservative rather than guessing further: the
derived codes must (a) all be distinct -- a real course table, not
noise -- and (b) include whatever course is currently selected
(x.com.samsung.da.options' Course_<code> token), which must always be a
member of its own device's valid list. If no split satisfies both, this
returns [] rather than guess.
Among splits that satisfy both, the *smallest* passing K wins, rather
than requiring a single unambiguous one -- more than one K reliably
does pass on real data (e.g. the shipped dishwasher fixture: true
K=7 passes, but so do 10, 14, and 35, none of which are multiples of
7 -- position 0 always lands on the same real course code regardless
of K, which is enough on its own to satisfy the current-course guard
for several unrelated splits). Smallest-K-wins is a heuristic, not a
proof: it matches the confirmed answer on every one of six independent
real-world dumps this was checked against, but a coincidentally
unique, current-course-inclusive *smaller* K is not mathematically
impossible on some future device, and would be picked silently. Not
guarded against further here, since course tables are typically large
enough (double digits) that colliding by chance on both checks is
unlikely, and no device seen so far actually needs it.
"""
raw = course_rep.get('x.com.samsung.da.supportedOptions')
hexstr = raw[0] if isinstance(raw, list) and raw else raw
if not isinstance(hexstr, str) or len(hexstr) < 3:
return []
body = hexstr[1:]
if len(body) % 2:
return []
total_bytes = len(body) // 2
current = option_value(course_rep.get('x.com.samsung.da.options'), 'Course')
for k in range(1, total_bytes + 1):
if total_bytes % k:
continue
n = total_bytes // k
if n < 2:
continue
firsts = [body[i * k * 2:i * k * 2 + 2] for i in range(n)]
if len(set(firsts)) != n:
continue
if current is not None and current not in firsts:
continue
return firsts
return []
def option_write(prefix, new_value):
"""A one-token x.com.samsung.da.options write -- see the module comment
above cycle_options for why this doesn't read/rewrite the whole array."""
return [f'{prefix}_{new_value}']
def cycle_write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, 'Course', p),
'x.com.samsung.da.options': option_write('Course', p),
}
def cycle_select(*, translation_key, icon):
def _table_id(resources, table_href):
rep = resources.get(table_href) or {}
return rep.get('x.com.samsung.da.st.courseTable')
def cycle_select(*, translation_key, icon, table_href=None):
"""A 'Cycle' select over /course/vs/0, labelled from `translation_key`.
The caller supplies the family's translation key (washer_cycle /
dishwasher_cycle / dryer_cycle) and icon; the option list, current value,
and write path are all shared.
The option list, current value, and write path are all shared across
washer/dryer/dishwasher; only the translation is family- (and, for
washer/dryer, board-) specific.
table_href (washer/dryer only -- see washer.py/dryer.py's call sites)
suffixes translation_key with the device's own course-table id, read
from /st/washercourse/vs/0 or /st/dryercourse/vs/0's
x.com.samsung.da.st.courseTable (e.g. 'washer_cycle' + 'Table_02' ->
'washer_cycle_table_02'). No table id available at all -- the href
absent or empty -- gets no translation_key, i.e. the raw course code
displayed as-is.
This matters because course codes are NOT guaranteed consistent across
board generations sharing the same /course/vs/0 contract: every code in
washer_cycle_table_02 was confirmed against Table_02-reporting devices
(DA_WM_TP1/TP2 boards); FlexWash's older DA_WM_A51 board reports
Table_00 instead, so the same hex code could mean a different course
there for all we've verified. Building the key from whatever table the
device actually reports, rather than gating a single hardcoded key on
an exact match, means a table we haven't built translations for yet
(like Table_00) just falls through Home Assistant's own missing-
translation handling to the same raw-code display -- exactly what
happens today for any individual code within a table's translations
that isn't populated yet -- and adding one later needs new strings.json
entries, not a code change here.
Left at its default for dishwasher, which has no equivalent table-id
resource in any dump seen and no evidence its course codes vary by
table the way washer/dryer's do -- there's nothing to build a
table-specific key from.
"""
key = translation_key
if table_href is not None:
def key(resources):
table = _table_id(resources, table_href)
return f'{translation_key}_{table.lower()}' if table else None
return SelectDesc(
key='cycle', name='Cycle', icon=icon, translation_key=translation_key,
key='cycle', name='Cycle', icon=icon, translation_key=key,
options=cycle_options,
exists_fn=lambda rep, resources: bool(cycle_options(resources)),
rep_fn=lambda rep: option_value(rep.get('x.com.samsung.da.options'), 'Course'),
@@ -231,11 +343,10 @@ 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:
if not rep.get('x.com.samsung.da.options'):
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, prefix, p),
'x.com.samsung.da.options': option_write(prefix, p),
}
return write
@@ -122,10 +122,17 @@ def _option_value(options, prefix):
return None
def _replace_in_options(options, prefix, new_value):
"""Return a new options list with the `<prefix>_*` slot replaced."""
return [f"{prefix}_{new_value}" if o.startswith(prefix + '_') else o
for o in options]
def _option_write(prefix, new_value):
"""A one-token x.com.samsung.da.options write, mirroring
laundry.option_write. NOT independently confirmed on an oven -- issue
#54 only confirmed prefix-merge-on-write for a washer's /course/vs/0.
This extrapolates that same vendor field/contract to the oven's
/mode/vs/0, on the assumption the firmware handles the array the same
way there. If that assumption is wrong for some oven, a device that
replaces the field outright instead of merging would drop every other
option in it (Sound/fastpreheat/etc.) on the next write -- revisit if a
real device report surfaces that."""
return [f'{prefix}_{new_value}']
# ---------------------------------------------------------------------------
@@ -175,47 +182,41 @@ def _oven_mode_write(p, rep, href=None):
def _lamp_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:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'UpperLamp', p),
'x.com.samsung.da.options': _option_write('UpperLamp', p),
}
def _sound_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:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'Sound', p),
'x.com.samsung.da.options': _option_write('Sound', p),
}
def _fastpreheat_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:
if not rep.get('x.com.samsung.da.options'):
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'fastpreheat', p),
'x.com.samsung.da.options': _option_write('fastpreheat', p),
}
def _naturalsteam_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:
if not rep.get('x.com.samsung.da.options'):
return None
if not any(o.startswith('NaturalSteam_') for o in opts):
opts = opts + [f'NaturalSteam_{p}']
else:
opts = _replace_in_options(opts, 'NaturalSteam', p)
return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts}
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _option_write('NaturalSteam', p),
}
# ---------------------------------------------------------------------------
@@ -19,7 +19,7 @@ from ..capability import Capability
from ..entities import BinarySensorDesc, SelectDesc, SensorDesc
from .laundry import (
bool_option_exists, bool_option_switch, cycle_options, cycle_select, hex_pairs, option_value,
replace_in_options,
option_write,
)
# ---------------------------------------------------------------------------
@@ -215,8 +215,7 @@ def _dosing_level(prefix):
def _level_write(prefix):
def write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
if not rep.get('x.com.samsung.da.options'):
return None
# `p` is the zero-padded supported code the UI selected (e.g. '03');
# the device stores the level un-padded (e.g. '3'), matching how it
@@ -226,7 +225,7 @@ def _level_write(prefix):
except (TypeError, ValueError):
native = p
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': replace_in_options(opts, prefix, native),
'x.com.samsung.da.options': option_write(prefix, native),
}
return write
@@ -293,7 +292,8 @@ def _bool_option_switch(key, name, icon, prefix, availability_field):
WASHER_COURSE = Capability(
href='/course/vs/0',
entities=(
cycle_select(translation_key='washer_cycle', icon='mdi:washing-machine'),
cycle_select(translation_key='washer_cycle', icon='mdi:washing-machine',
table_href='/st/washercourse/vs/0'),
SensorDesc(key='drum_clean_cycles_remaining', name='Drum clean due in',
icon='mdi:washing-machine-alert', unit='cycles',
state_class='measurement',
@@ -28,7 +28,12 @@ class SamsungEntityDescription:
key: str
field: str = ''
name: Optional[str] = None
translation_key: Optional[str] = None
translation_key: Any = None # str | Callable[[dict[str, dict]], Optional[str]]
# callable form receives the coordinator's full href->rep resource
# snapshot and returns the key to use (or None for no translation this
# device) -- for a descriptor shared across board generations whose
# state-code meaning isn't guaranteed consistent between them; see
# laundry.cycle_select's table-id-gated resolver.
icon: Optional[str] = None
entity_category: Optional[str] = None # 'diagnostic' | 'config' | None
enabled_default: bool = True
+13 -7
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import re
from typing import Optional
from homeassistant.components.select import SelectEntity
from homeassistant.config_entries import ConfigEntry
@@ -31,9 +32,14 @@ async def async_setup_entry(
_CAMEL_BOUNDARY_RE = re.compile(r'(?<=[a-z0-9])(?=[A-Z])')
def _display(value, desc: SelectDesc):
def _display(value, translation_key: Optional[str]):
"""Turn a raw device option/state value into what's shown in the UI.
`translation_key` is the entity's already-resolved key (SelectDesc.
translation_key can itself be a callable -- see entities.py -- so
callers pass the resolved value, e.g. self.translation_key, not
the raw descriptor field).
An entity with a translation_key looks its state up in strings.json,
and hassfest requires those keys to be lowercase -- so those values
must be lowercased exactly to match, and the device still expects
@@ -50,7 +56,7 @@ def _display(value, desc: SelectDesc):
"""
if not isinstance(value, str):
return value
if desc.translation_key:
if translation_key:
return value.lower()
if value.islower():
return value.replace('_', ' ').title()
@@ -63,7 +69,7 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
super().__init__(coordinator, bound)
desc: SelectDesc = bound.desc
if not desc.options_field and not callable(desc.options):
self._attr_options = [_display(o, desc) for o in desc.options]
self._attr_options = [_display(o, self.translation_key) for o in desc.options]
def _raw_options(self) -> list[str]:
desc: SelectDesc = self._bound.desc
@@ -83,17 +89,17 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
def options(self) -> list[str]:
desc: SelectDesc = self._bound.desc
if desc.options_field or callable(desc.options):
return [_display(o, desc) for o in self._raw_options()]
return [_display(o, self.translation_key) for o in self._raw_options()]
return self._attr_options
@property
def current_option(self):
raw = (self.coordinator.data or {}).get(self._state_key)
return _display(raw, self._bound.desc)
return _display(raw, self.translation_key)
async def async_select_option(self, option: str) -> None:
desc: SelectDesc = self._bound.desc
raw = next(
(o for o in self._raw_options() if _display(o, desc) == option), option
(o for o in self._raw_options() if _display(o, self.translation_key) == option),
option,
)
await self.coordinator.async_send_command(self._bound, raw)
+38 -2
View File
@@ -71,7 +71,7 @@
"8f": "Baby Care"
}
},
"washer_cycle": {
"washer_cycle_table_02": {
"state": {
"1c": "Eco 40-60",
"1d": "Super Speed",
@@ -103,7 +103,7 @@
"1f": "Intense Cold"
}
},
"dryer_cycle": {
"dryer_cycle_table_03": {
"state": {
"16": "Cotton",
"18": "Synthetics",
@@ -250,11 +250,47 @@
"step": {
"init": {
"title": "Local Things Options",
"menu_options": {
"settings": "Remote control write settings",
"debug_write": "Debug: write to a resource"
}
},
"settings": {
"title": "Remote control write settings",
"description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.",
"data": {
"bypass_remote_control_lock": "Allow writes even when remote control is reported off"
}
},
"debug_write": {
"title": "Debug: write to a resource",
"description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.",
"data": {
"href": "Resource href"
}
},
"debug_edit": {
"title": "Debug write: {href}",
"description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.",
"data": {
"payload": "Payload to write"
}
},
"debug_result": {
"title": "Debug write result",
"description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:",
"menu_options": {
"debug_write": "Write another resource",
"finish": "Finish"
}
}
},
"error": {
"empty_payload": "Enter at least one field to write.",
"write_failed": "The write failed. Check the Home Assistant logs for details."
},
"abort": {
"not_loaded": "This device isn't connected yet. Try again once it has loaded."
}
},
"issues": {
@@ -71,7 +71,7 @@
"8f": "Baby Care"
}
},
"washer_cycle": {
"washer_cycle_table_02": {
"state": {
"1c": "Eco 40-60",
"1d": "Super Speed",
@@ -103,7 +103,7 @@
"1f": "Intense Cold"
}
},
"dryer_cycle": {
"dryer_cycle_table_03": {
"state": {
"16": "Cotton",
"18": "Synthetics",
@@ -250,11 +250,47 @@
"step": {
"init": {
"title": "Local Things Options",
"menu_options": {
"settings": "Remote control write settings",
"debug_write": "Debug: write to a resource"
}
},
"settings": {
"title": "Remote control write settings",
"description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.",
"data": {
"bypass_remote_control_lock": "Allow writes even when remote control is reported off"
}
},
"debug_write": {
"title": "Debug: write to a resource",
"description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.",
"data": {
"href": "Resource href"
}
},
"debug_edit": {
"title": "Debug write: {href}",
"description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.",
"data": {
"payload": "Payload to write"
}
},
"debug_result": {
"title": "Debug write result",
"description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:",
"menu_options": {
"debug_write": "Write another resource",
"finish": "Finish"
}
}
},
"error": {
"empty_payload": "Enter at least one field to write.",
"write_failed": "The write failed. Check the Home Assistant logs for details."
},
"abort": {
"not_loaded": "This device isn't connected yet. Try again once it has loaded."
}
},
"issues": {
+486
View File
@@ -0,0 +1,486 @@
{
"device0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/aisleep/vs/0",
"rep": {
"x.com.samsung.da.displayNightMode": "Off",
"x.com.samsung.da.elapsedTime": "0",
"x.com.samsung.da.requestFeedback": "Off",
"x.com.samsung.da.resultFeedback": "0",
"x.com.samsung.da.statusFeedback": "Idle",
"x.com.samsung.da.sleepTime": "14002200",
"x.com.samsung.da.sleepMode": "Off"
}
},
{
"href": "/alarms/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "ErrorCode_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-23T05:13:21",
"x.com.samsung.da.state": "Deleted"
},
{
"x.com.samsung.da.id": "1",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "FilterAlarm_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-23T05:13:21",
"x.com.samsung.da.state": "Deleted"
}
]
}
},
{
"href": "/availablecontrolsets/vs/0",
"rep": {
"x.com.samsung.da.sets": "000100B4012C0000120000000000",
"x.com.samsung.da.id": "CAC",
"x.com.samsung.da.version": "1.0"
}
},
{
"href": "/configuration/vs/0",
"rep": {
"x.com.samsung.da.region": "4131000000",
"x.com.samsung.da.airconOptionList": [
"DR",
"SingleCommand_1.0",
"AI_2.0"
]
}
},
{
"href": "/diagnosis/vs/0",
"rep": {}
},
{
"href": "/drlc/vs/0",
"rep": {
"x.com.samsung.da.drlcLevel": "0",
"x.com.samsung.da.durationminutes": "0",
"x.com.samsung.da.start": "1970-01-01T00:00:00Z",
"x.com.samsung.da.override": "Off",
"x.com.samsung.da.realSaving": "Off"
}
},
{
"href": "/energy/consumption/vs/0",
"rep": {
"x.com.samsung.da.instantaneousPower": "153.000000",
"x.com.samsung.da.cumulativePower": "239209",
"x.com.samsung.da.cumulativeSavedPower": "7201",
"x.com.samsung.da.cumulativeUnit": "Wh",
"x.com.samsung.da.instantaneousPowerUnit": "W",
"x.com.samsung.da.cumulativePowerType": "total",
"x.com.samsung.da.saveLocation": "/mnt/usage.db"
}
},
{
"href": "/file/information/vs/0",
"rep": {
"x.com.samsung.timeoffset": "+09:00",
"x.com.samsung.supprtedtype": 1
}
},
{
"href": "/filter/airdustfilter/vs/0",
"rep": {
"x.com.samsung.da.filterUsage": "85",
"x.com.samsung.da.filterUsageResolution": "1",
"x.com.samsung.da.filterStatus": "normal",
"x.com.samsung.da.filterCapacity": "1000",
"x.com.samsung.da.filterCapacityUnit": "Hour",
"x.com.samsung.da.filterResetType": [
"replaceable",
"washable"
]
}
},
{
"href": "/humidity/vs/0",
"rep": {
"x.com.samsung.da.humidity": "0.000000",
"x.com.samsung.da.fivepercentHumidity": "83"
}
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "A-CAWW-TP2-20-COMMON|10241941|600301170016110B40000F2000F2AD00",
"x.com.samsung.da.description": "A-CAWW-TP2-20-COMMON",
"x.com.samsung.da.serialNum": "**REDACTED**",
"x.com.samsung.da.otnDUID": "**REDACTED**",
"x.com.samsung.da.diagProtocolType": "WIFI_HTTPS",
"x.com.samsung.da.diagLogType": [
"errCode",
"dump"
],
"x.com.samsung.da.diagDumpType": "file",
"x.com.samsung.da.diagEndPoint": "SSM",
"x.com.samsung.da.diagMnid": "0AJT",
"x.com.samsung.da.diagSetupid": "023",
"x.com.samsung.da.diagMinVersion": "1.0",
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Software",
"x.com.samsung.da.number": "02274A260512",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "1",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "02419A25051901,FFFFFFFFFFFFFF",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Outdoor",
"x.com.samsung.da.number": "FFFFFFFFFFFFFF,FFFFFFFFFFFFFF"
}
]
}
},
{
"href": "/keepnormalstate/vs/0",
"rep": {
"x.com.samsung.da.keepnormal": 0
}
},
{
"href": "/mode/convenient/vs/0",
"rep": {
"x.com.samsung.da.modes": "Off",
"x.com.samsung.da.supportedModes": [
"Off",
"Nano",
"LongWind",
"Speed",
"Sleep",
"NanoSleep"
]
}
},
{
"href": "/mode/vs/0",
"rep": {
"x.com.samsung.da.workingMode": "Auto",
"x.com.samsung.da.supportedModes": [
"Auto",
"Cool",
"Dry",
"Fan",
"AIComfort"
],
"x.com.samsung.da.modes": [
"Auto"
],
"x.com.samsung.da.options": [
"Sleep_0",
"ArtificialWorking_Off",
"ComfortAICooling_Off",
"AiTempChanged_Off",
"AiTemp_240",
"OutdoorTemp_82",
"Light_Off",
"Volume_100",
"KeyInputPermit_On",
"ModePermit_NoLimit",
"UpdateAllow_NotAllowed",
"DurationOn_419",
"WelcomeCoolingState_Off"
]
}
},
{
"href": "/option/autoclean/vs/0",
"rep": {
"x.com.samsung.da.status": "Stop",
"x.com.samsung.da.settingStatus": "On",
"x.com.samsung.da.progress": "0",
"x.com.samsung.da.supportedStatus": [
"Start",
"Stop"
],
"x.com.samsung.da.supportedSettingStatus": [
"On",
"Off"
]
}
},
{
"href": "/option/muteonce/vs/0",
"rep": {
"muteonce": "Off"
}
},
{
"href": "/otninformation/vs/0",
"rep": {
"x.com.samsung.da.target": "",
"x.com.samsung.da.newVersionAvailable": "false",
"x.com.samsung.da.newVersionNo": "00000000",
"x.com.samsung.da.currentVersionInfo": "00000000",
"otnStatus": "None",
"flashingProgress": "",
"otnTarget": "main",
"otnCompleteDate": "noHistory",
"otnList": [
{
"type": "WIFI",
"modelId": "A-CAWW-TP2-20-COMMON",
"versions": [
"14260512"
],
"visVersion": "260512"
},
{
"type": "Micom",
"modelId": "040010241941FFFFFFFF",
"versions": [
"25051901",
"FFFFFFFF"
],
"visVersion": "250519"
},
{
"type": "Micom",
"modelId": "0400FFFFFFFFFFFFFFFF",
"versions": [
"FFFFFFFF",
"FFFFFFFF"
],
"visVersion": ""
}
]
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "Off"
}
},
{
"href": "/realtimenotiforclient/vs/0",
"rep": {
"x.com.samsung.da.timeforshortnoti": "0",
"x.com.samsung.da.longnotisubscription": "true",
"x.com.samsung.da.periodicnotisubscription": "true"
}
},
{
"href": "/reserverulesets/vs/0",
"rep": {
"x.com.samsung.da.sets": "1F121E121E121E121EFFFFFFFF010001000F0001000E0000001C",
"x.com.samsung.da.id": "CAC",
"x.com.samsung.da.version": "1.0"
}
},
{
"href": "/sac/installationinfo/vs/0",
"rep": {
"installationOptions": [
{
"id": "0",
"option": "17444980C80171700010"
},
{
"id": "1",
"option": "28010000000000000100"
},
{
"id": "2",
"option": "50000000000000000000"
},
{
"id": "3",
"option": "30000000000000000000"
}
],
"installationDeviceNum": [
{
"id": "0",
"number": "01"
},
{
"id": "1",
"number": "04"
},
{
"id": "2",
"number": "00"
}
],
"outdoorInfo": [
{
"id": "0",
"info": "01900A230418",
"serial": "**REDACTED**"
},
{
"id": "1",
"info": "000000000000",
"serial": "**REDACTED**"
},
{
"id": "2",
"info": "000000000000",
"serial": "**REDACTED**"
},
{
"id": "3",
"info": "000000000000",
"serial": "**REDACTED**"
}
],
"mcuInfo": [
{
"id": "0",
"info": ""
},
{
"id": "1",
"info": ""
},
{
"id": "2",
"info": ""
},
{
"id": "3",
"info": ""
},
{
"id": "4",
"info": ""
},
{
"id": "5",
"info": ""
},
{
"id": "6",
"info": ""
},
{
"id": "7",
"info": ""
},
{
"id": "8",
"info": ""
},
{
"id": "9",
"info": ""
},
{
"id": "10",
"info": ""
},
{
"id": "11",
"info": ""
},
{
"id": "12",
"info": ""
},
{
"id": "13",
"info": ""
},
{
"id": "14",
"info": ""
},
{
"id": "15",
"info": ""
}
]
}
},
{
"href": "/temperature/control/vs/0",
"rep": {
"x.com.samsung.da.increment": "0.5"
}
},
{
"href": "/temperatures/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Temperature",
"x.com.samsung.da.desired": "27.0",
"x.com.samsung.da.current": "29.0",
"x.com.samsung.da.maximum": "30",
"x.com.samsung.da.minimum": "18",
"x.com.samsung.da.increment": "0.5",
"x.com.samsung.da.unit": "Celsius"
}
]
}
},
{
"href": "/timezone/vs/0",
"rep": {
"timezoneid": "Asia/Seoul",
"offset": "+09:00",
"DST": "OFF"
}
},
{
"href": "/wind/direction/vs/0",
"rep": {
"x.com.samsung.da.modes": "Fix",
"x.com.samsung.da.supportedModes": [
"Up_And_Low",
"Fix",
"Left_And_Right",
"All"
]
}
},
{
"href": "/wind/strength/vs/0",
"rep": {
"x.com.samsung.da.modes": "0",
"x.com.samsung.da.supportedModes": [
"0",
"1",
"2",
"3"
],
"x.com.samsung.da.modesName": [
"Auto",
"Low",
"Mid",
"High"
]
}
}
]
}
-1
View File
@@ -1,7 +1,6 @@
{
"state_keys": [
"alarm_code",
"blooming_level",
"clean_level",
"device_active",
"diagnosis_status",
+15
View File
@@ -0,0 +1,15 @@
{
"state_keys": [
"air_filter_status",
"air_filter_usage",
"alarm_code",
"auto_clean",
"climate",
"diagnosis_status",
"energy_kwh",
"energy_saved_kwh",
"firmware_update",
"mute_once",
"power_watts"
]
}
+1
View File
@@ -2,6 +2,7 @@
"state_keys": [
"alarm_code",
"child_lock",
"cycle",
"cycle_active",
"delay_start_hours",
"detergent_low",
+156 -1
View File
@@ -279,6 +279,20 @@ def test_probe_marks_washer_as_recognized(monkeypatch):
assert recognized is True
async def test_options_flow_init_shows_menu(hass: HomeAssistant) -> None:
"""The options flow's entry point is now a menu (issue #54's debug
panel lives alongside the remote-control settings), not the settings
form directly."""
entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result['type'] == FlowResultType.MENU
assert result['step_id'] == 'init'
assert set(result['menu_options']) == {'settings', 'debug_write'}
async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
"""The bypass defaults to False, so devices that never touch this
option see no change in the remote-control write block."""
@@ -286,9 +300,12 @@ async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'settings'}
)
assert result['type'] == FlowResultType.FORM
assert result['step_id'] == 'init'
assert result['step_id'] == 'settings'
assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is False
@@ -299,6 +316,9 @@ async def test_options_flow_can_enable_bypass(hass: HomeAssistant) -> None:
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'settings'}
)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={CONF_BYPASS_REMOTE_CONTROL: True}
)
@@ -317,5 +337,140 @@ async def test_options_flow_reflects_previously_saved_value(hass: HomeAssistant)
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'settings'}
)
assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is True
async def test_options_flow_debug_write_shows_hrefs_from_coordinator(
hass: HomeAssistant, mock_coordinator_session,
) -> None:
"""The debug panel's href dropdown is populated from the live
coordinator's cached resources, not a static list."""
entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'debug_write'}
)
assert result['type'] == FlowResultType.FORM
assert result['step_id'] == 'debug_write'
async def test_options_flow_debug_write_aborts_when_device_not_loaded(
hass: HomeAssistant,
) -> None:
"""No coordinator in hass.data (device never finished loading) means
the debug panel has nothing to write to or read from."""
entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
entry.add_to_hass(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'debug_write'}
)
assert result['type'] == FlowResultType.ABORT
assert result['reason'] == 'not_loaded'
async def test_options_flow_debug_edit_writes_and_shows_result(
hass: HomeAssistant, mock_coordinator_session,
) -> None:
"""Picking an href, then submitting a payload, drives
coordinator.async_raw_write and lands on the result menu with the
device's response."""
entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'debug_write'}
)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'href': '/washer/vs/0'},
)
assert result['type'] == FlowResultType.FORM
assert result['step_id'] == 'debug_edit'
with patch(
'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write',
return_value=(0x44, {'a': 1}),
):
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'payload': {'a': 1}},
)
assert result['type'] == FlowResultType.MENU
assert result['step_id'] == 'debug_result'
assert result['description_placeholders']['code'] == '2.04 (0x44)'
async def test_options_flow_debug_edit_rejects_empty_payload(
hass: HomeAssistant, mock_coordinator_session,
) -> None:
entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'debug_write'}
)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'href': '/washer/vs/0'},
)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'payload': {}},
)
assert result['type'] == FlowResultType.FORM
assert result['step_id'] == 'debug_edit'
assert result['errors'] == {'payload': 'empty_payload'}
async def test_options_flow_finish_preserves_existing_options(
hass: HomeAssistant, mock_coordinator_session,
) -> None:
"""Finishing from the debug-result menu must not clobber a previously
saved remote-control bypass setting."""
entry = MockConfigEntry(
domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}',
options={CONF_BYPASS_REMOTE_CONTROL: True},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'debug_write'}
)
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'href': '/washer/vs/0'},
)
with patch(
'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write',
return_value=(0x44, {'a': 1}),
):
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'payload': {'a': 1}},
)
assert result['type'] == FlowResultType.MENU
assert result['step_id'] == 'debug_result'
result = await hass.config_entries.options.async_configure(
result['flow_id'], user_input={'next_step_id': 'finish'}
)
assert result['type'] == FlowResultType.CREATE_ENTRY
assert entry.options[CONF_BYPASS_REMOTE_CONTROL] is True
+22 -14
View File
@@ -42,7 +42,7 @@ def test_expected_entities_present():
'power_switch', 'alarm_code', 'dust', 'fine_dust', 'super_fine_dust',
'odor', 'clean_level', 'filter_progress', 'device_active',
'diagnosis_status', 'fan_speed_level', 'fan_direction',
'display_light', 'operating_mode', 'blooming_level',
'display_light', 'operating_mode',
):
assert key in state, key
@@ -59,6 +59,8 @@ def test_air_quality_sensor_values():
def test_filter_progress_reads_named_consumable_item():
"""FilterProgress is confirmed (issue #56) to count up as the filter
wears -- 100 means fully used and needs replacing, not "brand new"."""
assert _state()['filter_progress'] == 100
@@ -71,9 +73,10 @@ def test_diagnosis_reuses_dishwasher_capability():
def test_light_switch_write_contract():
"""The display-light switch RMW-replaces only the 'Light_*' entry in the
packed /mode/vs/0 options list (via laundry.replace_in_options), leaving
the other flags and the list order untouched."""
"""The display-light switch writes only the changed 'Light_*' token
(via laundry.option_write) -- confirmed on real hardware (issue #54)
that the device merges by prefix itself, so no read-modify-write of the
whole packed /mode/vs/0 options list is needed."""
desc = next(e for e in air_purifier.MODE.entities if e.key == 'display_light')
rep = {'x.com.samsung.da.options': [
'Comode_Off', 'Blooming_0', 'Light_On', 'OptionCode_60282',
@@ -81,24 +84,29 @@ def test_light_switch_write_contract():
assert desc.rep_fn(rep) is True
assert desc.write_fn('Off', rep) == (
['mode', 'vs', '0'],
{'x.com.samsung.da.options': [
'Comode_Off', 'Blooming_0', 'Light_Off', 'OptionCode_60282',
]},
{'x.com.samsung.da.options': ['Light_Off']},
)
def test_mode_tokens_are_read_only_diagnostics():
"""Comode_/Blooming_ tokens surface as raw diagnostic sensors rather than
a select/control -- their valid value ranges aren't confirmed yet (see
the air_purifier.py module docstring and the issue #56 follow-up)."""
def test_operating_mode_is_a_read_only_diagnostic():
"""Comode_* surfaces as a raw diagnostic sensor rather than a select/
control -- issue #56's five running-state dumps confirmed it reads 'Off'
regardless of the device's actual fan setting, ruling out the original
guess that it was the fan-speed selector; its real purpose is still
unconfirmed (see the air_purifier.py module docstring)."""
operating_mode = next(e for e in air_purifier.MODE.entities if e.key == 'operating_mode')
blooming = next(e for e in air_purifier.MODE.entities if e.key == 'blooming_level')
rep = {'x.com.samsung.da.options': ['Comode_Off', 'Blooming_6']}
rep = {'x.com.samsung.da.options': ['Comode_Off']}
assert operating_mode.rep_fn(rep) == 'Off'
assert blooming.rep_fn(rep) == '6'
assert not hasattr(operating_mode, 'write_fn')
def test_blooming_not_modeled():
"""Confirmed (issue #56) to have no corresponding SmartThings app
setting -- dropped entirely rather than kept as an unexplained
diagnostic."""
assert not any(e.key == 'blooming_level' for e in air_purifier.MODE.entities)
def test_airflow_vs_fallback_only_binds_without_generic():
"""/airflow/vs/0 is a match_fn fallback -- it must not bind when the
OCF-standard /airflow/0 is also present (both are on every dump seen)."""
+24
View File
@@ -215,6 +215,30 @@ def test_tp1x_rac_expected_state_keys_present():
assert key in state, key
def test_caww_tp2_model_resolves_via_model_fallback():
"""A-CAWW-TP2-20-COMMON (issue #52, System AC) reports no oneUiVersion
and no '_RAC_'/'_PRAC_' token -- resolved via the '-CAWW-' modelNum
fallback added for this device."""
reg, _ = _resolve('airconditioner_caww_tp2')
assert reg is not None and reg.name == 'airconditioner'
def test_caww_tp2_no_unbound_hrefs():
"""Every resource in the issue #52 dump binds or is ignored -- clears
the coverage-gap repair. Only new href beyond the existing RAC/PRAC
surface is /sac/installationinfo/vs/0 (opaque SAC installation topology,
ignored)."""
reg, resources = _resolve('airconditioner_caww_tp2')
unbound = []
discover(resources, reg.capabilities, reg.pattern_capabilities, log=unbound.append)
assert unbound == []
def test_caww_tp2_sac_installationinfo_is_ignored():
ignored_hrefs = {cap.href for cap in airconditioner.COVERAGE}
assert '/sac/installationinfo/vs/0' in ignored_hrefs
def test_mute_once_write_target():
write = airconditioner.MUTE_ONCE.entities[0].write_fn
assert write('On', {}) == (['option', 'muteonce', 'vs', '0'], {'muteonce': 'On'})
+31
View File
@@ -26,6 +26,37 @@ def test_common_caps_discover_on_dishwasher(dishwasher_resources):
assert 'power_switch' in keys
class TestMergeOptionsField:
"""merge_options_field() is the read side of issue #54's finding: a
write only needs to carry the changed token, so the coordinator uses
this to keep its optimistic cache entry complete (every sibling option
still present) without waiting on a real poll."""
def test_replaces_matching_prefix(self):
cached = ['DeviceType_0167', 'Course_16', 'GMT_04']
assert common.merge_options_field(cached, ['Course_1D']) == [
'DeviceType_0167', 'Course_1D', 'GMT_04',
]
def test_appends_when_prefix_absent(self):
cached = ['DeviceType_0167']
assert common.merge_options_field(cached, ['NaturalSteam_On']) == [
'DeviceType_0167', 'NaturalSteam_On',
]
def test_merges_multiple_tokens_independently(self):
cached = ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_0', 'GMT_04']
merged = common.merge_options_field(cached, ['SoftenerLevelCtrl_2'])
assert merged == ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_2', 'GMT_04']
def test_handles_missing_cache(self):
assert common.merge_options_field(None, ['Course_1D']) == ['Course_1D']
def test_ignores_malformed_tokens(self):
cached = ['Course_16']
assert common.merge_options_field(cached, ['nounderscore']) == ['Course_16']
# ---------------------------------------------------------------------------
# OCF-native / vendor '-vs' fallback pairs (power, kids-lock, remote control).
# ---------------------------------------------------------------------------
+137
View File
@@ -0,0 +1,137 @@
"""Tests for LocalThingsCoordinator.async_raw_write -- the debug-only
arbitrary-href write primitive backing the options-flow debug panel
(issue #54). It deliberately bypasses the remote-control block and every
write_fn/validate_fn, so this only exercises the primitive itself: it
POSTs exactly the caller's body, reads the href back, and validates its
inputs.
"""
from __future__ import annotations
from unittest.mock import AsyncMock
import cbor2
import pytest
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.localthings.const import (
CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
ENTRY_DATA = {
CONF_HOST: '10.0.0.199',
CONF_PORT: 49154,
CONF_LEAF_CERT_PEM: '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----',
CONF_LEAF_KEY_PEM: '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----',
}
class _FakeRawWriteSession:
"""Stand-in for DtlsCoapSession: records every POST verbatim and
answers the follow-up GET with a canned representation -- no real
DTLS/network involved."""
def __init__(self, post_code: int = 0x44, get_rep: dict | None = None):
self.post_calls: list[tuple[list[str], bytes]] = []
self.get_calls: list[list[str]] = []
self._post_code = post_code
self._get_rep = {} if get_rep is None else get_rep
def post(self, path_segs, payload, timeout=None):
self.post_calls.append((list(path_segs), payload))
return self._post_code, b''
def get(self, path_segs, timeout=None):
self.get_calls.append(list(path_segs))
return 0x45, cbor2.dumps(self._get_rep)
def pace(self):
pass
@pytest.fixture
def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
entry = MockConfigEntry(
domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_RAWWRITE-TEST',
)
entry.add_to_hass(hass)
coord = LocalThingsCoordinator(hass, entry)
# async_raw_write kicks a full refresh after the write; a real refresh
# would try to poll a session that doesn't exist for this unit test, so
# replace it with a no-op the same way the reference test setup does.
coord.async_request_refresh = AsyncMock()
return coord
async def test_raw_write_splits_href_and_posts_exact_body(coordinator) -> None:
fake = _FakeRawWriteSession(post_code=0x44, get_rep={'x.field': 'after'})
coordinator._session = fake
body = {'x.com.samsung.da.field': 'value', 'n': 1}
code, new_rep = await coordinator.async_raw_write('/course/vs/0', body)
assert len(fake.post_calls) == 1
posted_path, posted_bytes = fake.post_calls[0]
assert posted_path == ['course', 'vs', '0']
assert cbor2.loads(posted_bytes) == body
assert code == 0x44
assert new_rep == {'x.field': 'after'}
coordinator.async_request_refresh.assert_awaited_once()
async def test_raw_write_reads_href_back_for_ground_truth(coordinator) -> None:
fake = _FakeRawWriteSession(post_code=0x45, get_rep={'value': 'confirmed'})
coordinator._session = fake
code, new_rep = await coordinator.async_raw_write('/washer/vs/0', {'a': 1})
assert fake.get_calls == [['washer', 'vs', '0']]
assert code == 0x45
assert new_rep == {'value': 'confirmed'}
async def test_raw_write_returns_coap_code_from_post_not_get(coordinator) -> None:
"""The returned code reflects the write's own response, even though a
follow-up GET (which could carry a different code in principle) runs
right after it."""
fake = _FakeRawWriteSession(post_code=0x80, get_rep={'value': 1})
coordinator._session = fake
code, _ = await coordinator.async_raw_write('/test/vs/0', {'a': 1})
assert code == 0x80
async def test_raw_write_rejects_empty_dict(coordinator) -> None:
with pytest.raises(ServiceValidationError):
await coordinator.async_raw_write('/washer/vs/0', {})
async def test_raw_write_rejects_non_dict_payload(coordinator) -> None:
with pytest.raises(ServiceValidationError):
await coordinator.async_raw_write('/washer/vs/0', ['not', 'a', 'dict']) # type: ignore[arg-type]
async def test_raw_write_rejects_empty_href(coordinator) -> None:
with pytest.raises(ServiceValidationError):
await coordinator.async_raw_write('', {'a': 1})
async def test_raw_write_rejects_root_href(coordinator) -> None:
with pytest.raises(ServiceValidationError):
await coordinator.async_raw_write('/', {'a': 1})
async def test_raw_write_validation_errors_do_not_touch_the_session(coordinator) -> None:
"""A rejected call must fail before ever reaching the network -- no
session needs to be connected at all for validation to run."""
assert coordinator._session is None
with pytest.raises(ServiceValidationError):
await coordinator.async_raw_write('/washer/vs/0', {})
assert coordinator._session is None
coordinator.async_request_refresh.assert_not_awaited()
+94
View File
@@ -0,0 +1,94 @@
"""Tests for LocalThingsCoordinator.async_send_command's optimistic-apply
step, specifically for x.com.samsung.da.options[] writes (issue #54).
The write itself only needs to carry the single changed token -- confirmed
on real hardware, the device merges by prefix and evicts the stale token
itself. But observe.ObserveManager.apply() does a shallow {**cached, **rep}
field merge, so handing it that same minimal single-token body would
overwrite the *whole* cached options[] field, wiping every sibling option
(other courses/levels/toggles packed into the same array) until the next
real poll lands. async_send_command must pre-merge the token into the
cached array itself before applying it optimistically, while still POSTing
only the minimal body over the wire.
"""
from __future__ import annotations
from unittest.mock import AsyncMock
import cbor2
import pytest
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.localthings.const import (
CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capabilities import laundry
from custom_components.localthings.registry.discovery import BoundEntity
ENTRY_DATA = {
CONF_HOST: '10.0.0.199',
CONF_PORT: 49154,
CONF_LEAF_CERT_PEM: '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----',
CONF_LEAF_KEY_PEM: '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----',
}
class _FakeSendSession:
def __init__(self):
self.post_calls: list[tuple[list[str], bytes]] = []
def post(self, path_segs, payload, timeout=None):
self.post_calls.append((list(path_segs), payload))
return 0x44, b''
def pace(self):
pass
@pytest.fixture
def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
entry = MockConfigEntry(
domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_SENDCMD-TEST',
)
entry.add_to_hass(hass)
coord = LocalThingsCoordinator(hass, entry)
coord.async_request_refresh = AsyncMock()
coord._session = _FakeSendSession()
return coord
async def test_options_write_posts_only_the_changed_token(coordinator) -> None:
href = '/course/vs/0'
coordinator._observe.apply(href, {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
}, source='poll')
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
bound = BoundEntity(href=href, capability=None, desc=desc)
await coordinator.async_send_command(bound, '1D')
posted_path, posted_bytes = coordinator._session.post_calls[0]
assert posted_path == ['course', 'vs', '0']
assert cbor2.loads(posted_bytes) == {'x.com.samsung.da.options': ['Course_1D']}
async def test_options_write_optimistic_cache_keeps_sibling_tokens(coordinator) -> None:
"""The regression this guards: applying the minimal wire body straight
to the cache would wipe DeviceType_0167/GMT_04 until the next poll."""
href = '/course/vs/0'
coordinator._observe.apply(href, {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
}, source='poll')
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
bound = BoundEntity(href=href, capability=None, desc=desc)
await coordinator.async_send_command(bound, '1D')
cached = coordinator._cache.get(href)
assert cached['x.com.samsung.da.options'] == [
'DeviceType_0167', 'Course_1D', 'GMT_04',
]
+1 -3
View File
@@ -28,9 +28,7 @@ class TestCycleOptions:
rep = {'x.com.samsung.da.options': ['DeviceType_0001', 'Course_0E', 'GMT_04']}
path, body = desc.write_fn('90', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0001', 'Course_90', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_90']}
class TestDishwasherOptions:
+9 -3
View File
@@ -59,11 +59,17 @@ def test_power_watts_gated_for_dead_sentinel():
def test_course_bound_to_shared_course_vs_0():
"""Dryer course uses the shared /course/vs/0 cycle select with dryer_cycle
translations, consistent with washer/dishwasher."""
"""Dryer course uses the shared /course/vs/0 cycle select, with the
translation key built from the device's own course table (see
laundry.cycle_select) -- confirmed dryers report Table_03, matching
the shipped dryer_cycle_table_03 translations, consistent with
washer/dishwasher."""
assert dryer.DRYER_COURSE.href == '/course/vs/0'
desc = next(e for e in dryer.DRYER_COURSE.entities if e.key == 'cycle')
assert desc.translation_key == 'dryer_cycle'
assert callable(desc.translation_key)
table_03 = {'/st/dryercourse/vs/0': {'x.com.samsung.da.st.courseTable': 'Table_03'}}
assert desc.translation_key(table_03) == 'dryer_cycle_table_03'
assert desc.translation_key({}) is None
assert desc.options is laundry.cycle_options
rep = {'x.com.samsung.da.options': ['Course_16', 'GMT_02']}
assert desc.rep_fn(rep) == '16'
+36
View File
@@ -258,3 +258,39 @@ class TestRefrigeratorAiEnergyLevelFixtureCoverage:
assert unbound == []
state = flatten(bound, resources)
assert state['ai_energy_level'] == '1'
class TestSwitchOffIsNotInverted:
"""Regression: several fridge SwitchDesc write_fns used ``'On' if p else 'Off'``.
The switch platform passes the literal string ``'Off'`` on turn-off (and
``'On'`` on turn-on). ``'Off'`` is a non-empty, truthy string, so
``'On' if p else 'Off'`` always evaluated to ``'On'`` — turning the switch
OFF silently re-sent ``On`` and the switch could never be turned off. The
guard must compare ``p == 'On'`` (as the other capability files already do).
"""
# (switch descriptor, payload key it writes)
CASES = [
(fridge.ICEMAKER_NIGHTTIME.entities[0], 'ice.night.status'),
(fridge.STATUS_LOCK.entities[0], 'x.com.samsung.da.ado.devicecontrol'),
(fridge.STATUS_LOCK.entities[1], 'x.com.samsung.da.device.sound'),
(fridge.DEFROST_DELAY.entities[0], 'x.com.samsung.da.delayDefrost'),
(fridge.WELCOME_LIGHTING.entities[0], 'status'),
(fridge.CABINET_LIGHT.entities[1], 'light.dimming.status'),
(fridge.ICEMAKER_STATUS_FALLBACK.entities[0], 'x.com.samsung.da.iceMaker'),
]
def test_turning_off_sends_off(self):
for desc, key in self.CASES:
_segs, payload = desc.write_fn('Off', {})
assert payload[key] == 'Off', (
f"{desc.key}: OFF must send 'Off', got {payload[key]!r} (inverted)"
)
def test_turning_on_sends_on(self):
for desc, key in self.CASES:
_segs, payload = desc.write_fn('On', {})
assert payload[key] == 'On', (
f"{desc.key}: ON must send 'On', got {payload[key]!r}"
)
+18
View File
@@ -213,6 +213,24 @@ def test_registry_reproduces_golden_state_keys_for_tp2x_rac_20k():
)
def test_registry_reproduces_golden_state_keys_for_caww_tp2():
"""A-CAWW-TP2-20-COMMON (issue #52, System AC / multi-indoor-unit
commercial install) reports no oneUiVersion and no '_RAC_'/'_PRAC_'
token; resolved via the '-CAWW-' modelNum fallback in
for_device_by_model. Otherwise binds cleanly against the existing
airconditioner registry -- same TP1X/TP2X-class resource surface, plus
one SAC-only installation-topology resource (ignored)."""
from tests.conftest import _load_device
resources = _load_device('airconditioner_caww_tp2')
golden = json.loads((GOLDEN / 'airconditioner_caww_tp2.json').read_text())
state_keys = _new_state_keys('airconditioner_caww_tp2', resources)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_registry_reproduces_golden_state_keys_for_tp1x_rac():
"""TP1X_DA-AC-RAC-01001_0000 (issue #38) -- fuller RAC board with display
light, self-check, mute-once, and a current-limit setting."""
+149 -4
View File
@@ -35,6 +35,103 @@ class TestCourseHelpers:
assert laundry.option_value(opts, 'Missing') is None
class TestCourseCodesFromSupportedOptions:
"""cycle_options()'s fallback for boards that populate
/wm/editcourse/vs/0 without ever filling in editCourseList itself
(issue #1) -- derives the course list from /course/vs/0's own
supportedOptions instead."""
# Real dump from issue #1 (DA_WM_TP1_21_COMMON, WW5000C): a 1-hex-nibble
# header followed by 14 self-indexed 7-byte-per-course records. '1C' (the
# first record) is confirmed as "Eco 40-60" both by this device's own
# currently-selected course matching the SmartThings app screenshot's
# checked item, and by six other independent devices' already-shipped
# translations agreeing on the same code -> name mapping.
_REAL_SUPPORTED_OPTIONS = (
'31C8410923FA67F1B847E923FA67F25843E933FA57F20857E943FA67F'
'088000913FA67F7485209204A5208780009000A00006841E930FA30F'
'7F841E920FA30F65841E943FA57F8F8102923FA57F96841E920FA37F'
'34841E923FA67FA0811E933FA33F'
)
def test_derives_codes_when_edit_course_list_is_empty(self):
resources = {
'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': ''},
'/course/vs/0': {
'x.com.samsung.da.options': ['Course_1C'],
'x.com.samsung.da.supportedOptions': [self._REAL_SUPPORTED_OPTIONS],
},
}
assert laundry.cycle_options(resources) == [
'1C', '1B', '25', '20', '08', '74', '87', '06',
'7F', '65', '8F', '96', '34', 'A0',
]
def test_edit_course_list_still_takes_priority(self):
"""A live editCourseList wins even with supportedOptions present --
no reason to prefer a derived list over the authoritative one."""
resources = {
'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_651C'},
'/course/vs/0': {
'x.com.samsung.da.options': ['Course_1C'],
'x.com.samsung.da.supportedOptions': [self._REAL_SUPPORTED_OPTIONS],
},
}
assert laundry.cycle_options(resources) == ['65', '1C']
def test_rejects_a_table_missing_the_current_course(self):
"""The device's own currently-selected course must be a member of
its derived list -- a mismatch means the guess is wrong, not that
the device selected something outside its own supported set."""
resources = {
'/course/vs/0': {
'x.com.samsung.da.options': ['Course_FF'],
'x.com.samsung.da.supportedOptions': [self._REAL_SUPPORTED_OPTIONS],
},
}
assert laundry.cycle_options(resources) == []
def test_smallest_passing_split_wins_over_its_own_multiples(self):
"""K=2 and K=4 both trivially re-pass the same two checks here --
each is just a sparser sampling of the true, smaller K=1 table (its
first bytes are a subset of K=1's, so uniqueness and "contains the
current course" carry over for free) -- but K=1 is the real, most
specific table and must be the one returned."""
resources = {
'/course/vs/0': {
'x.com.samsung.da.options': ['Course_AA'],
'x.com.samsung.da.supportedOptions': ['0AABBCCDD'],
},
}
assert laundry.cycle_options(resources) == ['AA', 'BB', 'CC', 'DD']
def test_empty_without_supported_options_or_course_href(self):
assert laundry.cycle_options({}) == []
assert laundry.cycle_options({'/course/vs/0': {}}) == []
def test_smallest_wins_even_when_a_larger_pass_is_not_a_multiple(self):
"""Real dishwasher dump (K=7, 10 courses): K=10, 14, and 35 also
pass both checks here, and none of them are multiples of 7 --
position 0 lands on the same real course code ('0e') regardless of
K, which alone satisfies the current-course guard for several
unrelated splits. Smallest-K-wins is a heuristic that matches every
real dump checked so far, not a proven guarantee -- see
_course_codes_from_supported_options's docstring."""
resources = {
'/course/vs/0': {
'x.com.samsung.da.options': ['Course_0E'],
'x.com.samsung.da.supportedOptions': [
'30E5434B102D102835034B002D002845034B002D002805034B000D000'
'865034B002D002075000B000D000905000B000D0008D5034B002D0028'
'E5034B000D0008F5034B000D000'
],
},
}
assert laundry.cycle_options(resources) == [
'0E', '83', '84', '80', '86', '07', '90', '8D', '8E', '8F',
]
class TestCycleSelect:
def test_builds_labelled_cycle_select(self):
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='mdi:tumble-dryer')
@@ -56,20 +153,68 @@ class TestCycleSelect:
live = {'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_16'}}
assert desc.exists_fn({}, live) is True
def test_cycle_write_rmw_on_options(self):
def test_cycle_write_is_single_token(self):
"""Confirmed on real hardware (issue #54): the device merges a
single-token options[] write by prefix itself, so the write only
needs to carry the changed token, not the whole rewritten array."""
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04']}
path, body = desc.write_fn('1D', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_1D']}
def test_cycle_write_noop_without_options(self):
desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
assert desc.write_fn('1D', {}) is None
class TestCycleSelectTableGating:
"""translation_key becomes a resolver, not a plain string, once
table_href is given -- washer/dryer's real call sites (issue: course
codes aren't guaranteed consistent across board generations sharing
the same /course/vs/0 contract; FlexWash's older board reports a
different course table than every device the shipped translations
were confirmed against). The resolved key is built from whatever table
the device actually reports -- a table with no strings.json entries
yet just falls through Home Assistant's own missing-translation
handling to raw-code display, the same as any individual untranslated
code within an existing table."""
def _desc(self):
return laundry.cycle_select(
translation_key='washer_cycle', icon='x',
table_href='/st/washercourse/vs/0',
)
def test_static_string_when_no_table_href_given(self):
"""dishwasher's call site -- no equivalent table-id resource in any
dump seen, no evidence of the same cross-board risk -- keeps the
plain static key unconditionally."""
desc = laundry.cycle_select(translation_key='dishwasher_cycle', icon='x')
assert desc.translation_key == 'dishwasher_cycle'
def test_resolved_key_is_built_from_the_reported_table(self):
desc = self._desc()
resources = {'/st/washercourse/vs/0': {'x.com.samsung.da.st.courseTable': 'Table_02'}}
assert callable(desc.translation_key)
assert desc.translation_key(resources) == 'washer_cycle_table_02'
def test_resolved_key_reflects_an_unbuilt_table_too(self):
"""No gating against a hardcoded 'known good' table -- a table we
haven't shipped translations for yet still gets a key built for
it, just one strings.json has nothing under (raw-code display)."""
desc = self._desc()
resources = {'/st/washercourse/vs/0': {'x.com.samsung.da.st.courseTable': 'Table_00'}}
assert desc.translation_key(resources) == 'washer_cycle_table_00'
def test_resolves_to_none_when_table_id_is_unknown(self):
"""No href, or an empty rep, gets no translation_key at all --
there's nothing to build a key from."""
desc = self._desc()
assert desc.translation_key({}) is None
assert desc.translation_key({'/st/washercourse/vs/0': {}}) is None
class TestBuzzerSound:
def test_href(self):
assert laundry.BUZZER_SOUND.href == '/buzzersound/vs/0'
+14 -13
View File
@@ -137,7 +137,10 @@ def test_oven_mode_rejects_unknown():
# ---------------------------------------------------------------------------
# OVEN_MODE options-array RMW (lamp, sound, fast_preheat, natural_steam)
# OVEN_MODE options-array writes (lamp, sound, fast_preheat, natural_steam).
# Confirmed on real hardware (issue #54): a write only needs to carry the
# changed token -- the device matches by prefix, evicts the stale token, and
# merges the result into the array itself. No read-modify-write needed.
# ---------------------------------------------------------------------------
def _mode_rep(*extra_opts):
@@ -146,12 +149,11 @@ def _mode_rep(*extra_opts):
]}
def test_lamp_write_replaces_slot():
def test_lamp_write_is_single_token():
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'lamp')
path, body = desc.write_fn('On', _mode_rep())
opts = body['x.com.samsung.da.options']
assert 'UpperLamp_On' in opts
assert 'UpperLamp_Off' not in opts
assert path == ['mode', 'vs', '0']
assert body == {'x.com.samsung.da.options': ['UpperLamp_On']}
def test_lamp_write_requires_existing_options():
@@ -159,20 +161,19 @@ def test_lamp_write_requires_existing_options():
assert desc.write_fn('On', {}) is None
def test_sound_write_preserves_other_options():
def test_sound_write_is_single_token():
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'sound')
path, body = desc.write_fn('Off', _mode_rep())
opts = body['x.com.samsung.da.options']
assert 'Sound_Off' in opts
assert 'UpperLamp_Off' in opts # other slot unchanged
assert body == {'x.com.samsung.da.options': ['Sound_Off']}
def test_natural_steam_appended_if_absent():
"""NaturalSteam slot is absent until first write — write_fn must append it."""
def test_natural_steam_write_is_single_token():
"""NaturalSteam's slot may be absent from the live rep until first
write -- the single-token write covers both the insert and replace case
identically, since the device merges by prefix either way."""
desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'natural_steam')
path, body = desc.write_fn('On', _mode_rep()) # no NaturalSteam_* in rep
opts = body['x.com.samsung.da.options']
assert any(o.startswith('NaturalSteam_') for o in opts)
assert body == {'x.com.samsung.da.options': ['NaturalSteam_On']}
# ---------------------------------------------------------------------------
+2 -3
View File
@@ -1,9 +1,8 @@
"""Tests for select-option display casing (custom_components/localthings/select.py)."""
from custom_components.localthings.registry.entities import SelectDesc
from custom_components.localthings.select import _display
_UNTRANSLATED = SelectDesc(key='x', options=())
_TRANSLATED = SelectDesc(key='y', options=(), translation_key='door_alert')
_UNTRANSLATED = None
_TRANSLATED = 'door_alert'
def test_display_titlecases_a_fully_lowercase_device_native_token():
+17
View File
@@ -57,3 +57,20 @@ def test_callable_options_empty_result():
desc = SelectDesc(key='cycle', options=lambda resources: [])
entity = _make_select(desc, '/x/vs/0', {})
assert entity.options == []
def test_callable_translation_key_reresolves_live_not_once_at_construction():
"""A callable translation_key (laundry.cycle_select's table-id-gated
resolver) must be re-evaluated against current coordinator data on
every access, not baked in once at __init__ -- discovery can run while
a sibling resource (e.g. /st/washercourse/vs/0) is still an empty stub
(see entity.py's _is_included docstring), and a one-time resolution
would permanently show untranslated codes even after a later poll
populates the real value."""
desc = SelectDesc(key='cycle', translation_key=lambda resources: resources.get('key'))
resources = {'key': None}
entity = _make_select(desc, '/x/vs/0', resources)
assert entity.translation_key is None
resources['key'] = 'washer_cycle_table_02'
assert entity.translation_key == 'washer_cycle_table_02'
+30 -14
View File
@@ -61,8 +61,19 @@ class TestWasherCourse:
assert washer.WASHER_COURSE.href == '/course/vs/0'
def test_translation_key(self):
"""Table-scoped (issue: course codes aren't guaranteed consistent
across board generations sharing /course/vs/0 -- FlexWash's older
board reports Table_00, not the Table_02 every washer_cycle_table_02
name was confirmed against) -- see laundry.cycle_select. The key is
built from whatever table the device reports, not gated against a
hardcoded 'known good' value."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
assert desc.translation_key == 'washer_cycle'
assert callable(desc.translation_key)
table_02 = {'/st/washercourse/vs/0': {'x.com.samsung.da.st.courseTable': 'Table_02'}}
assert desc.translation_key(table_02) == 'washer_cycle_table_02'
table_00 = {'/st/washercourse/vs/0': {'x.com.samsung.da.st.courseTable': 'Table_00'}}
assert desc.translation_key(table_00) == 'washer_cycle_table_00'
assert desc.translation_key({}) is None
def test_reads_raw_course_code_from_options_array(self):
"""rep_fn returns the raw device code; display names come from
@@ -91,13 +102,14 @@ class TestWasherCourse:
assert desc.exists_fn({}, live) is True
def test_cycle_write(self):
"""Confirmed on real hardware (issue #54): the write only needs to
carry the changed token -- the device matches by prefix, evicts the
stale token, and merges the result into the array itself."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1C', 'GMT_04']}
path, body = desc.write_fn('1D', rep)
assert path == ['course', 'vs', '0']
assert body == {
'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
}
assert body == {'x.com.samsung.da.options': ['Course_1D']}
class TestDrumClean:
@@ -219,20 +231,22 @@ class TestDetergentSoftenerDosing:
def test_quantity_write(self):
"""The UI selects a padded supported code ('01'); the write posts the
un-padded device code ('1'), mirroring how the device reports it."""
un-padded device code ('1'), mirroring how the device reports it.
Confirmed on real hardware (issue #54): the write only needs to carry
the changed token -- the device matches by prefix, evicts the stale
token, and merges the result into the array itself. No need to read
the current array back and rewrite it whole."""
rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
path, body = self._desc('detergent_quantity').write_fn('01', rep)
assert path == ['course', 'vs', '0']
assert 'DetergentLevelCtrl_1' in body['x.com.samsung.da.options']
assert 'DetergentLevelCtrl_3' not in body['x.com.samsung.da.options']
# untouched siblings survive the read-modify-write
assert 'SoftenerLevelCtrl_3' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': ['DetergentLevelCtrl_1']}
def test_hardness_write(self):
rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
path, body = self._desc('softener_concentration').write_fn('03', rep)
assert path == ['course', 'vs', '0']
assert 'SoftenerLevel2Ctrl_3' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': ['SoftenerLevel2Ctrl_3']}
def test_low_reservoir_off_when_alarm_off(self):
rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
@@ -294,18 +308,20 @@ class TestWashOptionToggles:
"""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."""
(including 'Off') is truthy.
The write body carries only the changed token (issue #54: confirmed
the device merges by prefix itself), not the whole options array."""
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('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']
assert body == {'x.com.samsung.da.options': [f'{prefix}_On']}
rep = {'x.com.samsung.da.options': [f'{prefix}_On']}
path, body = self._desc(key).write_fn('Off', rep)
assert f'{prefix}_Off' in body['x.com.samsung.da.options']
assert body == {'x.com.samsung.da.options': [f'{prefix}_Off']}
assert f'{prefix}_On' not in body['x.com.samsung.da.options']
def test_write_rejects_non_on_off_payload(self):