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

feat: enhance operational capabilities by adding completion_minutes and updating related logic; remove completion_time references

Browse Source
This commit is contained in:
Jeroen Hof
2026-07-24 15:35:17 +02:00
parent 7be91b56e8
commit 365a1c4722
8 changed files with 95 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/registry/capabilities/operational.py
+71 -61
View File
@@ -2,7 +2,8 @@
Shared by dryer/dishwasher/oven/washer families.
"""
from datetime import datetime, timezone, timedelta
import math
from datetime import datetime, timedelta, timezone
from ..capability import Capability
from ..entities import BinarySensorDesc, ButtonDesc, NumberDesc, SensorDesc
@@ -28,17 +29,41 @@ def _int(v):
return None
def _is_active(rep):
"""Check if appliance is actively running and cycle is not finished."""
return (
_SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
and rep.get('x.com.samsung.da.progress') != 'Finish'
)
def _remaining_seconds(raw):
"""'HH:MM:SS' (or 'MM:SS') -> total seconds, or None."""
if not isinstance(raw, str):
return None
try:
parts = [int(p) for p in raw.split(':')]
except (ValueError, TypeError):
return None
if len(parts) == 3:
h, m, s = parts
elif len(parts) == 2:
h, m, s = 0, *parts
else:
return None
return h * 3600 + m * 60 + s
def _delay_hours(v):
"""delayStartTime is a duration until the cycle starts, not a
wall-clock time -- "01:00" means "1 hour from when you press start",
not "1 AM"."""
if not v:
return 0.0
try:
h, m, s = v.split(':')
return int(h) + int(m) / 60 + int(s) / 3600
except Exception:
return None
total_seconds = _remaining_seconds(v)
if total_seconds is None:
return 0.0 if not v else None
return total_seconds / 3600.0
def _format_delay(hours):
@@ -53,59 +78,46 @@ def _delay_field(rep):
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
try:
h, m, s = remaining_str.split(':')
total_s = int(h) * 3600 + int(m) * 60 + int(s)
if total_s == 0:
return None
return datetime.now(timezone.utc) + timedelta(seconds=total_s)
except Exception:
return None
def _parse_remaining_time(rep):
return (
rep.get('x.com.samsung.da.remainingTime')
or rep.get('remainingTime')
'x.com.samsung.da.delayEndTime' if 'x.com.samsung.da.delayEndTime' in rep
else 'x.com.samsung.da.delayStartTime'
)
def _completion_time(rep):
return _parse_remaining_time(rep)
def _finish_time(rep):
if not _is_active(rep):
return None
total_s = _remaining_seconds(rep.get('x.com.samsung.da.remainingTime'))
if not total_s:
return None
return datetime.now(timezone.utc) + timedelta(seconds=total_s)
def _completion_minutes(rep):
raw = _parse_remaining_time(rep)
if not raw or not isinstance(raw, str):
"""Parse remaining time into minutes directly from device payload."""
raw = rep.get('x.com.samsung.da.remainingTime') or rep.get('remainingTime')
total_s = _remaining_seconds(raw)
if total_s is None:
return None
try:
parts = [int(p) for p in raw.split(':')]
if len(parts) == 3:
h, m, s = parts
return h * 60 + m + (1 if s > 0 else 0)
elif len(parts) == 2:
m, s = parts
return m + (1 if s > 0 else 0)
except (ValueError, TypeError):
pass
return None
# Avoid the firmware bug where it freezes at 1 minute post-cycle
if rep.get('x.com.samsung.da.progress') == 'Finish':
return 0
return math.ceil(total_s / 60)
# Shared by dryer/dishwasher/oven/washer -- oven.py imports this directly
# rather than keeping its own copy, since both wrote the identical
# state='Ready' RMW.
STOP_BUTTON = ButtonDesc(key='stop', field='', name='Stop', payload='Ready',
icon='mdi:stop',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p}))
STOP_BUTTON = ButtonDesc(
key='stop', field='', name='Stop', payload='Ready',
icon='mdi:stop',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p}
)
)
OPERATIONAL_STATE = Capability(
href='/operational/state/vs/0',
@@ -115,6 +127,7 @@ OPERATIONAL_STATE = Capability(
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.
@@ -126,40 +139,34 @@ OPERATIONAL_STATE = Capability(
# vocabulary that doesn't fit an oven's bake/roast/etc.
BinarySensorDesc(key='cycle_active', device_class='running',
name='Running',
rep_fn=lambda rep: (
_SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
and rep.get('x.com.samsung.da.progress') != 'Finish'
)),
rep_fn=_is_active),
SensorDesc(key='progress', name='Progress', icon='mdi:progress-wrench',
rep_fn=lambda rep: (
'Idle' if _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) != 'active'
else _progress(rep.get('x.com.samsung.da.progress'))
)),
SensorDesc(key='progress_percentage',
name='Progress percent', unit='%', state_class='measurement',
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'.
SensorDesc(key='finish_time', device_class='timestamp',
name='Estimated finish',
rep_fn=lambda rep: (
None if _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) != 'active'
or rep.get('x.com.samsung.da.progress') == 'Finish'
else _finish_time(rep.get('x.com.samsung.da.remainingTime'))
)),
SensorDesc(key='completion_time', name='Completion time',
icon='mdi:timer-sand',
exists_fn=lambda rep, resources: _completion_time(rep) is not None,
rep_fn=_completion_time),
rep_fn=_finish_time),
SensorDesc(key='completion_minutes', name='Remaining minutes',
icon='mdi:clock-outline', unit='min',
device_class='duration', state_class='measurement',
exists_fn=lambda rep, resources: _completion_minutes(rep) is not None,
rep_fn=_completion_minutes),
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,
@@ -169,16 +176,19 @@ OPERATIONAL_STATE = Capability(
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{_delay_field(rep): _format_delay(p)})),
ButtonDesc(key='start', field='', name='Start', payload='Run',
icon='mdi:play',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p})),
ButtonDesc(key='pause', field='', name='Pause', payload='Pause',
icon='mdi:pause',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p})),
STOP_BUTTON,
),
)
tests/fixtures/golden/dishwasher.json
Vendored
-1
View File
@@ -4,7 +4,6 @@
"auto_release_dry",
"child_lock",
"completion_minutes",
"completion_time",
"cycle",
"cycle_active",
"delay_start_hours",
tests/fixtures/golden/dryer.json
Vendored
-1
View File
@@ -4,7 +4,6 @@
"buzzer_sound",
"child_lock",
"completion_minutes",
"completion_time",
"cycle",
"cycle_active",
"delay_start_hours",
tests/fixtures/golden/washer.json
Vendored
-1
View File
@@ -6,7 +6,6 @@
"buzzer_sound",
"child_lock",
"completion_minutes",
"completion_time",
"cycle",
"cycle_active",
"delay_start_hours",
tests/fixtures/golden/washer_dryer_combo.json
Vendored
-1
View File
@@ -4,7 +4,6 @@
"bubble_soak",
"child_lock",
"completion_minutes",
"completion_time",
"cycle",
"cycle_active",
"delay_start_hours",
tests/fixtures/golden/washer_flexwash.json
Vendored
-1
View File
@@ -3,7 +3,6 @@
"alarm_code",
"child_lock",
"completion_minutes",
"completion_time",
"cycle",
"cycle_active",
"delay_start_hours",
tests/test_laundry_capabilities.py
+2 -2
View File
@@ -4,11 +4,11 @@ from custom_components.localthings.registry.capabilities import laundry
class TestCourseHelpers:
def test_parses_edit_course_list(self):
raw = 'EditCourseList_1C1D211B1E29243328262722202325322F2E30662D8F962b2a'
raw = 'EditCourseList_1C1D211B1E29243328262722202325322F2E30662D8F962B2A'
assert laundry.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',
'2b', '2a'
'2B', '2A'
]
def test_parse_edit_course_list_handles_missing_or_malformed(self):
tests/test_operational_capability.py
+22 -17
View File
@@ -1,3 +1,5 @@
"""Unit tests for operational state capabilities."""
from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
@@ -24,18 +26,8 @@ class TestProgressPercentage:
assert desc.rep_fn(rep) == 42
class TestCompletionTime:
"""Unit tests for completion_time and completion_minutes entities."""
def test_completion_time_string(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_time')
rep = {'x.com.samsung.da.remainingTime': '01:25:30'}
assert desc.rep_fn(rep) == '01:25:30'
def test_completion_time_fallback(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_time')
rep = {'remainingTime': '00:45:00'}
assert desc.rep_fn(rep) == '00:45:00'
class TestCompletionMinutes:
"""Unit tests for completion_minutes entity."""
def test_completion_minutes_parsing(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_minutes')
@@ -48,13 +40,26 @@ class TestCompletionTime:
rep_exact = {'x.com.samsung.da.remainingTime': '01:30:00'}
assert desc.rep_fn(rep_exact) == 90
def test_completion_time_missing_or_invalid(self):
desc_time = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_time')
desc_min = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_minutes')
def test_completion_minutes_fallback_key(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_minutes')
rep = {'remainingTime': '00:45:00'}
assert desc.rep_fn(rep) == 45
def test_completion_minutes_stale_finish_gated(self):
"""Firmware freezes remainingTime at '00:01:00' when progress reaches 'Finish'.
Should return 0 to prevent stuck values."""
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_minutes')
rep = {
'x.com.samsung.da.progress': 'Finish',
'x.com.samsung.da.remainingTime': '00:01:00',
}
assert desc.rep_fn(rep) == 0
def test_completion_minutes_missing_or_invalid(self):
desc = next(e for e in OPERATIONAL_STATE.entities if e.key == 'completion_minutes')
rep = {}
assert desc_time.rep_fn(rep) is None
assert desc_min.rep_fn(rep) is None
assert desc.rep_fn(rep) is None
class TestDelayFieldFallback: