Compare commits

...
29 Commits
Author SHA1 Message Date
Marc Billow 4cc073c88a Merge pull request #12 from mbillow/claude/readme-cleanup-b3ib70
docs: simplify and update README
2026-07-18 23:34:43 -05:00
Marc Billow 91013ece60 fix: translate machine_state sensor values; bump to 0.4.1
machine_state was rendering the raw lowercase OCF values (idle/active/
pause) untranslated in the UI. Add device_class=enum, options, and a
translation_key matching the existing pattern used by ice_making_status
and connection_mode, in both the shared operational.py capability and
oven.py's separate machine_state sensor.

Bump patch version for this fix plus the ice_type 'off' translation and
iot_class correction earlier on this branch.
2026-07-19 04:33:23 +00:00
Marc Billow 795763f3ca fix: add missing ice_type 'off' state translation
The refrigerator fixture's x.com.samsung.da.iceType.supported list
includes "Off" alongside the whiskey_iceball_* values, but strings.json
only translated the whiskey_iceball options, leaving "Off" to render
untranslated in the ice-type select.
2026-07-19 04:18:59 +00:00
Marc Billow a7ed230569 fix: correct iot_class to local_push
Coordinator prefers CoAP OBSERVE push notifications (observe.py) over
polling when the device supports it, falling back to polling only when
observe mode isn't available or drops.
2026-07-19 04:11:01 +00:00
Marc Billow 9fee5b5ec0 docs: simplify smartthings-local intro blurb 2026-07-19 04:07:52 +00:00
Marc Billow f24d6a0ae6 docs: simplify and update README
Add missing washer support to the appliance table, fix stale test-count
and repo-layout claims, trim protocol-internals jargon that belongs to
the smartthings-local library rather than this integration, and point
"Adding a new appliance type" at HA's own diagnostics download instead
of a gitignored local script.
2026-07-19 04:05:37 +00:00
Marc Billow 29b145ccdd Merge pull request #11 from mbillow/claude/issue-9-investigation-mdnwcd
chore: bump manifest version to 0.4.0
2026-07-18 13:27:57 -05:00
Marc Billow 5fbf58af30 chore: bump manifest version to 0.4.0 2026-07-18 18:26:58 +00:00
Marc Billow 9ae59889d4 Merge pull request #10 from mbillow/claude/issue-9-investigation-mdnwcd
fix(washer): stale idle progress%, add detergent/softener dosing entities (#9)
2026-07-18 13:25:33 -05:00
Marc Billow 35f76ce2df feat(washer): add label translations for detergent/softener dosing selects
Assumes the LevelCtrl code scheme is None/Low/Medium/High (00-03) on both
dispensers -- code 00 has no on-screen equivalent in the app's 3-choice
Faible/Moyen/Élevé picker, assumed to be what "Activation" off collapses
to -- and Level2Ctrl is Soft/Medium/Hard for detergent water hardness,
1x/2x/3x for softener concentration. detergent_quantity and
softener_quantity share one translation_key (same vocabulary), same
pattern as fridge.py's shared 'brightness_level' key.

Not cross-device verified: only one dump + screenshot set (issue #9) to go
on, and the softener concentration reading doesn't cleanly match its
screenshot (assumed to be a setting changed between dump and screenshots,
not a different code scheme -- see the comment in washer.py).
2026-07-18 18:24:10 +00:00
Marc Billow 2a6bbc9dfb fix(washer): stale idle progress%, add detergent/softener dosing entities (#9)
progress_percentage lacked the active-state gate already applied to
progress/cycle_active/finish_time, so it kept showing a stale device value
(e.g. 1%) while idle -- now zeroed the same way. Shared by dryer/dishwasher/
oven via operational.py's OPERATIONAL_STATE.

Also exposes detergent/softener auto-dispense quantity, water hardness/
concentration, and low-reservoir alarms from /course/vs/0's options array,
using the same decode/RMW helpers already used for course selection and
drum-clean tracking. Gated by exists_fn so washer models without these
fields (e.g. the existing test fixture) are unaffected.

Water consumption is unaffected -- common.WATER_METER is already wired
into the washer registry; this reporter's device just doesn't expose
/water/consumption/vs/0.
2026-07-18 18:15:41 +00:00
Marc Billow aee19fd953 Merge branch 'fix-washer-fridge-oven-issues'
* fix-washer-fridge-oven-issues:
  chore: bump manifest version to 0.3.0
  fix(washer,fridge,oven): correct energy/temperature-unit bugs from issues #6/#7
2026-07-15 16:02:05 -05:00
Marc Billow 37bd063c0a chore: bump manifest version to 0.3.0 2026-07-15 15:59:24 -05:00
Marc Billow e875266405 fix(washer,fridge,oven): correct energy/temperature-unit bugs from issues #6/#7
Washer (#6): instantaneousPower is a dead sentinel ('-500') on every
TP1-class washer dump collected so far, and cumulativePower is absent
outright on at least one model. WASHER_ENERGY_METER now hides both
sensors instead of showing a misleading "0 W"/perpetual "unavailable".

Fridge (#7): temperature sensors/setpoints hardcoded '°F', ignoring the
unit each device actually reports per-reading -- fixed via a new
unit_fn hook read live from the resource. Also corrects
DEFROST_BLOCK_STATUS's polarity (DEFROST_BLOCK_ON means actively
defrosting, not "blocked", confirmed against live dumps) and adds
REFRIGERATION_FALLBACK for /refrigeration/0, closing the last unbound
href surfaced by issue #7's diagnostic dump.

Oven: applies the same live-unit-reading fix defensively to
OVEN_SETPOINT, which shares the same aggregate resource shape.
2026-07-15 11:41:21 -05:00
Marc Billow 9b48ca0e82 Merge pull request #5 from mbillow/manifest-version-bump
chore: bump manifest version to 0.2.1
2026-07-11 20:27:19 -05:00
Marc Billow e66f4fd2ff chore: bump manifest version to 0.2.1
Was left at 0.1.0 across the last two releases. Also de-hardcode the
diagnostics test's expected version so this doesn't happen again --
it now reads manifest.json directly instead of a copy-pasted literal.
2026-07-11 20:26:52 -05:00
Marc Billow bf3b3a8bbd Merge pull request #4 from mbillow/washer-drum-clean
feat(washer): expose Drum Clean+ maintenance status
2026-07-11 20:20:12 -05:00
Marc Billow dd541b19a1 feat(washer): expose Drum Clean+ maintenance status
Adds cycles-until-due and last-cleaned sensors decoded from the same
options[] array the course selector already reads (DrumCleanProposal_N -
WashingTimes_N for cycles remaining, DrumCleanLog_<iso> for last-cleaned),
verified byte-for-byte against a live app screenshot ("Potreba cistenia po
37 cykloch" / "Naposledy cistene pred 10 dnami").
2026-07-11 20:18:28 -05:00
Marc Billow cf22f3b3fe Merge pull request #3 from mbillow/worktree-washer-device-type
feat(washer): add washer device type support
2026-07-11 20:03:55 -05:00
Marc Billow 211cf74296 feat(washer,dishwasher): live per-device cycle selector, localized names
Course selection is now a writable select entity sourced from each
device's own x.com.samsung.da.editCourseList (via a new options-as-callable
form on SelectDesc, reading the coordinator's full resource snapshot
instead of just the entity's own href), rather than a hardcoded course
table baked into Python. A MostUsed_ field on the same resource was
considered as a fallback but rejected after byte-level analysis showed it
doesn't reliably encode a course list. When a device never populates
editCourseList, the selector isn't created at all (exists_fn now takes
(rep, resources) to check a sibling href, matching the existing
match_fn(rep, resources) pattern).

Display names moved out of Python into strings.json/translations under
entity.select.{washer,dishwasher}_cycle.state.*, matching this
integration's existing translation_key convention (fridge.py's ice_type,
flex_zone_mode, etc.) instead of hardcoding English names in code.
2026-07-11 20:00:12 -05:00
Marc Billow 29b9e36699 fix(washer): ignore remaining unbound hrefs (cycleinterface, drlc, operational/state OCF-native)
discover() flagged /cycleinterface/vs/0, /drlc/0, and /operational/state/0
as unbound on real washer dumps, which would surface a spurious device
coverage gap for washer owners. All three are noise: cycleinterface is
empty on every dump seen, drlc/0 is an OCF-native duplicate of the
already-ignored /drlc/vs/0, and operational/state/0 is a read-only
OCF-native duplicate of /operational/state/vs/0 (already modeled by the
richer, write-capable operational.OPERATIONAL_STATE).
2026-07-11 13:30:00 -05:00
Marc Billow f4eca9a834 test(washer): add washer fixture and golden-regression coverage 2026-07-11 13:01:01 -05:00
Marc Billow 22c5295d18 feat(washer): wire model-number fallback into discovery and config flow 2026-07-11 12:53:57 -05:00
Marc Billow 2852a21c4a feat(washer): add model-number-based device-type detection fallback 2026-07-11 12:49:08 -05:00
Marc Billow 779c2f9881 feat(washer): add washer device registry 2026-07-11 12:45:05 -05:00
Marc Billow 1e70380ae3 chore(washer): declare washer's redundant/empty hrefs as ignored 2026-07-11 12:42:23 -05:00
Marc Billow 219cb70b77 fix(operational): support delayEndTime as an alternate delay-start field 2026-07-11 12:38:17 -05:00
Marc Billow 1b5e39ba46 feat(washer): prefer OCF-native power/kidslock/remotectrl hrefs, fall back to -vs 2026-07-11 12:35:47 -05:00
Marc Billow 2868938d37 feat(washer): add washer-specific capabilities 2026-07-11 12:31:20 -05:00
37 changed files with 2284 additions and 131 deletions
+45 -52
View File
@@ -2,31 +2,27 @@
**A native Home Assistant custom integration for local control of newer-generation Samsung connected appliances.** No cloud round-trip. Add a device through HA's normal *Settings > Devices & Services* flow and it talks CoAP-over-DTLS straight to the appliance on your LAN.
> ### Where things live
>
> This project split into two repos partway through development.
>
> - **[`smartthings-local`](https://github.com/QuiteYellow/SmartThings-Local)** (PyPI package): the reusable protocol layer. DTLS session handling, CoAP wire encoding, Block2 reads, bounded retry/retransmit, inter-request rate limiting, cert-chain validation. No HA dependency; usable from any Python project.
> - **This repo**: the Home Assistant integration built on top of it. Config flow, a per-device-type capability registry, the polling coordinator, and all the HA entity classes.
>
> `custom_components/localthings/manifest.json` pulls in `smartthings-local` from PyPI like any other HA integration dependency.
This integration uses the [`smartthings-local`](https://github.com/QuiteYellow/SmartThings-Local) library to handle the low-level DTLS/CoAP communication with devices.
### What you get
Adding a device just needs a host IP and your CA credentials in the UI. The integration reads the appliance's `oneUiVersion` and picks the matching capability registry (dryer, oven, dishwasher, refrigerator) on its own, so there's no per-model descriptor to write for a new unit of a type that's already supported.
Adding a device just needs a host IP and your CA credentials in the UI. The integration reads the appliance's identity and picks the matching capability registry on its own, so there's no per-model descriptor to write for a new unit of a type that's already supported.
Credential setup is one-time. The first device you add asks for the AC14K_M CA cert and key (see Part 2); every device after that reuses the same stored CA and only asks for the host IP, minting its own per-device leaf cert automatically.
Credential setup is one-time. The first device you add asks for a CA certificate and key (see Part 2); every device after that reuses the same stored CA and only asks for the host IP, minting its own per-device leaf cert automatically.
Your state stays on your LAN: HA talks to the appliance over a direct DTLS session, and Samsung's cloud sees nothing from this integration. (The appliance itself still maintains its own connection to Samsung; that's firmware behavior on the device side, not something this integration controls.)
### Supported appliance types
| Type | Registry | Notable capabilities |
|---|---|---|
| Dryer | `by_type/dryer.py` | Power, kids lock, remote control, alarms, energy meter, operational state, door LED, sound mode, dryer settings/course, job-beginning status, diagnosis, firmware-update sensor |
| Oven | `by_type/oven.py` | Power, kids lock, remote control, alarms, cavity state, setpoint, mode, operational state, door, connectivity, firmware-update sensor |
| Dishwasher | `by_type/dishwasher.py` | Power, kids lock, remote control, alarms, energy + water meters, water filter, operational state, cycle options/settings, door LED, sound mode/volume, firmware-update sensor |
| Refrigerator | `by_type/refrigerator.py` | Power, kids lock, remote control, alarms, energy meter, water filter, status lock, door alert, icemaker (nighttime + generic per-compartment), flex zone, refrigeration mode, autofill, welcome/cabinet lighting, Sabbath mode, beverage zone, plus pattern-matched per-compartment temperature/setpoint/icemaker/door capabilities for multi-cavity fridges, firmware-update sensor |
| Type | Registry |
|---|---|
| Dryer | `by_type/dryer.py` |
| Oven | `by_type/oven.py` |
| Dishwasher | `by_type/dishwasher.py` |
| Refrigerator | `by_type/refrigerator.py` |
| Washer | `by_type/washer.py` |
Each registry composes shared and family-specific `Capability` objects from `registry/capabilities/`; those modules document the individual resources/entities in more depth than a README table can stay current with.
Other Tizen RT / DAWIT-family appliances almost certainly speak the same protocol underneath, since the auth path and CoAP primitives are shared across the fleet. Adding a new type means writing a new `by_type/<name>.py` registry file; it doesn't require reverse-engineering the protocol again. See **Adding a new appliance type** below.
@@ -46,16 +42,10 @@ nmap -Pn -sU -p 49152-49160 "$APPLIANCE_IP"
## Part 2: One-time setup, get the AC14K_M CA credentials
The config flow (Part 3) needs a **CA certificate and CA private key** to mint each device's leaf cert itself. Specifically, it needs the `AC14K_M` intermediate CA: a cert chain that's been public for years and still ships in current Samsung firmware trust stores. It's required because every Samsung Tizen/RT-OCF appliance's factory ACL grants full CRUDN access (`perm=31` on `href=*`) to whatever identity is chained to that CA, so a cert signed by it is the one thing that lets HA talk to your appliance without Samsung's cloud in the loop. HA doesn't need the *device's* original cert or key, only something `AC14K_M` has signed, and it mints that itself once you give it the CA.
The config flow (Part 3) needs a **CA certificate and CA private key** to mint each device's leaf cert itself. Specifically, it needs the `AC14K_M` intermediate CA — a cert chain that's been public for years and still ships in current Samsung firmware trust stores. Every Samsung Tizen/RT-OCF appliance trusts identities chained to that CA with full access by default, so a cert signed by it is what lets HA talk to your appliance without Samsung's cloud in the loop. HA doesn't need the *device's* original cert or key, only something `AC14K_M` has signed, and it mints that itself once you give it the CA.
This repo doesn't include the needed CA bundle. For an example of how to obtain it, including fetching the AC14K_M cert and key and verifying they pair, see the `smartthings-local` protocol project's [`setup_cert.py`](https://github.com/QuiteYellow/SmartThings-Local/blob/main/setup_cert.py). However you obtain the CA cert and key, paste their PEM contents into the HA config flow's "CA Certificate (PEM)" and "CA Private Key (PEM)" fields in Part 3. You only need to do this once, since every appliance you add afterward reuses the same stored CA.
### Why this works
- Every Samsung Tizen/RT-OCF appliance has a factory-baked ACE in `/oic/sec/acl` granting the AC14K_M-chained identity `perm=31` on `href=*`.
- TizenRT iotivity derives the peer ID via `memmem(subject_dn, "uuid:")`, which is RDN-agnostic, so a cert with the UUID in any RDN authenticates the same way.
- You don't need the original keyholder's private key. The config flow mints its own key and has `AC14K_M` sign the leaf: different key, same identity, same access.
---
## Part 3: Add the integration in Home Assistant
@@ -80,7 +70,7 @@ docker compose up -d --build
docker compose logs -f
```
The `Dockerfile` builds on the official `home-assistant/home-assistant:stable` image and pre-installs `smartthings-local`, so the dependency is present at container start instead of depending on HA's own runtime pip-install step (which needs outbound network access at exactly the moment the integration loads, and repeats on every container recreate). Re-run with `--build` whenever the pinned `smartthings-local` version changes.
The `Dockerfile` builds on the official `home-assistant/home-assistant:stable` image and pre-installs `smartthings-local`, so the dependency is present at container start instead of depending on HA's own runtime pip-install step. Re-run with `--build` whenever the pinned `smartthings-local` version changes.
`docker-compose.yml` sets `network_mode: host`, which is required since DTLS is UDP and won't traverse Docker's bridge NAT to reach LAN appliances, and bind-mounts `custom_components/localthings/` read-only into `ha_config/custom_components/`. Bump `custom_components.localthings` to `debug` in `ha_config/configuration.yaml` for verbose protocol logging.
@@ -93,7 +83,7 @@ python3 -m venv .venv
.venv/bin/pytest tests/ -q
```
`requirements-dev.txt` pins `smartthings-local` the same way `manifest.json` does, so tests exercise the real published protocol layer rather than a vendored copy.
A large suite covering registry composition, discovery, entity descriptors, and golden-file regression against captured device dumps. `requirements-dev.txt` pins `smartthings-local` the same way `manifest.json` does, so tests exercise the real published protocol layer rather than a vendored copy.
---
@@ -101,29 +91,32 @@ python3 -m venv .venv
```
custom_components/localthings/
manifest.json Requirements (incl. the smartthings-local PyPI dep), version, domain
__init__.py async_setup_entry / async_unload_entry
config_flow.py UUID fetch, leaf cert minting, port probing, config entry creation
coordinator.py DataUpdateCoordinator: polling, stale-state fallback, write dispatch
const.py Domain, config keys, probe ports
entity.py Base entity wiring capability registry -> HA entity
manifest.json Requirements (incl. the smartthings-local PyPI dep), version, domain
__init__.py async_setup_entry / async_unload_entry
config_flow.py UUID fetch, leaf cert minting, port probing, config entry creation
coordinator.py Polling + push update coordination, stale-state fallback, write dispatch
observe.py CoAP OBSERVE (push-mode) support layered on the coordinator
diagnostics.py Redacted diagnostics download (device state + coverage metadata)
const.py Domain, config keys, probe ports
entity.py Base entity wiring capability registry -> HA entity
sensor.py / binary_sensor.py / switch.py / number.py / select.py / button.py / time.py
One module per HA platform
One module per HA platform
strings.json / translations/ Config-flow copy + entity state translations
registry/
batch.py /device/0 batch response parsing
capability.py Capability dataclass (href, entities, transforms)
entities.py Per-platform entity descriptor dataclasses
discovery.py Binds a device's live resources to registered capabilities
adapter.py Flattens bound entities into HA-ready state
identity.py Reads device identity (serial, oneUiVersion) for type detection
capabilities/ Shared + per-family Capability definitions (common, dryer, oven,
dishwasher, fridge, laundry, operational)
by_type/ One DeviceRegistry per appliance type, composed from capabilities/
tests/ 80+ tests: registry composition, discovery, entity descriptors,
golden-file regression against captured device dumps
requirements-dev.txt Test deps, including the smartthings-local package
docker-compose.yml / ha_config/ Local HA dev environment
registry.py Builds the global capability registry, validates href collisions
capability.py Capability dataclass (href, entities, transforms)
entities.py Per-platform entity descriptor dataclasses
discovery.py Binds a device's live resources to registered capabilities
adapter.py Flattens bound entities into HA-ready state
identity.py Reads device identity for type detection
redact.py Strips account/identity data before diagnostics leave HA
capabilities/ Shared + per-family Capability definitions (common, dryer, oven,
dishwasher, fridge, washer, laundry, operational, ignored)
by_type/ One DeviceRegistry per appliance type, composed from capabilities/
tests/ Registry composition, discovery, entity descriptors, coordinator/observe
behavior, and golden-file regression against captured device dumps
requirements-dev.txt Test deps, including the smartthings-local package
docker-compose.yml / ha_config/ Local HA dev environment
```
---
@@ -141,21 +134,21 @@ support for hardware the maintainers don't have.
## Adding a new appliance type
1. Capture the appliance's `/device/0` response to see what resources/fields it exposes. An authenticated `DtlsCoapSession` from `smartthings_local.protocol.dtls_session` GET is enough; `local-tools/probe_device.py` wraps this.
1. Get a capture of the appliance's `/device/0` response. The easiest way: add the device to HA (type detection failing is fine) and pull its Diagnostics download from Settings > Devices & Services > the device > the menu > Download diagnostics — it already contains a redacted dump of the device's resources.
2. Reuse existing `Capability` objects from `registry/capabilities/` wherever the resource matches one already declared. Most `common.py` capabilities (power, kids lock, remote control, alarms, energy/water meters) are shared verbatim across families; add new ones only for resources unique to the new type.
3. Create `registry/by_type/<name>.py` with a `DeviceRegistry(name=..., capabilities=_build([...]))`. Use `pattern_capabilities` instead of `capabilities` for any resource whose `href` isn't fixed (for example per-compartment fridge resources); see `refrigerator.py` for the pattern.
4. Register it in `_REGISTRY_BY_KEY` in `registry/by_type/__init__.py`, keyed on the lowercased, space/hyphen-to-underscore-converted suffix of the device's `oneUiVersion` string (see `_type_key()` in that file for the exact transform).
4. Register it in `_REGISTRY_BY_KEY` in `registry/by_type/__init__.py`, keyed on the lowercased, space/hyphen-to-underscore-converted suffix of the device's `oneUiVersion` string (see `_type_key()` in that file for the exact transform). If the device never reports `oneUiVersion` — confirmed true for washers — add its consumer-model prefix to `_CONSUMER_PREFIX_TO_KEY` instead, so `for_device_by_model()` can route it.
5. Add golden-file coverage in `tests/` against a captured `/device/0` dump for the new type.
No config-flow or coordinator changes are needed. Device-type detection and entity wiring are fully driven by the registry.
No config-flow changes are needed. Device-type detection and entity wiring are fully driven by the registry.
---
## Known DTLS behavior
## Known device behavior
Samsung's RT-OCF DTLS stack occasionally closes sessions actively, usually right after a Block2 GET or in the seconds after a POST. Retry/retransmit bounds and inter-request pacing live in the `smartthings-local` protocol layer (tuned against measured per-firmware request-rate ceilings); reconnect-with-backoff and stale-state fallback live in this repo's `coordinator.py`. From HA's perspective a brief reconnect looks like an entity holding its last value for one poll cycle rather than going `unavailable`.
Samsung's firmware occasionally drops the DTLS session briefly — this is normal appliance-side behavior, not a bug. The integration reconnects automatically, and from HA's perspective a brief reconnect looks like an entity holding its last value for one poll cycle rather than going `unavailable`. When an appliance supports it, the integration prefers push-based updates (instant, via `observe.py`) over polling, falling back to polling otherwise.
If reconnects become persistent (more than a handful per minute), something's actually wrong. Check the appliance's Wi-Fi link first, then look for a competing DTLS client on the LAN: Samsung's RT-OCF DTLS allows only one active session per peer.
If reconnects become persistent (more than a handful per minute), something's actually wrong. Check the appliance's Wi-Fi link first, then look for a competing DTLS client on the LAN — only one active session per appliance is allowed at a time.
---
@@ -163,7 +156,7 @@ If reconnects become persistent (more than a handful per minute), something's ac
Patches are welcome, especially:
- New `by_type/` registries for appliance families not yet covered (washer, AC, microwave, etc.) on the same Tizen RT 3.x firmware family.
- New `by_type/` registries for appliance families not yet covered (AC, microwave, etc.) on the same Tizen RT 3.x firmware family.
- Confirmation or refutation of compatibility on additional models within an already-supported type.
- Protocol-level fixes, which belong upstream in [`smartthings-local`](https://github.com/QuiteYellow/SmartThings-Local) rather than here. HA-side fixes (entities, config flow, coordinator, registry) belong in this repo.
+9 -4
View File
@@ -125,7 +125,7 @@ def _probe_and_validate(host: str, ca_cert_pem: str, ca_key_pem: str) -> dict:
import cbor2
from smartthings_local.protocol.dtls_session import DtlsCoapSession
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device
from .registry.by_type import for_device, for_device_by_model
_LOGGER.debug("Fetching Samsung cloud UUID from %s", _SAMSUNG_CLOUD_HOST)
try:
@@ -177,15 +177,20 @@ def _probe_and_validate(host: str, ca_cert_pem: str, ca_key_pem: str) -> dict:
.get('swVersionInfo', {})
.get('oneUiVersion', '')
)
info_resource = resources.get('/information/vs/0', {})
recognized_registry = (
for_device(one_ui_version) if one_ui_version else None
) or for_device_by_model(
info_resource.get('x.com.samsung.da.modelNum', ''),
info_resource.get('x.com.samsung.da.description', ''),
)
return {
"port": port,
"serial": serial,
"leaf_cert_pem": fullchain_pem,
"leaf_key_pem": leaf_key_pem,
"one_ui_version": one_ui_version,
"device_type_recognized": bool(
one_ui_version and for_device(one_ui_version) is not None
),
"device_type_recognized": recognized_registry is not None,
}
except CannotConnect:
raise
+14 -2
View File
@@ -20,7 +20,7 @@ from smartthings_local.protocol.dtls_session import DtlsCoapSession
from smartthings_local.ocf.state_cache import StateCache
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device
from .registry.by_type import for_device, for_device_by_model
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -115,6 +115,13 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
def last_resources(self) -> dict:
return self._cache.snapshot()
def resource(self, href: str) -> dict:
"""A single href's current rep. Cheaper than `last_resources.get(href)`
for callers that only need one href — `last_resources` copies every
tracked href's rep to build the snapshot dict, while this is a
direct O(1) cache lookup."""
return self._cache.get(href) or {}
@property
def observe_mode(self) -> str:
return self._observe.mode
@@ -273,7 +280,13 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
.get('swVersionInfo', {})
.get('oneUiVersion', ''))
self.one_ui_version = one_ui
info = resources.get('/information/vs/0', {})
reg = for_device(one_ui) if one_ui else None
if reg is None:
reg = for_device_by_model(
info.get('x.com.samsung.da.modelNum', ''),
info.get('x.com.samsung.da.description', ''),
)
unbound: list[str] = []
if reg is not None:
self._log.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui)
@@ -287,7 +300,6 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self.bound = bound
self._unbound_hrefs = unbound
info = resources.get('/information/vs/0', {})
serial = info.get('x.com.samsung.da.serialNum', '')
if not serial:
serial = self._entry.data[CONF_HOST]
+1 -1
View File
@@ -29,7 +29,7 @@ def _is_included(bound: BoundEntity, coordinator: 'LocalThingsCoordinator') -> b
if rep is None:
return False
if bound.desc.exists_fn is not None:
return bound.desc.exists_fn(rep)
return bound.desc.exists_fn(rep, coordinator.last_resources)
if bound.desc.field:
if not rep: # stub — resource known to exist, data not yet fetched
return True
+2 -2
View File
@@ -5,12 +5,12 @@
"config_flow": true,
"dependencies": [],
"documentation": "https://github.com/mbillow/localthings",
"iot_class": "local_polling",
"iot_class": "local_push",
"issue_tracker": "https://github.com/mbillow/localthings/issues",
"requirements": [
"cbor2>=5.4.6",
"pyOpenSSL>=23.0",
"smartthings-local>=0.1.0"
],
"version": "0.1.0"
"version": "0.4.1"
}
+8 -2
View File
@@ -42,12 +42,18 @@ class LocalThingsNumber(LocalThingsEntity, NumberEntity):
if desc.step is not None:
self._attr_native_step = desc.step
@property
def native_unit_of_measurement(self):
desc: NumberDesc = self._bound.desc
if desc.unit_fn is not None:
return desc.unit_fn(self.coordinator.resource(self._bound.href))
return self._attr_native_unit_of_measurement
def _range_from_resource(self) -> list | None:
desc: NumberDesc = self._bound.desc
if not desc.range_field:
return None
rep = self.coordinator.last_resources.get(self._bound.href) or {}
r = rep.get(desc.range_field)
r = self.coordinator.resource(self._bound.href).get(desc.range_field)
return r if (isinstance(r, (list, tuple)) and len(r) == 2) else None
@property
@@ -15,7 +15,7 @@ def flatten(bound: list[BoundEntity], resources: dict) -> dict[str, Any]:
out: dict[str, Any] = {}
for b in bound:
rep = resources.get(b.href) or {}
if b.desc.exists_fn is not None and not b.desc.exists_fn(rep):
if b.desc.exists_fn is not None and not b.desc.exists_fn(rep, resources):
continue
if b.desc.rep_fn is not None:
out[_key(b)] = b.desc.rep_fn(rep)
@@ -2,9 +2,9 @@
from typing import Optional
from ._base import DeviceRegistry
from . import dishwasher, dryer, oven, refrigerator
from . import dishwasher, dryer, oven, refrigerator, washer
__all__ = ['DeviceRegistry', '_type_key', 'for_device']
__all__ = ['DeviceRegistry', '_type_key', 'for_device', 'for_device_by_model']
_REGISTRY_BY_KEY: dict[str, DeviceRegistry] = {
@@ -12,6 +12,7 @@ _REGISTRY_BY_KEY: dict[str, DeviceRegistry] = {
'dryer': dryer.REGISTRY,
'oven': oven.REGISTRY,
'refrigerator': refrigerator.REGISTRY,
'washer': washer.REGISTRY,
}
@@ -55,3 +56,36 @@ def for_device(one_ui_version: str) -> Optional[DeviceRegistry]:
if key.endswith(f'_{rkey}'):
return reg
return None
# Consumer-model prefix (first two letters of the '_'-delimited token in
# `description` right before any '/board-info' suffix) -> registry key.
# NOT derived from `modelNum` -- washer and dryer share the same 'DA_WM_'
# internal board-family prefix there, and dishwasher's modelNum contains
# the substring 'WW', so a modelNum-only rule misroutes both.
_CONSUMER_PREFIX_TO_KEY: dict[str, str] = {
'WW': 'washer',
'WD': 'washer',
'DV': 'dryer',
'DW': 'dishwasher',
}
def for_device_by_model(model_num: str, description: str) -> Optional[DeviceRegistry]:
"""Fallback device-type detection for hardware that never reports
oneUiVersion (confirmed for washers -- their /otninformation/vs/0 has
no swVersionInfo key at all).
Args:
model_num: x.com.samsung.da.modelNum from /information/vs/0.
description: x.com.samsung.da.description from /information/vs/0.
Returns:
DeviceRegistry if the consumer-model code or modelNum resolves to a
known type, None otherwise.
"""
token = (description or '').split('/', 1)[0].rsplit('_', 1)[-1]
key = _CONSUMER_PREFIX_TO_KEY.get(token[:2].upper())
if key is None and '_REF_' in (model_num or ''):
key = 'refrigerator'
return _REGISTRY_BY_KEY.get(key) if key else None
@@ -30,6 +30,7 @@ REGISTRY = DeviceRegistry(
fridge.DOORS_FALLBACK,
fridge.TEMPERATURES_FALLBACK,
fridge.ICEMAKER_STATUS_FALLBACK,
fridge.REFRIGERATION_FALLBACK,
]),
pattern_capabilities=[
fridge.TEMP_CURRENT_GENERIC,
@@ -0,0 +1,26 @@
"""Washer device registry."""
from ..capabilities import common, dishwasher, fridge, ignored, operational, washer
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='washer',
capabilities=_build([
*ignored.IGNORED,
washer.POWER_GENERIC,
washer.POWER_VS_FALLBACK,
washer.KIDS_LOCK_GENERIC,
washer.KIDS_LOCK_VS_FALLBACK,
washer.REMOTE_CONTROL_GENERIC,
washer.REMOTE_CONTROL_VS_FALLBACK,
washer.WASHER_SETTINGS,
washer.WASHER_COURSE,
washer.BUZZER_SOUND,
washer.WASHER_JOB_BEGINNING_STATUS,
common.ALARMS,
washer.WASHER_ENERGY_METER,
common.WATER_METER,
operational.OPERATIONAL_STATE,
dishwasher.DIAGNOSIS,
fridge.FIRMWARE_UPDATE,
]),
)
@@ -21,16 +21,29 @@ def _num(v):
return None
def _clamp_power(v):
def clamp_power(v):
n = _num(v)
return 0.0 if (n is not None and n < 0) else n
def _wh_to_kwh(v):
def wh_to_kwh(v):
n = _num(v)
return round(n / 1000.0, 2) if n is not None else None
def normalize_temp_unit(raw, default='°F'):
"""'C'/'Celsius' -> '°C', 'F'/'Fahrenheit' -> '°F'. Falls back to
`default` for any other/missing value. Shared by fridge.py and oven.py,
both of which read a per-device unit off a `/temperature*` resource
instead of assuming one (see fridge.py's module docstring, issue #7)."""
raw = (raw or '').strip().upper()
if raw.startswith('C'):
return '°C'
if raw.startswith('F'):
return '°F'
return default
def _ml_to_l(v):
n = _num(v)
return round(n / 1000.0, 1) if n is not None else None
@@ -90,10 +103,10 @@ ENERGY_METER = Capability(
entities=(
SensorDesc(key='power_watts', field='x.com.samsung.da.instantaneousPower',
name='Power', device_class='power', state_class='measurement',
unit='W', value_fn=_clamp_power),
unit='W', value_fn=clamp_power),
SensorDesc(key='energy_kwh', field='x.com.samsung.da.cumulativePower',
name='Energy', device_class='energy',
state_class='total_increasing', unit='kWh', value_fn=_wh_to_kwh),
state_class='total_increasing', unit='kWh', value_fn=wh_to_kwh),
),
)
@@ -31,22 +31,37 @@ DISHWASHER_SETTINGS = Capability(
# /course/vs/0 — cycle selection and course options (RMW on options array)
# ---------------------------------------------------------------------------
# Course IDs in editCourseList byte order, matched to Samsung app display names.
# IDs are uppercase hex strings matching the Course_XX encoding in options[].
_COURSE_ID_TO_NAME: dict[str, str] = {
'0E': 'AI Wash',
'07': 'Pre blast',
'90': 'Self clean',
'86': 'Normal',
'83': 'Express 60',
'84': 'Heavy',
'8D': 'Pots and pans',
'80': 'Delicate',
'8E': 'Plastic',
'8F': 'Baby Care',
}
_COURSE_NAME_TO_ID = {v: k for k, v in _COURSE_ID_TO_NAME.items()}
_CYCLE_OPTIONS = tuple(_COURSE_ID_TO_NAME.values())
# Course IDs are uppercase hex strings matching the Course_XX encoding in
# options[]. Display names are not kept here -- they live in
# strings.json/translations under entity.select.dishwasher_cycle.state.<id,
# lowercased>, same as every other device-enum select in this integration
# (see fridge.py's translation_key entities and select.py's _display()), so
# they can be localized instead of hardcoded to English.
#
# No static fallback list is kept here -- a hardcoded course table would
# show options a given dishwasher model doesn't actually have (or hide ones
# it does). The only trustworthy per-device source is the live
# x.com.samsung.da.editCourseList on /wm/editcourse/vs/0. When that's
# absent (e.g. never populated until the app's course-edit screen has been
# opened at least once), the cycle select isn't created at all -- see
# CYCLE_OPTIONS's exists_fn below. (x.com.samsung.da.options' MostUsed_*
# entry was considered as a second fallback source, but its bytes beyond
# the first don't correspond to any confirmed course code on hardware we
# have dumps for, so it isn't trustworthy either -- see washer.py's
# _cycle_options docstring for the byte-level evidence.)
def _parse_edit_course_list(raw):
"""'EditCourseList_0E07908683848D808E8F' -> ['0E', '07', ...]."""
if not isinstance(raw, str) or '_' not in raw:
return []
codes = raw.split('_', 1)[1]
return [codes[i:i + 2] for i in range(0, len(codes) - 1, 2)]
def _cycle_options(resources):
rep = resources.get('/wm/editcourse/vs/0') or {}
return _parse_edit_course_list(rep.get('x.com.samsung.da.editCourseList'))
def _option_value(options, prefix):
@@ -63,14 +78,11 @@ def _replace_in_options(options, prefix, new_value):
def _cycle_write(p, rep, href=None):
course_id = _COURSE_NAME_TO_ID.get(p)
if not course_id:
return None
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'Course', course_id),
'x.com.samsung.da.options': _replace_in_options(opts, 'Course', p),
}
@@ -100,17 +112,18 @@ CYCLE_OPTIONS = Capability(
href='/course/vs/0',
entities=(
SelectDesc(key='cycle', name='Cycle', icon='mdi:dishwasher',
options=_CYCLE_OPTIONS,
rep_fn=lambda rep: _COURSE_ID_TO_NAME.get(
_option_value(rep.get('x.com.samsung.da.options'), 'Course')
),
translation_key='dishwasher_cycle',
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'),
write_fn=_cycle_write),
SwitchDesc(key='storm_wash', name='Storm Wash+', icon='mdi:weather-lightning-rainy',
rep_fn=lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), 'StormWashZone') == 'On',
write_fn=_storm_wash_write),
SwitchDesc(key='auto_release_dry', name='Auto release dry', icon='mdi:door-open',
exists_fn=lambda rep: any(
exists_fn=lambda rep, resources: any(
isinstance(o, str) and o.startswith('AutoDoorRelease_')
for o in (rep.get('x.com.samsung.da.options') or [])
),
@@ -2,9 +2,13 @@
Resources verified against the dump at local-tools/dumps/10.0.0.254.json.
Temperature fields in all /temperature/* resources are in Fahrenheit on this
model. Setpoints are NumberDesc with direct-write write_fn — generic caps
derive the CoAP PUT path from href at write time.
Temperature unit is read live from each resource, not assumed: the RF9000B
dump reports Fahrenheit ("units": "F" / "x.com.samsung.da.unit": "Fahrenheit"),
but a TP1X_REF_21K dump (issue #7) reports the same fields in Celsius for the
exact same resources — the device tells you which one it is, it's just never
been read before. See `_temp_unit`/`_temp_item_unit` below. Setpoints are
NumberDesc with direct-write write_fn — generic caps derive the CoAP PUT path
from href at write time.
Multi-instance note: the two door resources (/door/cooler/0 and
/door/freezer/0) and the two ice-maker resources (/icemaker/one/vs/0 and
@@ -18,6 +22,7 @@ from ..entities import (
BinarySensorDesc, ButtonDesc, NumberDesc, SelectDesc, SensorDesc,
SwitchDesc, TimeDesc,
)
from .common import normalize_temp_unit
# Display names for the beverage zone, flex zone, ice type, and
# ice-making-status enums below live in strings.json / translations/en.json,
@@ -33,6 +38,12 @@ def _int(v):
return None
def _temp_unit(rep):
"""'units': 'C'/'F' (or 'Celsius'/'Fahrenheit') -> '°C'/'°F'. Defaults to
°F (this module's original assumption) if the device omits the field."""
return normalize_temp_unit(rep.get('units'))
# ---------------------------------------------------------------------------
# Temperature (generic — covers /temperature/current/* and /temperature/desired/*)
# ---------------------------------------------------------------------------
@@ -45,7 +56,7 @@ TEMP_CURRENT_GENERIC = Capability(
entities=(
SensorDesc(key='temperature', field='temperature',
name=None, icon='mdi:thermometer',
device_class='temperature', unit='°F',
device_class='temperature', unit_fn=_temp_unit,
state_class='measurement'),
),
)
@@ -57,7 +68,7 @@ TEMP_SETPOINT_GENERIC = Capability(
poll_tier='warm',
entities=(
NumberDesc(key='setpoint', field='temperature',
name=None, device_class='temperature', unit='°F',
name=None, device_class='temperature', unit_fn=_temp_unit,
native_min=-20.0, native_max=50.0,
range_field='range', entity_category='config',
write_fn=lambda p, rep, href=None: (
@@ -119,7 +130,7 @@ ICEMAKER_GENERIC = Capability(
translation_key='ice_type',
entity_category='config',
options_field='x.com.samsung.da.iceType.supported',
exists_fn=lambda rep: bool(rep.get('x.com.samsung.da.iceType.supported')),
exists_fn=lambda rep, resources: bool(rep.get('x.com.samsung.da.iceType.supported')),
write_fn=_icemaker_write('x.com.samsung.da.iceType.desired')),
),
)
@@ -172,12 +183,16 @@ STATUS_LOCK = Capability(
)
# ---------------------------------------------------------------------------
# Defrost delay
# Defrost delay / active-defrost status
#
# /defrost/delay/vs/0 is the writable toggle. /defrost/block/vs/0 just
# reports whether the defrost cycle is currently blocked as a result (a
# derived status, not an independent control) — exposed as a read-only
# diagnostic binary sensor.
# /defrost/delay/vs/0 is the writable toggle to postpone a scheduled
# defrost. /defrost/block/vs/0 is an unrelated, independently-varying
# status: despite its "block" naming (originally assumed to mean "defrost
# is being withheld"), live dumps confirm DEFROST_BLOCK_ON means the
# defrost cycle is *actively running* right now, seen with defrost_delay
# off -- i.e. "block" refers to the evaporator/coil block being defrosted,
# not a blocking/prevention state. Exposed as a read-only diagnostic
# binary sensor.
# ---------------------------------------------------------------------------
DEFROST_DELAY = Capability(
@@ -198,8 +213,8 @@ DEFROST_BLOCK_STATUS = Capability(
href='/defrost/block/vs/0',
poll_tier='warm',
entities=(
BinarySensorDesc(key='defrost_blocked', field='x.com.samsung.da.modes',
name='Defrost blocked', icon='mdi:snowflake-off',
BinarySensorDesc(key='defrost_active', field='x.com.samsung.da.modes',
name='Defrost active', icon='mdi:snowflake-melt',
entity_category='diagnostic',
value_fn=lambda modes: bool(modes) and modes[0] == 'DEFROST_BLOCK_ON'),
),
@@ -504,7 +519,7 @@ FLEX_ZONE = Capability(
translation_key='flex_zone_mode',
entity_category='config',
options_field='x.com.samsung.da.supportedOptions',
exists_fn=lambda rep: bool(
exists_fn=lambda rep, resources: bool(
rep.get('x.com.samsung.da.supportedOptions')),
value_fn=lambda modes: next(
(m for m in (modes or [])
@@ -590,6 +605,13 @@ def _temp_item_value(items, keyword):
return None
def _temp_item_unit(items, keyword):
for item in (items or []):
if keyword.lower() in (item.get('x.com.samsung.da.description') or '').lower():
return normalize_temp_unit(item.get('x.com.samsung.da.unit'))
return '°F'
TEMPERATURES_FALLBACK = Capability(
href='/temperatures/vs/0',
match_fn=lambda rep, resources: not _any_temperature_generic(resources),
@@ -597,11 +619,15 @@ TEMPERATURES_FALLBACK = Capability(
entities=(
SensorDesc(key='freezer_temperature', field='x.com.samsung.da.items',
name='Freezer temperature', icon='mdi:thermometer',
device_class='temperature', unit='°F', state_class='measurement',
device_class='temperature', state_class='measurement',
unit_fn=lambda rep: _temp_item_unit(
rep.get('x.com.samsung.da.items'), 'Freezer'),
value_fn=lambda items: _temp_item_value(items, 'Freezer')),
SensorDesc(key='fridge_temperature', field='x.com.samsung.da.items',
name='Fridge temperature', icon='mdi:thermometer',
device_class='temperature', unit='°F', state_class='measurement',
device_class='temperature', state_class='measurement',
unit_fn=lambda rep: _temp_item_unit(
rep.get('x.com.samsung.da.items'), 'Fridge'),
value_fn=lambda items: _temp_item_value(items, 'Fridge')),
),
)
@@ -628,3 +654,34 @@ ICEMAKER_STATUS_FALLBACK = Capability(
{'x.com.samsung.da.iceMaker': 'On' if p else 'Off'})),
),
)
# OCF-native /refrigeration/0 (issue #7's unbound_hrefs) -- the odd one out
# in this section: its three fields duplicate two *different* richer
# hrefs (REFRIGERATION's rapidFridge/rapidFreezing and
# DEFROST_BLOCK_STATUS's defrost_active), each absent independently, so a
# single capability-level match_fn can't express it. Gated per-entity
# (exists_fn) instead: rapid_fridge/rapid_freezing back off only when
# REFRIGERATION's href is present; defrost_active only when
# DEFROST_BLOCK_STATUS's is. No write path confirmed for this href, so
# these are read-only, unlike REFRIGERATION's switches.
REFRIGERATION_FALLBACK = Capability(
href='/refrigeration/0',
poll_tier='warm',
entities=(
BinarySensorDesc(key='defrost_active', field='defrost',
name='Defrost active', icon='mdi:snowflake-melt',
entity_category='diagnostic',
value_fn=lambda v: bool(v),
exists_fn=lambda rep, resources: '/defrost/block/vs/0' not in resources),
BinarySensorDesc(key='rapid_fridge', field='rapidCool',
name='Rapid fridge', icon='mdi:fridge-industrial',
entity_category='config',
value_fn=lambda v: bool(v),
exists_fn=lambda rep, resources: '/refrigeration/vs/0' not in resources),
BinarySensorDesc(key='rapid_freezing', field='rapidFreeze',
name='Rapid freezing', icon='mdi:snowflake',
entity_category='config',
value_fn=lambda v: bool(v),
exists_fn=lambda rep, resources: '/refrigeration/vs/0' not in resources),
),
)
@@ -61,8 +61,12 @@ IGNORED: list[Capability] = [
# Redundant with capabilities already declared elsewhere.
# /speakersound/vs/0 duplicates /settings/sound/volume/vs/0 (laundry.SOUND_VOLUME).
Capability(href='/speakersound/vs/0'),
# /wm/editcourse/vs/0 is the raw encoded course-list byte string already
# decoded and exposed via /course/vs/0 (dishwasher.CYCLE_OPTIONS).
# /wm/editcourse/vs/0 has no entities of its own -- x.com.samsung.da.
# editCourseList is read directly out of the resource snapshot by
# dishwasher.CYCLE_OPTIONS's and washer.WASHER_COURSE's cycle select
# (options=_cycle_options) to build that device's actual supported
# course list, rather than exposing this href's raw byte string
# through its own entity.
Capability(href='/wm/editcourse/vs/0'),
# Bixby audio feedback (chime + volume played when Bixby starts/stops
@@ -85,6 +89,11 @@ IGNORED: list[Capability] = [
# supportedAiLevel on hardware seen so far, so there's no real choice
# to expose. Revisit if a device surfaces more than one supported level.
Capability(href='/energy/ailevel/vs/0'),
# OCF-native vacation-mode flag (fridge). Only one value ('RVACATION_OFF')
# has ever been seen in `modes` (issue #7's dump), same "no real choice
# to expose yet" situation as /energy/ailevel/vs/0 above. Revisit if a
# device surfaces it toggled on.
Capability(href='/mode/0'),
# Opaque integer with no supportedModes/options list to interpret it
# against — meaning unclear from the raw resource alone.
Capability(href='/runningmode/vs/0'),
@@ -94,4 +103,33 @@ IGNORED: list[Capability] = [
Capability(href='/energy/planner/vs/0'),
# Temperature-unit display preference, redundant with HA's own units.
Capability(href='/wm/submode/vs/0'),
# Read-only re-encoding of the course already exposed by
# washer.WASHER_COURSE at /course/vs/0 (x.com.samsung.da.st.washerMode
# is literally "Table_02_Course_<same hex code>").
Capability(href='/st/washercourse/vs/0'),
# Empty on every washer dump seen so far.
Capability(href='/wm/welcomemsg/vs/0'),
# User-saved custom course slots (F1-FA). No controllable/observable
# state without a multi-slot editor; revisit if that becomes valuable.
Capability(href='/wm/personalcourse/vs/0'),
# OCF-native energy resource is empty ({}) on washer hardware seen so
# far, unlike /power/0, /kidslock/0, /remotectrl/0 which do carry real
# data -- common.ENERGY_METER on /energy/consumption/vs/0 is the only
# real source for this control.
Capability(href='/energy/consumption/0'),
# Empty ({}) on every washer dump seen so far -- nothing to expose.
Capability(href='/cycleinterface/vs/0'),
# OCF-native duplicate of /drlc/vs/0 above -- same utility-program
# dependency this integration doesn't support locally.
Capability(href='/drlc/0'),
# OCF-native duplicate of /operational/state/vs/0, which is already
# modeled by operational.OPERATIONAL_STATE (a richer, write-capable
# capability with start/pause/stop buttons and a delay-start control)
# used by washer, dishwasher, dryer, and oven. This generic href only
# carries read-only overlapping data (current job state, remaining
# time, progress percentage) with no write path -- not worth building a
# parallel write-capable capability around an unverified generic OCF
# write contract.
Capability(href='/operational/state/0'),
]
@@ -47,6 +47,16 @@ def _format_delay(hours):
return f'{h}:{m:02d}:00'
def _delay_field(rep):
"""Washer hardware reports the delay-until-start duration under
'delayEndTime' instead of 'delayStartTime' (both hold a duration, not a
wall-clock time -- see _delay_hours). Write back whichever key the
device itself is using; default to delayStartTime for hardware that
reports neither yet (matches prior behavior)."""
return ('x.com.samsung.da.delayEndTime' if 'x.com.samsung.da.delayEndTime' in rep
else 'x.com.samsung.da.delayStartTime')
def _finish_time(remaining_str):
if not remaining_str:
return None
@@ -65,7 +75,9 @@ OPERATIONAL_STATE = Capability(
poll_tier='hot',
entities=(
SensorDesc(key='machine_state', field='x.com.samsung.da.state',
name='Machine state', value_fn=_to_ocf),
name='Machine state', device_class='enum',
options=('idle', 'active', 'pause'),
translation_key='machine_state', value_fn=_to_ocf),
# cycle_active is a bool derived from machine_state; used by the
# adapter to gate oven writes (cycle_active_field='cycle_active').
# Harmless for non-oven appliances — just an extra bool in state.
@@ -83,9 +95,11 @@ OPERATIONAL_STATE = Capability(
else _progress(rep.get('x.com.samsung.da.progress'))
)),
SensorDesc(key='progress_percentage',
field='x.com.samsung.da.progressPercentage',
name='Progress percent', unit='%', state_class='measurement',
value_fn=_int),
rep_fn=lambda rep: (
0 if _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) != 'active'
else _int(rep.get('x.com.samsung.da.progressPercentage'))
)),
# Only show finish time when machine is actively running. Samsung
# firmware leaves a stale remainingTime after a cycle ends, and
# freezes it at '00:01:00' when progress reaches 'Finish'.
@@ -96,14 +110,15 @@ OPERATIONAL_STATE = Capability(
or rep.get('x.com.samsung.da.progress') == 'Finish'
else _finish_time(rep.get('x.com.samsung.da.remainingTime'))
)),
NumberDesc(key='delay_start_hours', field='x.com.samsung.da.delayStartTime',
name='Delay start', icon='mdi:timer-plus-outline',
NumberDesc(key='delay_start_hours', name='Delay start', icon='mdi:timer-plus-outline',
device_class='duration', unit='h',
native_min=0, native_max=24, step=1,
value_fn=_delay_hours,
rep_fn=lambda rep: _delay_hours(
rep.get('x.com.samsung.da.delayStartTime')
or rep.get('x.com.samsung.da.delayEndTime')),
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.delayStartTime': _format_delay(p)})),
{_delay_field(rep): _format_delay(p)})),
ButtonDesc(key='start', field='', name='Start cycle', payload='Run',
icon='mdi:play',
write_fn=lambda p, rep, href=None: (
@@ -29,6 +29,7 @@ from ..entities import (
BinarySensorDesc, ButtonDesc, NumberDesc, SelectDesc, SensorDesc,
SwitchDesc,
)
from .common import normalize_temp_unit
# ---------------------------------------------------------------------------
# Constants
@@ -222,7 +223,9 @@ OVEN_OPERATIONAL_STATE = Capability(
poll_tier='hot',
entities=(
SensorDesc(key='machine_state', field='x.com.samsung.da.state',
name='Machine state', icon='mdi:stove', value_fn=_to_ocf),
name='Machine state', icon='mdi:stove',
device_class='enum', options=('idle', 'active', 'pause'),
translation_key='machine_state', value_fn=_to_ocf),
BinarySensorDesc(key='cycle_active', field='x.com.samsung.da.state',
name='Cycle active', device_class='running',
value_fn=lambda v: _SAMSUNG_STATE_TO_OCF.get(v) == 'active'),
@@ -255,13 +258,26 @@ OVEN_CAVITY = Capability(
),
)
def _oven_temp_unit(rep):
"""Same shape/risk as fridge.py's TEMPERATURES_FALLBACK: this is the
same aggregate `/temperatures/vs/0` items[] resource type, which on
fridge hardware carries a per-item `x.com.samsung.da.unit` field
('Celsius'/'Fahrenheit') that was previously hardcoded away (issue #7).
No live oven dump has surfaced a non-Celsius reading yet, so this keeps
the verified '°C' default when the field is absent, but reads it live
if a device ever reports otherwise."""
items = rep.get('x.com.samsung.da.items') or []
unit = items[0].get('x.com.samsung.da.unit') if items else None
return normalize_temp_unit(unit, default='°C')
OVEN_SETPOINT = Capability(
href='/temperatures/vs/0',
poll_tier='hot',
entities=(
# NumberDesc first — test_oven_setpoint_write_is_read_modify_write uses entities[0]
NumberDesc(key='oven_setpoint', field='x.com.samsung.da.items',
name='Setpoint', device_class='temperature', unit='°C',
name='Setpoint', device_class='temperature', unit_fn=_oven_temp_unit,
native_min=float(SETPOINT_MIN_C), native_max=float(SETPOINT_MAX_C),
step=float(SETPOINT_STEP_C), icon='mdi:thermometer-chevron-up',
value_fn=lambda items: _int(
@@ -269,7 +285,7 @@ OVEN_SETPOINT = Capability(
write_fn=_oven_setpoint_write),
SensorDesc(key='current_temp_c', field='x.com.samsung.da.items',
name='Temperature', device_class='temperature',
state_class='measurement', unit='°C',
state_class='measurement', unit_fn=_oven_temp_unit,
value_fn=lambda items: _int(
(items[0].get('x.com.samsung.da.current') if items else None))),
),
@@ -0,0 +1,440 @@
"""Capabilities specific to washer appliances (Samsung DA_WM_TP1-class
front-load washers).
Resources verified against two live WW90DG6U25LEU4 dumps (Table_02 course
family). Washers never report `oneUiVersion` -- see
`registry/by_type/__init__.py`'s `for_device_by_model()` for the fallback
detection this device type requires.
"""
from datetime import datetime, timezone
from ..capability import Capability
from ..entities import BinarySensorDesc, SelectDesc, SensorDesc, SwitchDesc
from .common import clamp_power, wh_to_kwh
# ---------------------------------------------------------------------------
# Course_XX hex codes. The 23 codes named in strings.json/translations
# under entity.select.washer_cycle.state.<id, lowercased> were captured
# from a live WW90DG6U25LEU4's x.com.samsung.da.editCourseList
# (EditCourseList_1C1D211B1E29243328262722202325322F2E30662D8F96), matched
# positionally against a Slovak-UI user's screenshots of their app's course
# list (same order, same count -- see issue #2) and cross-checked against
# the printed user manual's course table (confirming e.g. '8F' as 'Intense
# Cold', not the position-adjacent-looking but distinct 'Mixed Load', a
# cycle the manual marks "applicable models only" and that does not appear
# in this device's editCourseList -- nor does 'AI Wash', also "applicable
# models only"). FixedCourseList_1C29 (the two courses always pinned in the
# app) maps to '1C'/'29' = Eco 40-60 and Drum Clean+, which matches what
# you'd expect to be pinned (default cycle + maintenance cycle),
# corroborating the positional match.
#
# No static fallback list of those codes is kept here, deliberately: other
# washer models have a different actual course set (a second dump's active
# course, '65', isn't even in the list above; models with 'AI Wash'/'Mixed
# Load' -- both "applicable models only" per the manual -- would have yet
# another set), so hardcoding one device's list would show/hide the wrong
# options on a different model. _cycle_options() below reads only the live
# x.com.samsung.da.editCourseList; if a device doesn't populate that
# resource, the cycle select isn't created at all (see WASHER_COURSE's
# exists_fn). x.com.samsung.da.options' MostUsed_* entry was considered as
# a fallback source (its first byte reliably equals the currently-selected
# Course_XX on both dumps we have), but the bytes after that don't
# correspond to any confirmed course code on either device -- e.g. dump 1's
# MostUsed_1C8410923FA67F00000000000000 decodes to
# ['1C','84','10','92','3F','A6','7F',...] and only '1C' is a real code --
# so it isn't trustworthy as a list of selectable courses and isn't used.
# ---------------------------------------------------------------------------
def _hex_pairs(codes):
"""'1C1D211B1E29...' -> ['1C', '1D', '21', '1B', '1E', '29', ...]."""
return [codes[i:i + 2] for i in range(0, len(codes) - 1, 2)]
def _parse_edit_course_list(raw):
"""'EditCourseList_1C1D211B1E29...' -> ['1C', '1D', '21', '1B', '1E', '29', ...]."""
if not isinstance(raw, str) or '_' not in raw:
return []
return _hex_pairs(raw.split('_', 1)[1])
def _cycle_options(resources):
rep = resources.get('/wm/editcourse/vs/0') or {}
return _parse_edit_course_list(rep.get('x.com.samsung.da.editCourseList'))
# ---------------------------------------------------------------------------
# /washer/vs/0 -- wash temperature, spin speed, rinse cycle count
#
# Despite the shared href, this is unrelated to dryer.DRYER_SETTINGS (also
# bound to '/washer/vs/0') -- an artifact of Samsung reusing the same OCF
# path for different device families. Only one of the two ever binds for a
# given device, since dryer and washer are separate by_type registries.
# ---------------------------------------------------------------------------
WASHER_SETTINGS = Capability(
href='/washer/vs/0',
entities=(
SelectDesc(key='wash_temperature', field='x.com.samsung.da.waterTemperature',
name='Wash temperature', icon='mdi:thermometer-water',
entity_category='config',
options_field='x.com.samsung.da.supportedWaterTemperature',
write_fn=lambda p, rep, href=None: (
['washer', 'vs', '0'], {'x.com.samsung.da.waterTemperature': p})),
SelectDesc(key='spin_speed', field='x.com.samsung.da.spinLevel',
name='Spin speed', icon='mdi:sync',
entity_category='config',
options_field='x.com.samsung.da.supportedSpinLevel',
write_fn=lambda p, rep, href=None: (
['washer', 'vs', '0'], {'x.com.samsung.da.spinLevel': p})),
SelectDesc(key='rinse_cycles', field='x.com.samsung.da.rinseCycles',
name='Rinse cycles', icon='mdi:water-sync',
entity_category='config',
options_field='x.com.samsung.da.supportedRinseCycles',
write_fn=lambda p, rep, href=None: (
['washer', 'vs', '0'], {'x.com.samsung.da.rinseCycles': p})),
),
)
# ---------------------------------------------------------------------------
# /course/vs/0 -- selected course, read/write (RMW on the options array,
# same shape as dishwasher.CYCLE_OPTIONS._cycle_write).
# ---------------------------------------------------------------------------
def _option_value(options, prefix):
for o in (options or []):
if isinstance(o, str) and o.startswith(prefix + '_'):
return o.split('_', 1)[1]
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 _cycle_write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'Course', p),
}
# Drum Clean+ maintenance tracking, from the same options[] array as the
# selected course. DrumCleanProposal_<N> is the wash-cycle interval between
# recommended cleans; WashingTimes_<N> is the count since the last one --
# their difference is exactly the "N cycles until due" figure the Samsung
# app shows (verified: DrumCleanProposal_40 - WashingTimes_3 == 37, matching
# a live app screenshot's "Potreba cistenia po 37 cykloch"). DrumCleanLog_
# is the last-clean timestamp (verified against the same screenshot's "10
# days ago"); no explicit timezone field accompanies it on this resource,
# so it's treated as UTC, matching this integration's convention for other
# bare ISO datetime fields (see fridge.py's night-light schedule comment).
def _drum_clean_cycles_remaining(rep):
opts = rep.get('x.com.samsung.da.options') or []
proposal = _option_value(opts, 'DrumCleanProposal')
washed = _option_value(opts, 'WashingTimes')
if proposal is None or washed is None:
return None
try:
return max(int(proposal) - int(washed), 0)
except ValueError:
return None
def _drum_clean_last_cleaned(rep):
raw = _option_value(rep.get('x.com.samsung.da.options'), 'DrumCleanLog')
if not raw:
return None
try:
return datetime.fromisoformat(raw).replace(tzinfo=timezone.utc)
except ValueError:
return None
# Detergent/softener auto-dispense dosing, from the same options[] array
# (issue #9). '<Prefix>LevelCtrl_<code>' is the selected dose quantity;
# '<Prefix>Level2Ctrl_<code>' is a second dial -- water hardness for
# detergent, concentration for softener -- matching the SmartThings app's
# two-field dispenser screens ("Distributeur de lessive": Quantité + Dureté
# de l'eau; "Distributeur d'adoucissant": Quantité + Concentration, per
# issue #9's screenshots). 'Supported<Prefix>Ctrl_<hexpairs>' lists the
# valid raw codes for its field, same hex-pair shape as EditCourseList.
# '<Prefix>Alarm_<On/Off>' is a low-reservoir warning flag.
#
# Label mapping (entity.select.washer_dosing_quantity/washer_detergent_
# water_hardness/washer_softener_concentration in strings.json) is an
# assumed, not cross-device-verified, reading of the single issue #9 dump +
# screenshots: LevelCtrl's 4 codes as None/Low/Medium/High (00 has no
# on-screen equivalent -- the app's Quantité picker only offers
# Faible/Moyen/Élevé, i.e. codes 01-03; 00 is assumed to be what
# "Activation" off collapses to) matches DetergentLevelCtrl_3/
# SoftenerLevelCtrl_3 = "Élevé" on both dispensers. Level2Ctrl's 3 codes as
# Soft/Medium/Hard for detergent (Dureté de l'eau: Douce/Moyenne/Dure)
# matches DetergentLevel2Ctrl_2 = "Moyenne". The same 3-code shape as
# 1x/2x/3x for softener concentration does *not* cleanly match
# SoftenerLevel2Ctrl_2 against the screenshot's "3x" -- assumed to be a
# setting the user changed in the app between the dump (issue body) and the
# screenshots (a later comment), not a different code scheme, since it's
# otherwise identical in shape to the detergent side. Revisit if a second
# device's dump contradicts this.
def _supported_level_options(resources, prefix):
rep = resources.get('/course/vs/0') or {}
raw = _option_value(rep.get('x.com.samsung.da.options'), f'Supported{prefix}')
return _hex_pairs(raw) if raw else []
def _level_options(prefix):
return lambda resources: _supported_level_options(resources, prefix)
def _level_write(prefix):
def write(p, rep, href=None):
opts = list(rep.get('x.com.samsung.da.options') or [])
if not opts:
return None
return ['course', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, prefix, p),
}
return write
def _dosing_low(prefix):
return lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), prefix) not in (None, 'Off')
def _dosing_alarm_exists(prefix):
return lambda rep, resources: _option_value(
rep.get('x.com.samsung.da.options'), prefix) is not None
WASHER_COURSE = Capability(
href='/course/vs/0',
entities=(
SelectDesc(key='cycle', name='Cycle', icon='mdi:washing-machine',
translation_key='washer_cycle',
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'),
write_fn=_cycle_write),
SensorDesc(key='drum_clean_cycles_remaining', name='Drum clean due in',
icon='mdi:washing-machine-alert', unit='cycles',
state_class='measurement',
exists_fn=lambda rep, resources: _drum_clean_cycles_remaining(rep) is not None,
rep_fn=_drum_clean_cycles_remaining),
SensorDesc(key='drum_clean_last_cleaned', name='Drum last cleaned',
icon='mdi:calendar-clock', device_class='timestamp',
entity_category='diagnostic',
exists_fn=lambda rep, resources: _drum_clean_last_cleaned(rep) is not None,
rep_fn=_drum_clean_last_cleaned),
SelectDesc(key='detergent_quantity', name='Detergent quantity', icon='mdi:cup-water',
translation_key='washer_dosing_quantity',
entity_category='config',
options=_level_options('DetergentLevelCtrl'),
exists_fn=lambda rep, resources: bool(
_level_options('DetergentLevelCtrl')(resources)),
rep_fn=lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), 'DetergentLevelCtrl'),
write_fn=_level_write('DetergentLevelCtrl')),
SelectDesc(key='detergent_water_hardness', name='Detergent water hardness',
icon='mdi:water-opacity',
translation_key='washer_detergent_water_hardness',
entity_category='config',
options=_level_options('DetergentLevel2Ctrl'),
exists_fn=lambda rep, resources: bool(
_level_options('DetergentLevel2Ctrl')(resources)),
rep_fn=lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), 'DetergentLevel2Ctrl'),
write_fn=_level_write('DetergentLevel2Ctrl')),
SelectDesc(key='softener_quantity', name='Softener quantity', icon='mdi:flask-outline',
translation_key='washer_dosing_quantity',
entity_category='config',
options=_level_options('SoftenerLevelCtrl'),
exists_fn=lambda rep, resources: bool(
_level_options('SoftenerLevelCtrl')(resources)),
rep_fn=lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), 'SoftenerLevelCtrl'),
write_fn=_level_write('SoftenerLevelCtrl')),
SelectDesc(key='softener_concentration', name='Softener concentration',
icon='mdi:flask-plus-outline',
translation_key='washer_softener_concentration',
entity_category='config',
options=_level_options('SoftenerLevel2Ctrl'),
exists_fn=lambda rep, resources: bool(
_level_options('SoftenerLevel2Ctrl')(resources)),
rep_fn=lambda rep: _option_value(
rep.get('x.com.samsung.da.options'), 'SoftenerLevel2Ctrl'),
write_fn=_level_write('SoftenerLevel2Ctrl')),
BinarySensorDesc(key='detergent_low', name='Detergent low',
icon='mdi:alert-circle-outline', device_class='problem',
exists_fn=_dosing_alarm_exists('DetergentAlarm'),
rep_fn=_dosing_low('DetergentAlarm')),
BinarySensorDesc(key='softener_low', name='Softener low',
icon='mdi:alert-circle-outline', device_class='problem',
exists_fn=_dosing_alarm_exists('SoftenerAlarm'),
rep_fn=_dosing_low('SoftenerAlarm')),
),
)
# ---------------------------------------------------------------------------
# /buzzersound/vs/0 -- buzzer volume and (on some units) a separate finish
# chime. Fields have no 'x.com.samsung.da.' prefix in this resource, unlike
# most other washer hrefs.
# ---------------------------------------------------------------------------
BUZZER_SOUND = Capability(
href='/buzzersound/vs/0',
entities=(
SelectDesc(key='buzzer_sound', field='setBuzzerSound',
name='Buzzer sound', icon='mdi:volume-high',
entity_category='config',
options_field='supportedBuzzerSound',
write_fn=lambda p, rep, href=None: (
['buzzersound', 'vs', '0'], {'setBuzzerSound': p})),
SelectDesc(key='finish_sound', field='setFinishSound',
name='Finish sound', icon='mdi:bell-ring',
entity_category='config',
exists_fn=lambda rep, resources: 'supportedFinishSound' in rep,
options_field='supportedFinishSound',
write_fn=lambda p, rep, href=None: (
['buzzersound', 'vs', '0'], {'setFinishSound': p})),
),
)
# ---------------------------------------------------------------------------
# /wm/jobbeginingstatus/vs/0 -- same href as dryer.JOB_BEGINNING_STATUS, but
# a different field name (currentStatus, not jobBeginingStatus).
# ---------------------------------------------------------------------------
WASHER_JOB_BEGINNING_STATUS = Capability(
href='/wm/jobbeginingstatus/vs/0',
poll_tier='warm',
entities=(
SensorDesc(key='job_beginning_status',
field='x.com.samsung.da.currentStatus',
name='Job beginning status',
entity_category='diagnostic'),
),
)
# ---------------------------------------------------------------------------
# OCF-native / '-vs' fallback pairs for power, kids-lock, remote control.
#
# Same shape as fridge.py's "Aggregate-resource fallbacks": the generic OCF
# href (/power/0, plain boolean 'value') is preferred when present; the
# vendor '-vs' href (richer historically, but for these three controls just
# a string-encoded duplicate) only binds when the generic href is absent
# from this device's resource set, via match_fn. Scoped to washer.py, not
# common.py -- no other device type has been confirmed to expose the
# generic hrefs, so this must not change behavior for dishwasher/dryer/
# oven/refrigerator.
# ---------------------------------------------------------------------------
POWER_GENERIC = Capability(
href='/power/0',
entities=(
SwitchDesc(key='power_switch', field='value',
name='Power',
value_fn=lambda v: bool(v),
write_fn=lambda p, rep, href=None: (
['power', '0'], {'value': p == 'On'})),
),
)
POWER_VS_FALLBACK = Capability(
href='/power/vs/0',
match_fn=lambda rep, resources: '/power/0' not in resources,
entities=(
SwitchDesc(key='power_switch', field='x.com.samsung.da.power',
name='Power',
value_fn=lambda v: v == 'On',
write_fn=lambda p, rep, href=None: (
['power', 'vs', '0'],
{'x.com.samsung.da.power': 'On' if p == 'On' else 'Off'})),
),
)
KIDS_LOCK_GENERIC = Capability(
href='/kidslock/0',
entities=(
SwitchDesc(key='child_lock', field='value',
name='Child lock', device_class='lock',
value_fn=lambda v: bool(v),
write_fn=lambda p, rep, href=None: (
['kidslock', '0'], {'value': p == 'On'})),
),
)
KIDS_LOCK_VS_FALLBACK = Capability(
href='/kidslock/vs/0',
match_fn=lambda rep, resources: '/kidslock/0' not in resources,
entities=(
SwitchDesc(key='child_lock', field='x.com.samsung.da.kidsLock',
name='Child lock', device_class='lock',
value_fn=lambda v: v != 'Ready',
write_fn=lambda p, rep, href=None: (
['kidslock', 'vs', '0'],
{'x.com.samsung.da.kidsLock': 'Enable' if p == 'On' else 'Ready'})),
),
)
REMOTE_CONTROL_GENERIC = Capability(
href='/remotectrl/0',
entities=(
BinarySensorDesc(key='remote_control', field='value',
name='Smart Control', device_class='connectivity',
value_fn=lambda v: bool(v)),
),
)
REMOTE_CONTROL_VS_FALLBACK = Capability(
href='/remotectrl/vs/0',
match_fn=lambda rep, resources: '/remotectrl/0' not in resources,
entities=(
BinarySensorDesc(key='remote_control',
field='x.com.samsung.da.remoteControlEnabled',
name='Smart Control', device_class='connectivity',
value_fn=lambda v: str(v).lower() == 'true'),
),
)
# ---------------------------------------------------------------------------
# /energy/consumption/vs/0 -- washer-specific override of common.ENERGY_METER
# (issue #6).
#
# instantaneousPower is a dead field on every TP1-class washer dump collected
# so far (7 dumps, 3 physical devices): always the literal sentinel '-500',
# unchanged between off/idle-on/mid-cycle states and across different
# courses. common.clamp_power floors that to a misleading "0 W" that reads
# as a real (if idle) measurement rather than "unsupported". Gate the entity
# out entirely when the sentinel is seen, but only then -- if some washer
# model ever reports a real value, this still shows it.
#
# cumulativePower is absent outright on at least one washer model (issue #6),
# unlike every other washer dump. entity.py's generic field-presence gate
# already excludes the entity in that case; the exists_fn here just makes
# that explicit at the capability level instead of relying on the fallback.
# ---------------------------------------------------------------------------
_DEAD_INSTANTANEOUS_POWER = '-500'
WASHER_ENERGY_METER = Capability(
href='/energy/consumption/vs/0',
entities=(
SensorDesc(key='power_watts', field='x.com.samsung.da.instantaneousPower',
name='Power', device_class='power', state_class='measurement',
unit='W', value_fn=clamp_power,
exists_fn=lambda rep, resources: (
rep.get('x.com.samsung.da.instantaneousPower')
!= _DEAD_INSTANTANEOUS_POWER)),
SensorDesc(key='energy_kwh', field='x.com.samsung.da.cumulativePower',
name='Energy', device_class='energy',
state_class='total_increasing', unit='kWh', value_fn=wh_to_kwh,
exists_fn=lambda rep, resources: 'x.com.samsung.da.cumulativePower' in rep),
),
)
@@ -27,7 +27,10 @@ class SamsungEntityDescription:
enabled_default: bool = True
value_fn: Callable[[Any], Any] = _identity
rep_fn: Optional[Callable[[dict], Any]] = None # replaces field+value_fn; receives full rep
exists_fn: Optional[Callable[[dict], bool]] = None
# (rep, resources): rep is this entity's own href's representation;
# resources is the coordinator's full href->rep snapshot, for gating
# presence on a sibling resource (e.g. washer._cycle_options's source).
exists_fn: Optional[Callable[[dict, dict], bool]] = None
@dataclass(frozen=True, kw_only=True)
@@ -35,6 +38,7 @@ class SensorDesc(SamsungEntityDescription):
device_class: Optional[str] = None
state_class: Optional[str] = None
unit: Optional[str] = None
unit_fn: Optional[Callable[[dict], str]] = None # overrides `unit` from the live rep, when set
options: Optional[tuple] = None # required by HA when device_class == 'enum'
@@ -45,7 +49,10 @@ class BinarySensorDesc(SamsungEntityDescription):
@dataclass(frozen=True, kw_only=True)
class SelectDesc(SamsungEntityDescription):
options: Any = () # tuple[str,...] | Callable[[dict], list[str]]
options: Any = () # tuple[str,...] | Callable[[dict[str, dict]], list[str]]
# callable form receives the coordinator's full href->rep resource
# snapshot (not just this entity's own href) and returns raw device
# option values; see select.py's LocalThingsSelect._raw_options().
options_field: Optional[str] = None # resource field that contains the live options list
write_fn: WriteFn = None
@@ -66,6 +73,7 @@ class ButtonDesc(SamsungEntityDescription):
class NumberDesc(SamsungEntityDescription):
device_class: Optional[str] = None
unit: Optional[str] = None
unit_fn: Optional[Callable[[dict], str]] = None # overrides `unit` from the live rep, when set
native_min: Optional[float] = None
native_max: Optional[float] = None
step: Optional[float] = None
+9 -2
View File
@@ -62,11 +62,18 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None:
super().__init__(coordinator, bound)
desc: SelectDesc = bound.desc
if not desc.options_field:
if not desc.options_field and not callable(desc.options):
self._attr_options = [_display(o, desc) for o in desc.options]
def _raw_options(self) -> list[str]:
desc: SelectDesc = self._bound.desc
if callable(desc.options):
# Per-device option list computed from the full resource
# snapshot (not just this entity's own href) -- e.g. a course
# list decoded from a sibling resource. There is no static
# fallback: when that resource isn't populated the callable
# returns [] and the entity's exists_fn suppresses it entirely.
return list(desc.options(self.coordinator.last_resources) or [])
if desc.options_field:
rep = self.coordinator.last_resources.get(self._bound.href) or {}
return list(rep.get(desc.options_field) or [])
@@ -75,7 +82,7 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
@property
def options(self) -> list[str]:
desc: SelectDesc = self._bound.desc
if desc.options_field:
if desc.options_field or callable(desc.options):
return [_display(o, desc) for o in self._raw_options()]
return self._attr_options
+7
View File
@@ -43,6 +43,13 @@ class LocalThingsSensor(LocalThingsEntity, SensorEntity):
if desc.options:
self._attr_options = list(desc.options)
@property
def native_unit_of_measurement(self):
desc: SensorDesc = self._bound.desc
if desc.unit_fn is not None:
return desc.unit_fn(self.coordinator.resource(self._bound.href))
return self._attr_native_unit_of_measurement
@property
def native_value(self):
return (self.coordinator.data or {}).get(self._state_key)
@@ -18,6 +18,7 @@
},
"ice_type": {
"state": {
"off": "Off",
"whiskey_iceball_3": "3 Balls/Day",
"whiskey_iceball_6": "6 Balls/Day",
"whiskey_iceball_9": "9 Balls/Day"
@@ -37,6 +38,69 @@
"cv_ttype_rf9000a_fruit_veggies": "Fruit & Veggies",
"cv_ttype_rf9000a_beverage": "Beverage"
}
},
"dishwasher_cycle": {
"state": {
"0e": "AI Wash",
"07": "Pre blast",
"90": "Self clean",
"86": "Normal",
"83": "Express 60",
"84": "Heavy",
"8d": "Pots and pans",
"80": "Delicate",
"8e": "Plastic",
"8f": "Baby Care"
}
},
"washer_cycle": {
"state": {
"1c": "Eco 40-60",
"1d": "Super Speed",
"21": "Colours",
"1b": "Cotton",
"1e": "15' Quick Wash",
"29": "Drum Clean+",
"24": "Towels",
"33": "Bedding",
"28": "Drain/Spin",
"26": "Delicates",
"27": "Rinse+Spin",
"22": "Wool",
"20": "Hygiene Steam",
"23": "Outdoor",
"25": "Synthetics",
"32": "Shirts",
"2f": "Activewear",
"2e": "Baby Care",
"30": "Cloudy Day",
"66": "Denim",
"2d": "Silent Wash",
"8f": "Intense Cold",
"96": "Less Microfiber"
}
},
"washer_dosing_quantity": {
"state": {
"00": "None",
"01": "Low",
"02": "Medium",
"03": "High"
}
},
"washer_detergent_water_hardness": {
"state": {
"01": "Soft",
"02": "Medium",
"03": "Hard"
}
},
"washer_softener_concentration": {
"state": {
"01": "1x",
"02": "2x",
"03": "3x"
}
}
},
"sensor": {
@@ -51,6 +115,13 @@
"observe": "Push (observe)",
"poll": "Polling"
}
},
"machine_state": {
"state": {
"idle": "Idle",
"active": "Active",
"pause": "Paused"
}
}
}
},
@@ -18,6 +18,7 @@
},
"ice_type": {
"state": {
"off": "Off",
"whiskey_iceball_3": "3 Balls/Day",
"whiskey_iceball_6": "6 Balls/Day",
"whiskey_iceball_9": "9 Balls/Day"
@@ -37,6 +38,69 @@
"cv_ttype_rf9000a_fruit_veggies": "Fruit & Veggies",
"cv_ttype_rf9000a_beverage": "Beverage"
}
},
"dishwasher_cycle": {
"state": {
"0e": "AI Wash",
"07": "Pre blast",
"90": "Self clean",
"86": "Normal",
"83": "Express 60",
"84": "Heavy",
"8d": "Pots and pans",
"80": "Delicate",
"8e": "Plastic",
"8f": "Baby Care"
}
},
"washer_cycle": {
"state": {
"1c": "Eco 40-60",
"1d": "Super Speed",
"21": "Colours",
"1b": "Cotton",
"1e": "15' Quick Wash",
"29": "Drum Clean+",
"24": "Towels",
"33": "Bedding",
"28": "Drain/Spin",
"26": "Delicates",
"27": "Rinse+Spin",
"22": "Wool",
"20": "Hygiene Steam",
"23": "Outdoor",
"25": "Synthetics",
"32": "Shirts",
"2f": "Activewear",
"2e": "Baby Care",
"30": "Cloudy Day",
"66": "Denim",
"2d": "Silent Wash",
"8f": "Intense Cold",
"96": "Less Microfiber"
}
},
"washer_dosing_quantity": {
"state": {
"00": "None",
"01": "Low",
"02": "Medium",
"03": "High"
}
},
"washer_detergent_water_hardness": {
"state": {
"01": "Soft",
"02": "Medium",
"03": "Hard"
}
},
"washer_softener_concentration": {
"state": {
"01": "1x",
"02": "2x",
"03": "3x"
}
}
},
"sensor": {
@@ -51,6 +115,13 @@
"observe": "Push (observe)",
"poll": "Polling"
}
},
"machine_state": {
"state": {
"idle": "Idle",
"active": "Active",
"pause": "Paused"
}
}
}
},
+5
View File
@@ -36,3 +36,8 @@ def dishwasher_resources() -> dict[str, dict]:
@pytest.fixture
def fridge_resources() -> dict[str, dict]:
return _load_device('refrigerator')
@pytest.fixture
def washer_resources() -> dict[str, dict]:
return _load_device('washer')
+2 -2
View File
@@ -10,7 +10,7 @@
"cooler_setpoint",
"cooler_temperature",
"day_brightness",
"defrost_blocked",
"defrost_active",
"defrost_delay",
"door_alert",
"door_cooler_open",
@@ -50,7 +50,7 @@
"samsung_refrigerator_cooler_setpoint",
"samsung_refrigerator_cooler_temperature",
"samsung_refrigerator_day_brightness",
"samsung_refrigerator_defrost_blocked",
"samsung_refrigerator_defrost_active",
"samsung_refrigerator_defrost_delay",
"samsung_refrigerator_door_alert",
"samsung_refrigerator_door_cooler_open",
+28
View File
@@ -0,0 +1,28 @@
{
"state_keys": [
"alarm_code",
"buzzer_sound",
"child_lock",
"cycle",
"cycle_active",
"delay_start_hours",
"detergent_low",
"diagnosis_status",
"drum_clean_cycles_remaining",
"drum_clean_last_cleaned",
"energy_kwh",
"finish_time",
"firmware_update",
"job_beginning_status",
"machine_state",
"power_switch",
"progress",
"progress_percentage",
"remote_control",
"rinse_cycles",
"softener_low",
"spin_speed",
"wash_temperature",
"water_liters"
]
}
+473
View File
@@ -0,0 +1,473 @@
{
"device0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/realtimenotiforclient/vs/0",
"rep": {
"x.com.samsung.da.timeforshortnoti": "0",
"x.com.samsung.da.periodicnotisubscription": "true"
}
},
{
"href": "/alarms/vs/0",
"rep": {}
},
{
"href": "/diagnosis/vs/0",
"rep": {
"x.com.samsung.da.diagnosisStart": "Ready"
}
},
{
"href": "/energy/consumption/vs/0",
"rep": {
"x.com.samsung.da.instantaneousPower": "-500",
"x.com.samsung.da.instantaneousPowerUnit": "W",
"x.com.samsung.da.cumulativePower": "18800",
"x.com.samsung.da.cumulativeUnit": "Wh",
"x.com.samsung.da.cumulativeDate": "1783774800",
"x.com.samsung.da.cumulativeDateUTC": "1783767600",
"x.com.samsung.da.cumulativeSavedPower": "0"
}
},
{
"href": "/energy/consumption/0",
"rep": {}
},
{
"href": "/course/vs/0",
"rep": {
"x.com.samsung.da.supportedModes": [
"HOMECARE_WIZARD_V2"
],
"x.com.samsung.da.options": [
"DeviceType_0167",
"UpdateAllow_NotAllowed",
"Course_1C",
"AiOption_On",
"LaundryOutTime_0",
"SteamWash_Disable",
"SeamlessControl_Disable",
"KidsLockBypass_On",
"WashingTimes_3",
"DrumCleanProposal_40",
"DetergentOnce_1",
"DetergentLeft_0",
"DetergentBase_5",
"DetergentAlarm_Off",
"DetergentType_2",
"DetergentTotal_0",
"SoftenerOnce_0",
"SoftenerLeft_0",
"SoftenerBase_0",
"SoftenerAlarm_Off",
"SoftenerType_0",
"SoftenerTotal_0",
"SpecialFunction_4",
"AvailableDelayTime_163",
"BubbleSoak_Off",
"LaundryPlannerUserSetTime_0",
"ProgressTimeSet_4201048201A4A200D4",
"SendToDevice_Off",
"GMT_04",
"PreWashSetting_Off",
"IntensiveSetting_Off",
"BubbleSoakSet_00F000F00000F000F0F0F0F00000F000F0F0F0F0000000",
"EnergyLevelSet_050304010402010402040402050302020202050303050101",
"MostUsed_1C8410923FA67F00000000000000",
"TextureLevel_None",
"SavingModeCondition_010102151B1D1C1E8F202122232425261C2D30323396662E2F030149035E2830",
"SavingMode_Off",
"DetergentUserSet_0000",
"SoftenerUserSet_0000",
"DetergentSetting_01",
"DetergentSettingList_01",
"SoftenerCustomSetting_01",
"SoftenerCustomSettingList_0102",
"DetergentDefaultCourseList_030422232F2604012E",
"LaundryManageFinish_Off",
"UserLocation_Indoor",
"GeoFenceAlarm",
"UsagesDB_ok",
"EnergyKW_396",
"DrumCleanLog_2026-07-01T20:18:07",
"TimeSync_NotSupported"
],
"x.com.samsung.da.supportedOptions": [
"31C8410923FA67F1B847E923FA67F1E811E933FA33F1D841E923FA67F96841E920FA37F8F8102923FA57F25843E933FA57F26831E920FA20733857E933FA67F24841E930FA30F32833E923FA37F20857E943FA67F22841E920FA30F23831E930FA57F2F831E933FA30F21841E943FA57F66831E933FA30F2E867E943FA67F2D841E923FA30F30843E923FA67F2985209204A520278000913FA67F2880009000A67E"
]
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "On"
}
},
{
"href": "/power/0",
"rep": {
"value": true
}
},
{
"href": "/cycleinterface/vs/0",
"rep": {}
},
{
"href": "/kidslock/vs/0",
"rep": {
"x.com.samsung.da.kidsLock": "Ready"
}
},
{
"href": "/kidslock/0",
"rep": {
"value": false
}
},
{
"href": "/operational/state/vs/0",
"rep": {
"x.com.samsung.da.state": "Ready",
"x.com.samsung.da.remainingTime": "02:43:00",
"x.com.samsung.da.progressPercentage": "1",
"x.com.samsung.da.progress": "None",
"x.com.samsung.da.delayEndTime": "00:00:00",
"x.com.samsung.da.supportedProgress": [
"None",
"Weightsensing",
"Wash",
"Rinse",
"Spin",
"Finish"
]
}
},
{
"href": "/operational/state/0",
"rep": {
"currentMachineState": "**REDACTED**",
"machineStates": "**REDACTED**",
"jobStates": [
"None",
"Weightsensing",
"Wash",
"Rinse",
"Spin",
"Finish"
],
"currentJobState": "None",
"remainingTime": "02:43:00",
"progressPercentage": "1"
}
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "TEST-MODEL",
"x.com.samsung.da.description": "DA_WM_TP1_21_COMMON_WD7000B/DC92-03724A_004D",
"x.com.samsung.da.serialNum": "**REDACTED**",
"x.com.samsung.da.otnDUID": "**REDACTED**",
"x.com.samsung.da.diagProtocolType": "BLE_OCF",
"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": "WFC",
"x.com.samsung.da.diagMinVersion": "3.0",
"x.com.samsung.da.diagTsId": "DA01",
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "DA_WM_TP1_21_COMMON|20375141|20010002001811424AA30217008A0000",
"x.com.samsung.da.type": "Software",
"x.com.samsung.da.number": "02986A260118(A182)",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "1",
"x.com.samsung.da.description": "Firmware_1_DB_20375141240308040FFFFF203724412507152904FFFF(01672037514120372441_30000000)(FileDown:0)(Type:0)",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "03751A24030804,03724A25071529",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Firmware_2_DB_2036284224030602042FFFFFFFFFFFFFFFFFFFFFFFFE(016720362842FFFFFFFF_30000000)(FileDown:0)(Type:0)",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "03628B24030602,FFFFFFFFFFFFFF"
}
]
}
},
{
"href": "/file/information/vs/0",
"rep": {
"x.com.samsung.timeoffset": "+02:00"
}
},
{
"href": "/washer/vs/0",
"rep": {
"x.com.samsung.da.waterTemperature": "40",
"x.com.samsung.da.supportedWaterTemperature": [
"None",
"Cold",
"20",
"30",
"40",
"60",
"90"
],
"x.com.samsung.da.spinLevel": "1400",
"x.com.samsung.da.supportedSpinLevel": [
"RinseHold",
"NoSpin",
"400",
"800",
"1000",
"1200",
"1400"
],
"x.com.samsung.da.rinseCycles": "2",
"x.com.samsung.da.supportedRinseCycles": [
"0",
"1",
"2",
"3",
"4",
"5"
],
"x.com.samsung.da.currentAvailableTemperature": "00",
"x.com.samsung.da.currentAvailableRinse": "3F",
"x.com.samsung.da.currentAvailableSpin": "7F",
"x.com.samsung.da.currentAvailableSubOptions": "00000000000000000000"
}
},
{
"href": "/st/washercourse/vs/0",
"rep": {
"x.com.samsung.da.st.washerMode": "Table_02_Course_1C",
"x.com.samsung.da.st.courseTable": "Table_02"
}
},
{
"href": "/water/consumption/vs/0",
"rep": {
"x.com.samsung.da.cumulativeWater": "1966500"
}
},
{
"href": "/setting/vs/0",
"rep": {
"x.com.samsung.da.supportedSetLanguage": [
"ko_KR",
"en_US"
],
"x.com.samsung.da.setLanguage": "sk_SK"
}
},
{
"href": "/wm/editcourse/vs/0",
"rep": {
"x.com.samsung.da.editCourseList": "EditCourseList_1C1D211B1E29243328262722202325322F2E30662D8F96",
"x.com.samsung.da.fixedCourseList": "FixedCourseList_1C29"
}
},
{
"href": "/wm/setinfo/vs/0",
"rep": {
"x.com.samsung.da.isModelSettingWithoutSC": "true",
"x.com.samsung.da.aiCourse": "true",
"x.com.samsung.da.isModelSettingPowerOnOff": "false",
"x.com.samsung.da.modelCode": "M(None),W(WW90DG6U25LEU4)"
}
},
{
"href": "/wm/jobbeginingstatus/vs/0",
"rep": {
"x.com.samsung.da.currentStatus": "None"
}
},
{
"href": "/wm/welcomemsg/vs/0",
"rep": {}
},
{
"href": "/wm/personalcourse/vs/0",
"rep": {
"x.com.samsung.da.courses": [
"F1_00",
"F2_00",
"F3_00",
"F4_00",
"F5_00",
"F6_00",
"F7_00",
"F8_00",
"F9_00",
"FA_00"
],
"x.com.samsung.da.maxCourseNum": "10",
"x.com.samsung.da.labelScanCourse": "FB_00"
}
},
{
"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": "2026-03-22",
"otnList": [
{
"type": "WIFI",
"modelId": "DA_WM_TP1_21_COMMON",
"versions": [
"30260118"
],
"visVersion": "260118"
},
{
"type": "Micom",
"modelId": "01672037514120372441",
"versions": [
"24030804",
"25071529"
],
"visVersion": "250715"
},
{
"type": "Micom",
"modelId": "016720362842FFFFFFFF",
"versions": [
"24030602",
"FFFFFFFF"
],
"visVersion": "240306"
}
]
}
},
{
"href": "/buzzersound/vs/0",
"rep": {
"supportedBuzzerSound": [
"Off",
"On"
],
"setBuzzerSound": "On"
}
},
{
"href": "/remotectrl/vs/0",
"rep": {
"x.com.samsung.da.remoteControlEnabled": "false"
}
},
{
"href": "/remotectrl/0",
"rep": {
"value": false
}
},
{
"href": "/configuration/vs/0",
"rep": {
"x.com.samsung.da.region": "0000000000",
"x.com.samsung.da.countryCode": "SK"
}
},
{
"href": "/drlc/0",
"rep": {
"DRLevel": 0,
"start": "0000-00-00T00:00:00Z",
"duration": 0,
"override": false
}
},
{
"href": "/drlc/vs/0",
"rep": {
"x.com.samsung.da.drlcLevel": "0",
"x.com.samsung.da.durationminutes": "0",
"x.com.samsung.da.start": "0000-00-00T00:00:00Z",
"x.com.samsung.da.override": "Off",
"x.com.samsung.da.realSaving": "Off"
}
},
{
"href": "/timezone/vs/0",
"rep": {
"timezoneid": "Europe/Bratislava",
"offset": "+02:00",
"DST": "ON"
}
},
{
"href": "/connectionconfig/vs/0",
"rep": {
"autoReconnectionMinVersion": "1.0",
"autoReconnection": "true",
"autoReconnectionProtocolType": [
"helper_hotspot",
"ble_ocf"
],
"supportedWiFiAuthType": [
"OPEN",
"WEP",
"WPA-PSK",
"WPA2-PSK",
"SAE"
],
"supportedWiFiCryptoType": [
"TKIP",
"AES",
"WEP-64",
"WEP-128"
],
"supportedWiFiFreq": [
"2.4G"
],
"calmConnectionCare": {
"version": "1.0",
"role": [
"things"
]
}
}
},
{
"href": "/wirelessinfo/vs/0",
"rep": {
"macaddressWiFi": "**REDACTED**",
"macaddressBLE": "**REDACTED**"
}
},
{
"href": "/quickcontrol/info/vs/0",
"rep": {
"supportedVersion": "1.0"
}
}
]
}
+35
View File
@@ -134,3 +134,38 @@ async def test_duplicate_device_aborted(hass: HomeAssistant, mock_probe) -> None
)
assert result['type'] == FlowResultType.ABORT
assert result['reason'] == 'already_configured'
def test_probe_marks_washer_as_recognized(monkeypatch):
"""A washer's probe response (no oneUiVersion) must still resolve via
the modelNum/description fallback so setup doesn't warn about an
unrecognized device type."""
from custom_components.localthings import config_flow
device0 = [
{'rt': ['x.com.samsung.devcol']},
{'href': '/information/vs/0', 'rep': {
'x.com.samsung.da.modelNum':
'DA_WM_TP1_21_COMMON|20375141|20010002001811424AA30217008A0000',
'x.com.samsung.da.description':
'DA_WM_TP1_21_COMMON_WW5000C/DC92-03495A_B048',
'x.com.samsung.da.serialNum': 'TEST-SERIAL',
}},
{'href': '/otninformation/vs/0', 'rep': {'otnStatus': 'None'}},
]
from custom_components.localthings.registry.batch import parse_device0_batch
resources = parse_device0_batch(device0)
info_resource = resources.get('/information/vs/0', {})
one_ui_version = (
resources.get('/otninformation/vs/0', {}).get('swVersionInfo', {}).get('oneUiVersion', '')
)
from custom_components.localthings.registry.by_type import for_device, for_device_by_model
recognized = bool(
(one_ui_version and for_device(one_ui_version) is not None)
or for_device_by_model(
info_resource.get('x.com.samsung.da.modelNum', ''),
info_resource.get('x.com.samsung.da.description', ''),
) is not None
)
assert recognized is True
+21
View File
@@ -113,6 +113,27 @@ async def test_coverage_gap_issue_absent_for_fully_covered_real_fixture(
assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None
def test_run_discovery_detects_washer_via_model_fallback(
hass: HomeAssistant, mock_entry
) -> None:
"""Washer hardware reports no oneUiVersion; device type must resolve
via for_device_by_model instead of falling back to generic CAPABILITIES."""
resources = {
'/information/vs/0': {
'x.com.samsung.da.modelNum':
'DA_WM_TP1_21_COMMON|20375141|20010002001811424AA30217008A0000',
'x.com.samsung.da.description':
'DA_WM_TP1_21_COMMON_WW5000C/DC92-03495A_B048',
'x.com.samsung.da.serialNum': 'TEST-SERIAL',
},
'/otninformation/vs/0': {'otnStatus': 'None'},
'/power/vs/0': {'x.com.samsung.da.power': 'On'},
}
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._run_discovery(resources)
assert coordinator.device_type_name == 'washer'
def test_update_coverage_gap_issue_creates_issue_for_unknown_type(
hass: HomeAssistant, mock_entry
) -> None:
+8 -1
View File
@@ -1,12 +1,19 @@
"""Tests for the diagnostics platform."""
from __future__ import annotations
import json
from pathlib import Path
from homeassistant.core import HomeAssistant
from custom_components.localthings.const import DOMAIN
from custom_components.localthings.diagnostics import async_get_config_entry_diagnostics
from custom_components.localthings.registry.redact import REDACTED
_MANIFEST_VERSION = json.loads(
(Path(__file__).parents[2] / 'custom_components' / 'localthings' / 'manifest.json').read_text()
)['version']
async def test_diagnostics_shape_and_redaction(
hass: HomeAssistant, mock_entry, mock_coordinator_session
@@ -19,7 +26,7 @@ async def test_diagnostics_shape_and_redaction(
assert diagnostics["device_type"] == 'refrigerator'
assert diagnostics["one_ui_version"] == '7.0 Refrigerator'
assert diagnostics["unbound_hrefs"] == []
assert diagnostics["integration_version"] == '0.1.0'
assert diagnostics["integration_version"] == _MANIFEST_VERSION
assert diagnostics["smartthings_local_version"]
resources = diagnostics["resources"]
+106
View File
@@ -70,3 +70,109 @@ class TestDeviceRegistries:
for cap in caps:
assert cap.rt_filter is not None or cap.match_fn is not None, \
f"href {href!r} has multiple caps but {cap!r} lacks rt_filter and match_fn"
class TestWasherRegistry:
def test_washer_registry_registered(self):
from custom_components.localthings.registry.by_type import _REGISTRY_BY_KEY
assert 'washer' in _REGISTRY_BY_KEY
assert _REGISTRY_BY_KEY['washer'].name == 'washer'
def test_washer_registry_has_no_dup_hrefs(self):
from custom_components.localthings.registry.by_type import _REGISTRY_BY_KEY
registry = _REGISTRY_BY_KEY['washer']
for href, caps in registry.capabilities.items():
if len(caps) > 1:
for cap in caps:
assert cap.rt_filter is not None or cap.match_fn is not None, \
f"href {href!r} has multiple caps but {cap!r} lacks rt_filter and match_fn"
def test_washer_registry_covers_known_hrefs(self):
from custom_components.localthings.registry.by_type import _REGISTRY_BY_KEY
registry = _REGISTRY_BY_KEY['washer']
for href in (
'/power/0', '/power/vs/0', '/kidslock/0', '/kidslock/vs/0',
'/remotectrl/0', '/remotectrl/vs/0', '/alarms/vs/0',
'/energy/consumption/vs/0', '/water/consumption/vs/0',
'/operational/state/vs/0', '/washer/vs/0', '/course/vs/0',
'/buzzersound/vs/0', '/wm/jobbeginingstatus/vs/0',
'/diagnosis/vs/0', '/otninformation/vs/0',
):
assert href in registry.capabilities, f"{href} missing from washer registry"
class TestForDeviceByModel:
"""Fallback device-type detection for hardware without oneUiVersion."""
def test_washer_ww_prefix(self):
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'DA_WM_TP1_21_COMMON|20375141|20010002001811424AA30217008A0000',
'DA_WM_TP1_21_COMMON_WW5000C/DC92-03495A_B048',
)
assert reg is not None
assert reg.name == 'washer'
def test_washer_wd_prefix(self):
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'DA_WM_TP1_21_COMMON|20375141|20010002001811424AA30217008A0000',
'DA_WM_TP1_21_COMMON_WD7000B/DC92-03724A_004D',
)
assert reg is not None
assert reg.name == 'washer'
def test_dryer_not_misdetected_as_washer(self):
"""Dryer shares the DA_WM_ board prefix with washer -- must not
be misrouted despite the shared prefix."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'DA_WM_TP2_20_COMMON_DV5000T', 'DA_WM_TP2_20_COMMON_DV5000T',
)
assert reg is not None
assert reg.name == 'dryer'
def test_dishwasher_not_misdetected_as_washer(self):
"""Dishwasher's modelNum contains the substring 'WW' -- must not
be misrouted by a naive substring match."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'ADW-WW-RTL-24-AILITE|90000541|400002010019130059C1000500E10000',
'ADW-WW-RTL-24-AILITE_DW9000F/DD91-00002A_0002',
)
assert reg is not None
assert reg.name == 'dishwasher'
def test_refrigerator_via_modelnum_ref_token(self):
"""Refrigerator's description has no consumer-model suffix; falls
back to the '_REF_' token in modelNum."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'TP1X_REF_21K|00176141|00000850031813294103010041030000',
'TP1X_REF_21K',
)
assert reg is not None
assert reg.name == 'refrigerator'
def test_refrigerator_rl_series_via_ref_token(self):
"""Issue #7: RL38C6B0CWW/EG (a bottom-freezer RL-series fridge, not
the RF9000-style french-door this module was originally verified
against) reports description/modelNum 'TP1X_REF_21K' -- same
internal platform code as any other TP1X-based fridge, so the
existing '_REF_' fallback already resolves it correctly."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'TP1X_REF_21K|00156941|00050126001611304100000031010000',
'TP1X_REF_21K',
)
assert reg is not None
assert reg.name == 'refrigerator'
def test_unknown_model_returns_none(self):
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model('SOME-UNKNOWN-BOARD', 'SOME-UNKNOWN-BOARD')
assert reg is None
def test_empty_inputs_return_none(self):
from custom_components.localthings.registry.by_type import for_device_by_model
assert for_device_by_model('', '') is None
+51
View File
@@ -0,0 +1,51 @@
"""Tests for dishwasher-specific capabilities."""
from custom_components.localthings.registry.capabilities import dishwasher
class TestCycleOptions:
def test_parses_edit_course_list(self):
raw = 'EditCourseList_0E07908683848D808E8F'
assert dishwasher._parse_edit_course_list(raw) == [
'0E', '07', '90', '86', '83', '84', '8D', '80', '8E', '8F',
]
def test_parse_edit_course_list_handles_missing_or_malformed(self):
assert dishwasher._parse_edit_course_list(None) == []
assert dishwasher._parse_edit_course_list('') == []
assert dishwasher._parse_edit_course_list('no underscore') == []
def test_cycle_options_reads_live_edit_course_list(self):
resources = {
'/wm/editcourse/vs/0': {
'x.com.samsung.da.editCourseList': 'EditCourseList_0E9086',
},
}
assert dishwasher._cycle_options(resources) == ['0E', '90', '86']
def test_cycle_options_empty_when_resource_absent(self):
assert dishwasher._cycle_options({}) == []
def test_cycle_options_empty_when_resource_empty(self):
resources = {'/wm/editcourse/vs/0': {}}
assert dishwasher._cycle_options(resources) == []
def test_cycle_desc_uses_cycle_options_callable(self):
desc = next(e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == 'cycle')
assert desc.options is dishwasher._cycle_options
assert desc.translation_key == 'dishwasher_cycle'
def test_exists_only_when_edit_course_list_is_live(self):
desc = next(e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == 'cycle')
assert desc.exists_fn({}, {}) is False
assert desc.exists_fn({}, {'/wm/editcourse/vs/0': {}}) is False
live = {'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_0E'}}
assert desc.exists_fn({}, live) is True
def test_cycle_write_uses_raw_code_directly(self):
desc = next(e for e in dishwasher.CYCLE_OPTIONS.entities if e.key == 'cycle')
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'],
}
+104
View File
@@ -0,0 +1,104 @@
"""Tests for fridge-specific capabilities."""
from custom_components.localthings.registry.capabilities import fridge
class TestTempCurrentGeneric:
"""Issue #7: unit must be read live from the device, not assumed to be
Fahrenheit -- a TP1X_REF_21K fridge (RL38C6B0CWW/EG) reports the same
resource in Celsius."""
def test_unit_reads_celsius(self):
desc = fridge.TEMP_CURRENT_GENERIC.entities[0]
assert desc.unit_fn({'temperature': 3.0, 'units': 'C'}) == '°C'
def test_unit_reads_fahrenheit(self):
desc = fridge.TEMP_CURRENT_GENERIC.entities[0]
assert desc.unit_fn({'temperature': 5.0, 'units': 'F'}) == '°F'
def test_unit_defaults_to_fahrenheit_when_missing(self):
desc = fridge.TEMP_CURRENT_GENERIC.entities[0]
assert desc.unit_fn({'temperature': 5.0}) == '°F'
class TestTempSetpointGeneric:
def test_unit_reads_celsius(self):
desc = fridge.TEMP_SETPOINT_GENERIC.entities[0]
assert desc.unit_fn({'temperature': -19.0, 'units': 'C'}) == '°C'
class TestTemperaturesFallback:
"""Aggregate /temperatures/vs/0 -- per-item unit from
x.com.samsung.da.unit ('Celsius'/'Fahrenheit')."""
def test_freezer_unit_celsius(self):
desc = next(e for e in fridge.TEMPERATURES_FALLBACK.entities
if e.key == 'freezer_temperature')
rep = {'x.com.samsung.da.items': [
{'x.com.samsung.da.description': 'Freezer', 'x.com.samsung.da.current': '-19',
'x.com.samsung.da.unit': 'Celsius'},
{'x.com.samsung.da.description': 'Fridge', 'x.com.samsung.da.current': '3',
'x.com.samsung.da.unit': 'Celsius'},
]}
assert desc.unit_fn(rep) == '°C'
assert desc.value_fn(rep['x.com.samsung.da.items']) == -19
def test_fridge_unit_fahrenheit(self):
desc = next(e for e in fridge.TEMPERATURES_FALLBACK.entities
if e.key == 'fridge_temperature')
rep = {'x.com.samsung.da.items': [
{'x.com.samsung.da.description': 'Fridge', 'x.com.samsung.da.current': '37',
'x.com.samsung.da.unit': 'Fahrenheit'},
]}
assert desc.unit_fn(rep) == '°F'
def test_unit_defaults_to_fahrenheit_when_item_missing(self):
desc = next(e for e in fridge.TEMPERATURES_FALLBACK.entities
if e.key == 'freezer_temperature')
assert desc.unit_fn({'x.com.samsung.da.items': []}) == '°F'
class TestDefrostBlockStatus:
"""DEFROST_BLOCK_ON means the defrost cycle is actively running, not
that defrost is being withheld -- confirmed against live dumps showing
it ON while defrost_delay is off."""
def test_key_and_name_reflect_active_defrosting(self):
desc = fridge.DEFROST_BLOCK_STATUS.entities[0]
assert desc.key == 'defrost_active'
def test_value_fn(self):
desc = fridge.DEFROST_BLOCK_STATUS.entities[0]
assert desc.value_fn(['DEFROST_BLOCK_ON']) is True
assert desc.value_fn(['DEFROST_BLOCK_OFF']) is False
class TestRefrigerationFallback:
"""/refrigeration/0 duplicates two of REFRIGERATION's (/refrigeration/vs/0)
fields under different names (rapidFreeze/rapidCool vs
rapidFreezing/rapidFridge) but also carries one genuinely new field
(defrost) that has no vs-href equivalent."""
def test_href(self):
assert fridge.REFRIGERATION_FALLBACK.href == '/refrigeration/0'
def test_defrost_active_is_fallback_of_defrost_block_status(self):
# Duplicates DEFROST_BLOCK_STATUS's defrost_active
# (/defrost/block/vs/0) -- only a true fallback when that richer
# href is absent (issue #7's device: /refrigeration/0 only).
desc = next(e for e in fridge.REFRIGERATION_FALLBACK.entities
if e.key == 'defrost_active')
assert desc.value_fn(True) is True
assert desc.value_fn(False) is False
assert desc.exists_fn({}, {'/refrigeration/0': {}}) is True
assert desc.exists_fn(
{}, {'/refrigeration/0': {}, '/defrost/block/vs/0': {}}) is False
def test_rapid_switches_hidden_when_vs_href_present(self):
for key in ('rapid_fridge', 'rapid_freezing'):
desc = next(e for e in fridge.REFRIGERATION_FALLBACK.entities if e.key == key)
assert desc.exists_fn({}, {'/refrigeration/vs/0': {}, '/refrigeration/0': {}}) is False
def test_rapid_switches_shown_when_vs_href_absent(self):
for key in ('rapid_fridge', 'rapid_freezing'):
desc = next(e for e in fridge.REFRIGERATION_FALLBACK.entities if e.key == key)
assert desc.exists_fn({}, {'/refrigeration/0': {}}) is True
+19 -1
View File
@@ -7,12 +7,18 @@ GOLDEN = Path(__file__).parent / 'fixtures' / 'golden'
def _new_state_keys(name, resources):
from custom_components.localthings.registry.by_type import for_device
from custom_components.localthings.registry.by_type import for_device, for_device_by_model
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.adapter import flatten
otn = resources.get('/otninformation/vs/0', {})
one_ui = otn.get('swVersionInfo', {}).get('oneUiVersion', '')
info = resources.get('/information/vs/0', {})
reg = for_device(one_ui) if one_ui else None
if reg is None:
reg = for_device_by_model(
info.get('x.com.samsung.da.modelNum', ''),
info.get('x.com.samsung.da.description', ''),
)
if reg is None:
from custom_components.localthings.registry.registry import CAPABILITIES
caps, pats = CAPABILITIES, []
@@ -39,6 +45,18 @@ def test_registry_reproduces_golden_state_keys(name, ip, request):
)
def test_registry_reproduces_golden_state_keys_for_washer():
from tests.conftest import _load_device
resources = _load_device('washer')
golden = json.loads((GOLDEN / 'washer.json').read_text())
state_keys = _new_state_keys('washer', 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_resources_from_batch_preferred_over_flat():
from tests.conftest import _resources_from_dump
dump = {
+48
View File
@@ -6,3 +6,51 @@ def test_machine_state_maps_samsung_to_ocf():
assert ms.value_fn('Run') == 'active'
assert ms.value_fn('Pause') == 'pause'
assert ms.value_fn('Ready') == 'idle'
class TestProgressPercentage:
"""issue #9: device firmware leaves progressPercentage stale (e.g. '1')
after a cycle ends instead of resetting it, so it must be gated on
active state the same way `progress`/`cycle_active`/`finish_time` are."""
def test_zeroed_when_not_active(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'progress_percentage')
rep = {'x.com.samsung.da.state': 'Ready', 'x.com.samsung.da.progressPercentage': '1'}
assert desc.rep_fn(rep) == 0
def test_passes_through_when_active(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'progress_percentage')
rep = {'x.com.samsung.da.state': 'Run', 'x.com.samsung.da.progressPercentage': '42'}
assert desc.rep_fn(rep) == 42
class TestDelayFieldFallback:
def test_reads_delay_end_time_when_delay_start_time_absent(self):
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'delay_start_hours')
rep = {'x.com.samsung.da.delayEndTime': '02:30:00'}
assert desc.rep_fn(rep) == 2.5
def test_prefers_delay_start_time_when_both_present(self):
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'delay_start_hours')
rep = {
'x.com.samsung.da.delayStartTime': '01:00:00',
'x.com.samsung.da.delayEndTime': '02:00:00',
}
assert desc.rep_fn(rep) == 1.0
def test_write_targets_delay_end_time_when_that_is_what_device_reports(self):
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'delay_start_hours')
rep = {'x.com.samsung.da.delayEndTime': '00:00:00'}
path, body = desc.write_fn(1.5, rep)
assert path == ['operational', 'state', 'vs', '0']
assert body == {'x.com.samsung.da.delayEndTime': '1:30:00'}
def test_write_targets_delay_start_time_by_default(self):
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'delay_start_hours')
rep = {'x.com.samsung.da.delayStartTime': '00:00:00'}
path, body = desc.write_fn(1.5, rep)
assert body == {'x.com.samsung.da.delayStartTime': '1:30:00'}
+16
View File
@@ -57,3 +57,19 @@ def test_dishwasher_fixture_has_no_remaining_gaps():
resources = _load('dishwasher_device.json')
unbound = set(_unbound_hrefs(resources, dishwasher.REGISTRY))
assert unbound == set()
class TestWasherIgnoredHrefs:
def test_washer_hrefs_are_ignored(self):
from custom_components.localthings.registry.capabilities.ignored import IGNORED
ignored_hrefs = {cap.href for cap in IGNORED}
for href in (
'/st/washercourse/vs/0',
'/wm/welcomemsg/vs/0',
'/wm/personalcourse/vs/0',
'/energy/consumption/0',
'/cycleinterface/vs/0',
'/drlc/0',
'/operational/state/0',
):
assert href in ignored_hrefs, f"{href} should be in ignored.IGNORED"
+59
View File
@@ -0,0 +1,59 @@
"""Tests for LocalThingsSelect's option-list resolution
(custom_components/localthings/select.py) -- the static tuple, options_field,
and callable forms of SelectDesc.options.
"""
from custom_components.localthings.registry.capability import Capability
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import SelectDesc
from custom_components.localthings.select import LocalThingsSelect
class _FakeCoordinator:
device_serial = 'TEST-SERIAL'
def __init__(self, last_resources):
self.last_resources = last_resources
def _make_select(desc, href, last_resources):
capability = Capability(href=href, entities=(desc,))
bound = BoundEntity(href=href, capability=capability, desc=desc)
return LocalThingsSelect(_FakeCoordinator(last_resources), bound)
def test_static_options_unaffected():
desc = SelectDesc(key='x', options=('A', 'B'))
entity = _make_select(desc, '/x/vs/0', {})
assert entity.options == ['A', 'B']
def test_options_field_unaffected():
desc = SelectDesc(key='x', options_field='supported')
entity = _make_select(desc, '/x/vs/0', {'/x/vs/0': {'supported': ['Lo', 'Hi']}})
assert entity.options == ['Lo', 'Hi']
def test_callable_options_receives_full_resource_snapshot():
"""A callable options is handed the coordinator's full href->rep
snapshot, not just this entity's own href's rep -- needed for course
lists decoded from a sibling resource (see washer._cycle_options)."""
calls = []
def _options_fn(resources):
calls.append(resources)
return list(resources.get('/other/vs/0', {}).get('codes', []))
desc = SelectDesc(key='cycle', translation_key='fake_cycle', options=_options_fn)
resources = {
'/x/vs/0': {},
'/other/vs/0': {'codes': ['1C', '1D']},
}
entity = _make_select(desc, '/x/vs/0', resources)
assert entity.options == ['1c', '1d']
assert calls == [resources]
def test_callable_options_empty_result():
desc = SelectDesc(key='cycle', options=lambda resources: [])
entity = _make_select(desc, '/x/vs/0', {})
assert entity.options == []
+349
View File
@@ -0,0 +1,349 @@
"""Tests for washer-specific capabilities."""
from custom_components.localthings.registry.capabilities import washer
def _rep_by_href(cap, href):
assert cap.href == href
return cap
class TestWasherSettings:
def test_href(self):
assert washer.WASHER_SETTINGS.href == '/washer/vs/0'
def test_wash_temperature_read(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == 'wash_temperature')
assert desc.field == 'x.com.samsung.da.waterTemperature'
assert desc.options_field == 'x.com.samsung.da.supportedWaterTemperature'
def test_wash_temperature_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == 'wash_temperature')
path, body = desc.write_fn('60', {})
assert path == ['washer', 'vs', '0']
assert body == {'x.com.samsung.da.waterTemperature': '60'}
def test_spin_speed_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == 'spin_speed')
path, body = desc.write_fn('1400', {})
assert path == ['washer', 'vs', '0']
assert body == {'x.com.samsung.da.spinLevel': '1400'}
def test_rinse_cycles_write(self):
desc = next(e for e in washer.WASHER_SETTINGS.entities if e.key == 'rinse_cycles')
path, body = desc.write_fn('3', {})
assert path == ['washer', 'vs', '0']
assert body == {'x.com.samsung.da.rinseCycles': '3'}
class TestWasherCourse:
def test_href(self):
assert washer.WASHER_COURSE.href == '/course/vs/0'
def test_translation_key(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
assert desc.translation_key == 'washer_cycle'
def test_reads_raw_course_code_from_options_array(self):
"""rep_fn returns the raw device code; display names come from
strings.json via translation_key, not from Python (see select.py's
_display())."""
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']}
assert desc.rep_fn(rep) == '1C'
def test_missing_course_option_returns_none(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
assert desc.rep_fn({'x.com.samsung.da.options': ['GMT_04']}) is None
def test_cycle_desc_uses_cycle_options_callable(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
assert desc.options is washer._cycle_options
def test_exists_only_when_edit_course_list_is_live(self):
"""No hardcoded course table is kept -- the selector only appears
when a device actually populates editCourseList (see
_cycle_options's docstring for why MostUsed_ isn't used either)."""
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
assert desc.exists_fn({}, {}) is False
assert desc.exists_fn({}, {'/wm/editcourse/vs/0': {}}) is False
live = {'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_1C'}}
assert desc.exists_fn({}, live) is True
def test_cycle_write(self):
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'],
}
class TestCycleOptions:
def test_parses_edit_course_list(self):
raw = 'EditCourseList_1C1D211B1E29243328262722202325322F2E30662D8F96'
assert washer._parse_edit_course_list(raw) == [
'1C', '1D', '21', '1B', '1E', '29', '24', '33', '28', '26', '27',
'22', '20', '23', '25', '32', '2F', '2E', '30', '66', '2D', '8F', '96',
]
def test_parse_edit_course_list_handles_missing_or_malformed(self):
assert washer._parse_edit_course_list(None) == []
assert washer._parse_edit_course_list('') == []
assert washer._parse_edit_course_list('no underscore') == []
def test_cycle_options_reads_live_edit_course_list(self):
"""A different washer model's own course list -- including a code
('65') never seen on the primary dump -- is used as-is; there is
no hardcoded table to fall back to or reconcile against."""
resources = {
'/wm/editcourse/vs/0': {
'x.com.samsung.da.editCourseList': 'EditCourseList_651C',
},
}
assert washer._cycle_options(resources) == ['65', '1C']
def test_cycle_options_empty_when_resource_absent(self):
assert washer._cycle_options({}) == []
def test_cycle_options_empty_when_resource_empty(self):
"""The second known washer dump has /wm/editcourse/vs/0 == {}."""
resources = {'/wm/editcourse/vs/0': {}}
assert washer._cycle_options(resources) == []
class TestDrumClean:
def test_cycles_remaining(self):
"""DrumCleanProposal_40 - WashingTimes_3 == 37, matching a live
app screenshot's 'Potreba cistenia po 37 cykloch'."""
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_cycles_remaining')
rep = {'x.com.samsung.da.options': ['WashingTimes_3', 'DrumCleanProposal_40']}
assert desc.rep_fn(rep) == 37
def test_cycles_remaining_never_negative(self):
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_cycles_remaining')
rep = {'x.com.samsung.da.options': ['WashingTimes_50', 'DrumCleanProposal_40']}
assert desc.rep_fn(rep) == 0
def test_cycles_remaining_missing_fields(self):
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_cycles_remaining')
assert desc.rep_fn({'x.com.samsung.da.options': []}) is None
def test_cycles_remaining_exists_only_when_computable(self):
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_cycles_remaining')
assert desc.exists_fn({'x.com.samsung.da.options': []}, {}) is False
rep = {'x.com.samsung.da.options': ['WashingTimes_3', 'DrumCleanProposal_40']}
assert desc.exists_fn(rep, {}) is True
def test_last_cleaned(self):
"""DrumCleanLog_2026-07-01T20:18:07 -> a UTC-aware datetime,
matching the same screenshot's '10 days ago' (as of 2026-07-11)."""
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_last_cleaned')
rep = {'x.com.samsung.da.options': ['DrumCleanLog_2026-07-01T20:18:07']}
from datetime import datetime, timezone
assert desc.rep_fn(rep) == datetime(2026, 7, 1, 20, 18, 7, tzinfo=timezone.utc)
def test_last_cleaned_missing(self):
desc = next(e for e in washer.WASHER_COURSE.entities
if e.key == 'drum_clean_last_cleaned')
assert desc.rep_fn({'x.com.samsung.da.options': []}) is None
assert desc.exists_fn({'x.com.samsung.da.options': []}, {}) is False
# Raw options array from issue #9's diagnostic dump (trimmed to the fields
# relevant to detergent/softener dosing).
_DOSING_OPTIONS = [
'Course_1C',
'DetergentAlarm_Off',
'SoftenerAlarm_Off',
'DetergentLevelCtrl_3',
'SoftenerLevelCtrl_3',
'SupportedDetergentLevelCtrl_00010203',
'SupportedSoftenerLevelCtrl_00010203',
'DetergentLevel2Ctrl_2',
'SoftenerLevel2Ctrl_2',
'SupportedDetergentLevel2Ctrl_010203',
'SupportedSoftenerLevel2Ctrl_010203',
]
_DOSING_RESOURCES = {'/course/vs/0': {'x.com.samsung.da.options': _DOSING_OPTIONS}}
class TestDetergentSoftenerDosing:
@staticmethod
def _desc(key):
return next(e for e in washer.WASHER_COURSE.entities if e.key == key)
def test_quantity_and_hardness_read(self):
rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
assert self._desc('detergent_quantity').rep_fn(rep) == '3'
assert self._desc('detergent_water_hardness').rep_fn(rep) == '2'
assert self._desc('softener_quantity').rep_fn(rep) == '3'
assert self._desc('softener_concentration').rep_fn(rep) == '2'
def test_translation_keys(self):
"""detergent_quantity and softener_quantity share one translation_key
(same 00-03 -> None/Low/Medium/High vocabulary on both dispensers,
same shape as fridge.py's shared 'brightness_level' key); hardness
and concentration each have their own since their labels differ."""
assert self._desc('detergent_quantity').translation_key == 'washer_dosing_quantity'
assert self._desc('softener_quantity').translation_key == 'washer_dosing_quantity'
assert self._desc('detergent_water_hardness').translation_key == 'washer_detergent_water_hardness'
assert self._desc('softener_concentration').translation_key == 'washer_softener_concentration'
def test_quantity_and_hardness_options_decode_supported_list(self):
assert self._desc('detergent_quantity').options(_DOSING_RESOURCES) == ['00', '01', '02', '03']
assert self._desc('softener_quantity').options(_DOSING_RESOURCES) == ['00', '01', '02', '03']
assert self._desc('detergent_water_hardness').options(_DOSING_RESOURCES) == ['01', '02', '03']
assert self._desc('softener_concentration').options(_DOSING_RESOURCES) == ['01', '02', '03']
def test_exists_only_when_supported_list_present(self):
for key in ('detergent_quantity', 'detergent_water_hardness',
'softener_quantity', 'softener_concentration'):
desc = self._desc(key)
assert desc.exists_fn({}, {}) is False
assert desc.exists_fn({}, _DOSING_RESOURCES) is True
def test_quantity_write(self):
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_01' in body['x.com.samsung.da.options']
assert 'DetergentLevelCtrl_3' not in body['x.com.samsung.da.options']
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_03' in body['x.com.samsung.da.options']
def test_low_reservoir_off_when_alarm_off(self):
rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
assert self._desc('detergent_low').rep_fn(rep) is False
assert self._desc('softener_low').rep_fn(rep) is False
def test_low_reservoir_on_when_alarm_active(self):
opts = ['DetergentAlarm_On', 'SoftenerAlarm_On']
rep = {'x.com.samsung.da.options': opts}
assert self._desc('detergent_low').rep_fn(rep) is True
assert self._desc('softener_low').rep_fn(rep) is True
def test_low_reservoir_exists_only_when_alarm_field_present(self):
assert self._desc('detergent_low').exists_fn({'x.com.samsung.da.options': []}, {}) is False
rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
assert self._desc('detergent_low').exists_fn(rep, {}) is True
class TestBuzzerSound:
def test_href(self):
assert washer.BUZZER_SOUND.href == '/buzzersound/vs/0'
def test_buzzer_sound_write(self):
desc = next(e for e in washer.BUZZER_SOUND.entities if e.key == 'buzzer_sound')
path, body = desc.write_fn('Volume_High', {})
assert path == ['buzzersound', 'vs', '0']
assert body == {'setBuzzerSound': 'Volume_High'}
def test_finish_sound_exists_only_when_supported(self):
desc = next(e for e in washer.BUZZER_SOUND.entities if e.key == 'finish_sound')
assert desc.exists_fn({'setBuzzerSound': 'On'}, {}) is False
assert desc.exists_fn({'supportedFinishSound': ['FinishSound_1']}, {}) is True
def test_finish_sound_write(self):
desc = next(e for e in washer.BUZZER_SOUND.entities if e.key == 'finish_sound')
path, body = desc.write_fn('FinishSound_2', {})
assert path == ['buzzersound', 'vs', '0']
assert body == {'setFinishSound': 'FinishSound_2'}
class TestJobBeginningStatus:
def test_href_and_field(self):
assert washer.WASHER_JOB_BEGINNING_STATUS.href == '/wm/jobbeginingstatus/vs/0'
desc = washer.WASHER_JOB_BEGINNING_STATUS.entities[0]
assert desc.field == 'x.com.samsung.da.currentStatus'
class TestPowerFallback:
def test_generic_href(self):
assert washer.POWER_GENERIC.href == '/power/0'
def test_generic_read_write(self):
desc = washer.POWER_GENERIC.entities[0]
assert desc.value_fn(True) is True
assert desc.value_fn(False) is False
path, body = desc.write_fn('On', {})
assert path == ['power', '0']
assert body == {'value': True}
path, body = desc.write_fn('Off', {})
assert body == {'value': False}
def test_vs_fallback_binds_only_when_generic_absent(self):
assert washer.POWER_VS_FALLBACK.href == '/power/vs/0'
assert washer.POWER_VS_FALLBACK.match_fn({}, {'/power/vs/0': {}}) is True
assert washer.POWER_VS_FALLBACK.match_fn({}, {'/power/0': {}, '/power/vs/0': {}}) is False
def test_vs_fallback_read_write(self):
desc = washer.POWER_VS_FALLBACK.entities[0]
assert desc.value_fn('On') is True
assert desc.value_fn('Off') is False
path, body = desc.write_fn('On', {})
assert path == ['power', 'vs', '0']
assert body == {'x.com.samsung.da.power': 'On'}
class TestKidsLockFallback:
def test_generic_read_write(self):
assert washer.KIDS_LOCK_GENERIC.href == '/kidslock/0'
desc = washer.KIDS_LOCK_GENERIC.entities[0]
assert desc.value_fn(True) is True
path, body = desc.write_fn('On', {})
assert path == ['kidslock', '0']
assert body == {'value': True}
def test_vs_fallback_gated(self):
assert washer.KIDS_LOCK_VS_FALLBACK.match_fn({}, {'/kidslock/vs/0': {}}) is True
assert washer.KIDS_LOCK_VS_FALLBACK.match_fn(
{}, {'/kidslock/0': {}, '/kidslock/vs/0': {}}) is False
class TestRemoteControlFallback:
def test_generic_read(self):
assert washer.REMOTE_CONTROL_GENERIC.href == '/remotectrl/0'
desc = washer.REMOTE_CONTROL_GENERIC.entities[0]
assert desc.value_fn(True) is True
assert desc.value_fn(False) is False
def test_vs_fallback_gated(self):
assert washer.REMOTE_CONTROL_VS_FALLBACK.match_fn({}, {'/remotectrl/vs/0': {}}) is True
assert washer.REMOTE_CONTROL_VS_FALLBACK.match_fn(
{}, {'/remotectrl/0': {}, '/remotectrl/vs/0': {}}) is False
class TestWasherEnergyMeter:
"""Issue #6: instantaneousPower is a dead sentinel ('-500', unchanged
across off/idle-on/running-eco/running-fabrics states and across 3
physical devices) on every TP1-class washer dump collected so far;
cumulativePower is outright absent on at least one washer model."""
def test_href(self):
assert washer.WASHER_ENERGY_METER.href == '/energy/consumption/vs/0'
def test_power_watts_hidden_for_dead_sentinel(self):
desc = next(e for e in washer.WASHER_ENERGY_METER.entities if e.key == 'power_watts')
assert desc.exists_fn({'x.com.samsung.da.instantaneousPower': '-500'}, {}) is False
def test_power_watts_shown_for_real_value(self):
desc = next(e for e in washer.WASHER_ENERGY_METER.entities if e.key == 'power_watts')
assert desc.exists_fn({'x.com.samsung.da.instantaneousPower': '150'}, {}) is True
def test_energy_kwh_hidden_when_cumulative_power_absent(self):
desc = next(e for e in washer.WASHER_ENERGY_METER.entities if e.key == 'energy_kwh')
assert desc.exists_fn({'x.com.samsung.da.instantaneousPower': '-500'}, {}) is False
def test_energy_kwh_shown_when_present(self):
desc = next(e for e in washer.WASHER_ENERGY_METER.entities if e.key == 'energy_kwh')
assert desc.exists_fn({'x.com.samsung.da.cumulativePower': '58900'}, {}) is True