feat: add operational-state capability with extrapolation hooks

samsung_appliance/registry/capabilities/__init__.py

@@ -1,11 +1,10 @@
+from . import common, operational
 from ..capability import Capability
-from . import common
 
 
 def _is_capability(v):
     return isinstance(v, Capability)
 
 
-ALL = [
-    *[v for v in vars(common).values() if _is_capability(v)],
-]
+ALL = [v for mod in (common, operational)
+       for v in vars(mod).values() if _is_capability(v)]

samsung_appliance/registry/capabilities/operational.py

@@ -0,0 +1,94 @@
+"""Operational-state capability: machine state, progress, remaining-time.
+
+Carries the rare cross-field hooks (active_when, on_observation, project) for
+remaining-time extrapolation. Shared by dryer/dishwasher/oven/washer families.
+"""
+import time
+
+from ..capability import Capability
+from ..entities import SensorDesc
+
+_SAMSUNG_STATE_TO_OCF = {
+    'Ready': 'idle', 'Run': 'active', 'Running': 'active',
+    'Pause': 'pause', 'Paused': 'pause', 'End': 'idle', 'Stop': 'idle',
+}
+
+
+def _to_ocf(v):
+    return _SAMSUNG_STATE_TO_OCF.get(v, v) if v is not None else None
+
+
+def _progress(v):
+    return 'Idle' if v in (None, 'None') else v
+
+
+def _int(v):
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _active_when(rep):
+    return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
+
+
+def _on_observation(state, rep):
+    rem = rep.get('x.com.samsung.da.remainingTime')
+    if not isinstance(rem, str):
+        return
+    try:
+        h, m, s = rem.split(':')
+        state['remaining_anchor'] = (time.time(),
+                                     int(h) * 3600 + int(m) * 60 + int(s))
+    except (ValueError, AttributeError):
+        pass
+
+
+def _project(state, sensors):
+    anchor = state.get('remaining_anchor')
+    if sensors.get('machine_state') != 'active' or anchor is None:
+        return sensors
+    ts, total = anchor
+    remaining = max(0, int(total - (time.time() - ts)))
+    h, rest = divmod(remaining, 3600)
+    m, s = divmod(rest, 60)
+    sensors = dict(sensors)
+    sensors['completion_time'] = f"{h}:{m:02d}:{s:02d}"
+    sensors['completion_minutes'] = h * 60 + m + (1 if s > 0 else 0)
+    return sensors
+
+
+def _rem_minutes(remaining):
+    if not remaining:
+        return None
+    try:
+        h, m, s = remaining.split(':')
+        return int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
+    except Exception:
+        return None
+
+
+OPERATIONAL_STATE = Capability(
+    href='/operational/state/vs/0',
+    poll_tier='hot',
+    entities=(
+        SensorDesc(key='machine_state', field='x.com.samsung.da.state',
+                   name='Machine state', value_fn=_to_ocf),
+        SensorDesc(key='progress', field='x.com.samsung.da.progress',
+                   name='Progress', icon='mdi:progress-wrench', value_fn=_progress),
+        SensorDesc(key='progress_percentage',
+                   field='x.com.samsung.da.progressPercentage',
+                   name='Progress percent', unit='%', state_class='measurement',
+                   value_fn=_int),
+        SensorDesc(key='completion_time', field='x.com.samsung.da.remainingTime',
+                   name='Completion time', icon='mdi:timer-sand'),
+        SensorDesc(key='completion_minutes', field='x.com.samsung.da.remainingTime',
+                   name='Remaining minutes', unit='min', device_class='duration',
+                   state_class='measurement',
+                   value_fn=lambda r: _rem_minutes(r)),
+    ),
+    active_when=_active_when,
+    on_observation=_on_observation,
+    project=_project,
+)

tests/test_operational_capability.py

@@ -0,0 +1,22 @@
+import time
+from samsung_appliance.registry.capabilities.operational import OPERATIONAL_STATE
+
+
+def test_machine_state_maps_samsung_to_ocf():
+    ms = next(e for e in OPERATIONAL_STATE.entities if e.key == 'machine_state')
+    assert ms.value_fn('Run') == 'active'
+    assert ms.value_fn('Pause') == 'pause'
+    assert ms.value_fn('Ready') == 'idle'
+
+
+def test_active_when_true_only_when_running():
+    assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Run'}) is True
+    assert OPERATIONAL_STATE.active_when({'x.com.samsung.da.state': 'Ready'}) is False
+
+
+def test_project_extrapolates_remaining_time():
+    state = {}
+    OPERATIONAL_STATE.on_observation(state, {'x.com.samsung.da.remainingTime': '1:00:00'})
+    assert 'remaining_anchor' in state
+    out = OPERATIONAL_STATE.project(state, {'machine_state': 'active', 'completion_minutes': 60})
+    assert out['completion_minutes'] <= 60