Merge pull request #4 from mbillow/washer-drum-clean

feat(washer): expose Drum Clean+ maintenance status

custom_components/localthings/registry/capabilities/washer.py

@@ -6,6 +6,8 @@ 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
 
@@ -114,6 +116,38 @@ def _cycle_write(p, rep, href=None):
     }
 
 
+# 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
+
+
 WASHER_COURSE = Capability(
     href='/course/vs/0',
     entities=(
@@ -124,6 +158,16 @@ WASHER_COURSE = Capability(
                    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),
     ),
 )
 

tests/fixtures/golden/washer.json

@@ -7,6 +7,8 @@
     "cycle_active",
     "delay_start_hours",
     "diagnosis_status",
+    "drum_clean_cycles_remaining",
+    "drum_clean_last_cleaned",
     "energy_kwh",
     "finish_time",
     "firmware_update",

tests/test_washer_capabilities.py

@@ -112,6 +112,49 @@ class TestCycleOptions:
         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
+
+
 class TestBuzzerSound:
     def test_href(self):
         assert washer.BUZZER_SOUND.href == '/buzzersound/vs/0'