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

refactor: discover entities at connect; delete appliance descriptor files

Browse Source 
PushBridge no longer takes a descriptor at construction time.  Instead it
discovers entities from the live device cache immediately after the seed
(via registry.discover + build_runtime_descriptor), publishes HA discovery
payloads, wires on_observation, and builds command handlers — all in the
new idempotent _discover_and_publish(sess) method called from
_run_session_inner.

Changes:
- bridge.py: remove descriptor param; self.descriptor=None until first
  connect; SEED_PATH constant replaces self.descriptor.seed_path; all
  self.descriptor accesses guarded or moved post-discovery; inlined
  _bridge_diagnostic_discovery (was in appliances/base.py); subscribe to
  observe paths now happens after discovery (seed → discover → subscribe).
- config.py: klass is now optional (defaults to ''); topic_prefix/name
  defaults handle empty klass.
- main.py: drop get_descriptor import and pairs loop; build bridges as
  PushBridge(shared, app, cli); on_connect no longer publishes
  discovery_payloads (bridge self-publishes at discovery time).
- Delete samsung_appliance/appliances/ entirely (6 files, ~2600 lines).
This commit is contained in:
Marc Billow
2026-06-30 16:32:42 -05:00
parent 35222e6be5
commit cd146ed2f0
10 changed files with 191 additions and 2678 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+6 -20
View File
@@ -23,7 +23,6 @@ import threading
import paho.mqtt.client as mqtt
from samsung_appliance.appliances import get_descriptor
from samsung_appliance.bridge import PushBridge
from samsung_appliance.config import SharedConfig, load_appliances
from samsung_appliance.logger import logger
@@ -46,26 +45,15 @@ def main():
logger.error("client cert/key not found: %s", path)
return 2
# Resolve descriptors up front — bad APPLIANCE_<n>_CLASS should fail
# at startup, not 10s into the first DTLS attempt.
pairs = []
for app in appliances:
try:
desc = get_descriptor(app.klass)
except ValueError as e:
logger.error("APPLIANCE_%d_CLASS: %s", app.index, e)
return 2
pairs.append((app, desc))
logger.info("SmartThings-Local Bridge starting (%d appliance%s)",
len(appliances), '' if len(appliances) == 1 else 's')
logger.info(" broker = %s:%d (user=%s)",
shared.MQTT_BROKER, shared.MQTT_PORT,
shared.MQTT_USER or '<anon>')
for app, desc in pairs:
port = app.ocf_port or desc.default_observe_port
for app in appliances:
port = app.ocf_port or 49154
logger.info(" [%d] %s @ %s:%d (DTLS) → topic %s/*",
app.index, app.klass, app.ip, port, app.topic_prefix)
app.index, app.klass or '?', app.ip, port, app.topic_prefix)
# --- MQTT client (shared) ---
cli = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
@@ -85,9 +73,9 @@ def main():
cli.will_set(f"{first_app.topic_prefix}/availability",
payload='offline', qos=1, retain=True)
# Build bridges. Each builds its own discovery payloads in __init__.
bridges: list[PushBridge] = [PushBridge(shared, app, desc, cli)
for app, desc in pairs]
# Build bridges. Each will discover entities at first connect.
bridges: list[PushBridge] = [PushBridge(shared, app, cli)
for app in appliances]
by_prefix = {b.cmd_topic_prefix.rstrip('/'): b for b in bridges}
def on_connect(client, userdata, flags, rc, props=None):
@@ -97,8 +85,6 @@ def main():
logger.info("MQTT connected → %s:%d",
shared.MQTT_BROKER, shared.MQTT_PORT)
for b in bridges:
for topic, payload in b.discovery_payloads:
client.publish(topic, payload, qos=1, retain=True)
cmd_wildcard = f"{b.app.topic_prefix}/cmd/#"
client.subscribe(cmd_wildcard, qos=1)
b.reassert_availability()
samsung_appliance/__init__.py
+2 -2
View File
@@ -1,5 +1,5 @@
"""Samsung appliance local-API → MQTT bridge with HA discovery.
Multi-device: dryer, oven, etc. The device class is selected at startup
via the DEVICE_CLASS env var (see samsung_appliance.appliances)."""
Multi-device: dryer, oven, etc. Entities are discovered at connect-time
from the live device resources via samsung_appliance.registry."""
__version__ = "2.0.0"
samsung_appliance/appliances/__init__.py
-36
View File
@@ -1,36 +0,0 @@
"""Appliance descriptors registry.
Adding a new appliance class:
1. Write `appliances/<class>.py` with an ApplianceDescriptor named
after the class (uppercase, e.g. OVEN).
2. Add it to DESCRIPTORS below.
3. Set DEVICE_CLASS=<class> in the per-device .env.
main.py imports get_descriptor(name) to look up the descriptor at
startup; the bridge itself stays class-agnostic.
"""
from .base import ApplianceDescriptor
from .dishwasher import DISHWASHER
from .dryer import DRYER
from .oven import OVEN
from .refrigerator import REFRIGERATOR
DESCRIPTORS: dict[str, ApplianceDescriptor] = {
DISHWASHER.name: DISHWASHER,
DRYER.name: DRYER,
OVEN.name: OVEN,
REFRIGERATOR.name: REFRIGERATOR,
}
def get_descriptor(name: str) -> ApplianceDescriptor:
try:
return DESCRIPTORS[name]
except KeyError:
raise ValueError(
f"unknown DEVICE_CLASS={name!r}; "
f"available: {sorted(DESCRIPTORS)}") from None
__all__ = ['ApplianceDescriptor', 'DESCRIPTORS', 'get_descriptor']
samsung_appliance/appliances/base.py
-228
View File
@@ -1,228 +0,0 @@
"""ApplianceDescriptor — the per-device-class abstraction.
The bridge is appliance-class-agnostic: it owns the DTLS session, the
OBSERVE-token bookkeeping, and MQTT publish gating. Each appliance
class (dryer, oven, …) provides a descriptor that supplies:
* observe_paths — which CoAP resources to OBSERVE
* seed_path — the resource to fetch on connect (usually
/device/0) to populate the link dict
* flatten(links) — links → flat-dict that lands on MQTT
* build_discovery(…) — list of (HA-discovery topic, payload)
* command_handlers() — MQTT command-suffix → (path_segs, body_dict)
Optional hooks let an appliance hold transient state across pushes:
* on_observation(state, href, rep) — capture anchors, e.g. for
time extrapolation
* project(state, sensors) — fill in extrapolated fields
on publish
state is a free-form dict the bridge owns and threads into both hooks.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Callable, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from ..poll_scheduler import PollTier
@dataclass
class ApplianceDescriptor:
"""Static description of a Samsung appliance class.
`name` is the value DEVICE_CLASS resolves to (e.g. 'dryer').
`default_observe_port` documents the UDP port this firmware
exposes its DTLS-CoAP endpoint on, for the .env templates and
docs — config.APPLIANCE_OCF_PORT still wins at runtime."""
name: str
default_observe_port: int
observe_paths: list[list[str]]
seed_path: list[str]
flatten: Callable[[dict], dict]
build_discovery: Callable[[str, str, str], list[tuple[str, bytes]]]
# command_handlers() returns {topic_suffix: fn(payload_str, links_snapshot)}
# where the fn returns (path_segs, body_dict) | None. Handlers receive
# a snapshot of the bridge's link dict so they can do read-modify-write
# on resources like `/mode/vs/0` options or `/temperatures/vs/0` items.
command_handlers: Callable[[], dict[str, Callable[[str, dict], Optional[tuple]]]]
# Optional behavioural hooks. state is a mutable dict the bridge
# threads in; the descriptor decides what keys to put in it.
on_observation: Optional[Callable[[dict, str, dict], None]] = None
project: Optional[Callable[[dict, dict], dict]] = None
# If set, the bridge maintains a second availability topic
# `<prefix>/remote_available` derived from sensors[remote_field].
# HA entities that the appliance only honours with Remote-Control
# enabled gate themselves on it.
remote_available_field: Optional[str] = None
# If set, the bridge maintains a third availability topic
# `<prefix>/cycle_active` derived from sensors[cycle_active_field]
# (boolean). HA entities that the appliance only accepts writes for
# while a cycle is running (e.g. oven setpoint, cook time, options
# toggles) gate themselves on it.
cycle_active_field: Optional[str] = None
# Optional log-line callback for state-change notifications. Gets
# the freshly-projected sensors dict; returns a short string.
log_state_change: Optional[Callable[[dict], str]] = None
# Tiered polling cadence. Hot tier resources are the user-visible
# state that needs sub-second freshness; warm/cold/sweep cover the
# rest. Empirically measured per-device ceilings in
# local-tools/probe_poll_rate_combined.py (dryer ~14 req/s, oven
# ~8 req/s) inform the defaults each descriptor sets.
poll_tiers: list['PollTier'] = field(default_factory=list)
# Predicate the PollScheduler calls each tick to decide whether to
# use a tier's active_interval_s. Gets a shallow snapshot of the
# link dict so it can read whichever resource indicates activity.
is_active: Optional[Callable[[dict], bool]] = None
# --- HA-discovery helpers ----------------------------------------------
# Pure builder fns used by descriptor build_discovery() implementations.
# Kept here so the per-appliance modules stay focused on their entity
# inventory.
def device_block(topic_prefix: str, device_name: str,
model: str) -> dict:
return {
'identifiers': [topic_prefix],
'name': device_name,
'manufacturer': 'Samsung',
'model': model,
}
def avail_base(avail_topic: str) -> list[dict]:
return [{'topic': avail_topic,
'payload_available': 'online',
'payload_not_available': 'offline'}]
def avail_with_remote(avail_topic: str,
remote_topic: str) -> list[dict]:
return [
{'topic': avail_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
{'topic': remote_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
]
def avail_with_cycle(avail_topic: str,
cycle_topic: str) -> list[dict]:
return [
{'topic': avail_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
{'topic': cycle_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
]
def avail_with_remote_and_cycle(avail_topic: str,
remote_topic: str,
cycle_topic: str) -> list[dict]:
return [
{'topic': avail_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
{'topic': remote_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
{'topic': cycle_topic,
'payload_available': 'online',
'payload_not_available': 'offline'},
]
def encode(cfg: dict) -> bytes:
return json.dumps(cfg).encode()
def bridge_diagnostic_discovery(topic_prefix: str,
ha_discovery_prefix: str,
device_name: str,
model: str = 'Bridge') -> list[tuple[str, bytes]]:
"""HA-discovery payloads for the per-bridge diagnostic entities
(push state, polling RTT, freshness). Returned as a list of
(topic, payload-bytes) tuples to be merged with the descriptor's
own build_discovery output."""
avail_topic = f"{topic_prefix}/availability"
health_topic = f"{topic_prefix}/bridge/health"
push_topic = f"{topic_prefix}/bridge/push_active"
device = device_block(topic_prefix, device_name, model)
avail = avail_base(avail_topic)
def sensor(slug: str, name: str, value_template: str,
unit: str | None = None, device_class: str | None = None,
icon: str | None = None) -> tuple[str, bytes]:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_bridge_{slug}",
'object_id': f"{topic_prefix}_bridge_{slug}",
'state_topic': health_topic,
'value_template': value_template,
'availability': avail,
'device': device,
'entity_category': 'diagnostic',
}
if unit is not None: cfg['unit_of_measurement'] = unit
if device_class is not None: cfg['device_class'] = device_class
if icon is not None: cfg['icon'] = icon
topic = (f"{ha_discovery_prefix}/sensor/"
f"{topic_prefix}/bridge_{slug}/config")
return topic, encode(cfg)
push_active_cfg = {
'name': 'Push Active',
'unique_id': f"{topic_prefix}_bridge_push_active",
'object_id': f"{topic_prefix}_bridge_push_active",
'state_topic': push_topic,
'payload_on': 'online',
'payload_off': 'offline',
'availability': avail,
'device': device,
'entity_category': 'diagnostic',
'icon': 'mdi:rss',
}
push_active_topic = (f"{ha_discovery_prefix}/binary_sensor/"
f"{topic_prefix}/bridge_push_active/config")
return [
(push_active_topic, encode(push_active_cfg)),
sensor('update_source', 'Last Update Source',
"{{ value_json.last_change_source | default('?') }}",
icon='mdi:transit-connection-variant'),
sensor('stalest_age_s', 'Stalest Resource Age',
"{{ value_json.stalest_age_s | default(0) }}",
unit='s', icon='mdi:clock-alert-outline'),
sensor('poll_max_rtt_ms', 'Poll Max RTT',
"{{ value_json.poll_window_max_rtt_ms | default(0) | int }}",
unit='ms', icon='mdi:speedometer'),
sensor('slow_polls', 'Slow Polls (window)',
"{{ value_json.poll_window_slow_count | default(0) }}",
icon='mdi:timer-sand'),
sensor('poll_timeouts', 'Poll Timeouts (window)',
"{{ value_json.poll_window_timeout_count | default(0) }}",
icon='mdi:timer-off-outline'),
sensor('poll_errors', 'Poll Errors (window)',
"{{ value_json.poll_window_errors | default(0) }}",
icon='mdi:alert-circle-outline'),
sensor('observe_age_s', 'Last OBSERVE Age',
"{{ value_json.last_observe_age_s if value_json.last_observe_age_s is not none else 'never' }}",
icon='mdi:radar'),
]
samsung_appliance/appliances/dishwasher.py
-452
View File
@@ -1,452 +0,0 @@
"""Dishwasher descriptor (Samsung DW9000F-class / ADW-WW-RTL-24-AILITE).
Resource map captured 2026-06-27 via DTLS-CoAP with the ab0b0ac4 cert
from a DW90F89T0USRAA at 10.0.0.129:49154.
See local-tools/dumps/10.0.0.129.json for the full field reference.
Write surfaces exposed by this descriptor:
modelled on dryer/operational-state write semantics (same firmware family):
* Start / Pause / Stop via /operational/state/vs/0
— requires Remote Control on.
modelled on oven option-write semantics (unproven, first HA use is the test):
* Sound mode select via /settings/sound/mode/vs/0
* Door LED brightness select via /doorled/light/vs/0
* Door LED night light switch via /doorled/light/vs/0
Course selection is not yet exposed — the course encoding in
/wm/editcourse/vs/0 uses an opaque hex string (EditCourseList_…) whose
byte-to-course-name mapping requires the same capture process as the
dryer's COURSE_NAMES table. Capture with the dishwasher running each
course and compare /course/vs/0 + /dishwasher/vs/0 before/after.
"""
import time
from .base import (
ApplianceDescriptor,
avail_base,
avail_with_remote,
device_block,
encode,
)
from ..poll_scheduler import PollTier
# --- OBSERVE paths ---------------------------------------------------------
# Only /<x>/vs/0 siblings push notifications. /course/vs/0 and
# /dishwasher/vs/0 were included in the device tree but returned no
# parseable state fields in the 2026-06-27 capture — include them so
# future firmware pushes surface through the cache.
OBSERVE_PATHS = [
['operational', 'state', 'vs', '0'], # state, progress, remainingTime
['power', 'vs', '0'], # power On/Off
['kidslock', 'vs', '0'], # child lock
['remotectrl', 'vs', '0'], # remote control enabled
['energy', 'consumption', 'vs', '0'], # watts + cumulative Wh
['water', 'consumption', 'vs', '0'], # cumulative water (mL)
['alarms', 'vs', '0'],
['course', 'vs', '0'],
['dishwasher', 'vs', '0'],
['doorled', 'light', 'vs', '0'], # door LED brightness + night light
['settings', 'sound', 'mode', 'vs', '0'],
['wm', 'jobbeginingstatus', 'vs', '0'],
]
_SAMSUNG_STATE_TO_OCF = {
'Ready': 'idle',
'Run': 'active',
'Running': 'active',
'Pause': 'pause',
'Paused': 'pause',
'End': 'idle',
'Stop': 'idle',
}
def _num(v):
try:
return float(v)
except (TypeError, ValueError):
return None
def _int(v):
try:
return int(v)
except (TypeError, ValueError):
return None
# --- flatten ---------------------------------------------------------------
def flatten(links):
g = lambda href, k, default=None: (links.get(href) or {}).get(k, default)
sam_state = g('/operational/state/vs/0', 'x.com.samsung.da.state')
machine_state = (_SAMSUNG_STATE_TO_OCF.get(sam_state, sam_state)
if sam_state is not None else None)
progress = g('/operational/state/vs/0', 'x.com.samsung.da.progress')
if progress in (None, 'None'):
progress = 'Idle'
remaining = g('/operational/state/vs/0', 'x.com.samsung.da.remainingTime')
rem_min = None
if remaining:
try:
h, m, s = remaining.split(':')
rem_min = int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
except Exception:
pass
delay_start = g('/operational/state/vs/0', 'x.com.samsung.da.delayStartTime')
# Energy — same phantom -500W idle issue as the dryer.
inst_w = _num(g('/energy/consumption/vs/0', 'x.com.samsung.da.instantaneousPower'))
cum_wh = _num(g('/energy/consumption/vs/0', 'x.com.samsung.da.cumulativePower'))
if inst_w is not None and inst_w < 0:
inst_w = 0.0
# Water — cumulative in mL, exposed as liters.
cum_water_ml = _num(g('/water/consumption/vs/0', 'x.com.samsung.da.cumulativeWater'))
cum_water_l = round(cum_water_ml / 1000.0, 1) if cum_water_ml is not None else None
sam_power = g('/power/vs/0', 'x.com.samsung.da.power')
sam_kids = g('/kidslock/vs/0', 'x.com.samsung.da.kidsLock')
sam_rc = g('/remotectrl/vs/0', 'x.com.samsung.da.remoteControlEnabled')
power_bin = (sam_power == 'On') if sam_power is not None else None
kids_bin = (sam_kids != 'Ready') if sam_kids is not None else None
rc_bin = (str(sam_rc).lower() == 'true') if sam_rc is not None else None
filter_usage = _int(g('/filter/waterfilter/vs/0', 'x.com.samsung.da.filterUsage'))
filter_status = g('/filter/waterfilter/vs/0', 'x.com.samsung.da.filterStatus')
alarm_items = g('/alarms/vs/0', 'x.com.samsung.da.items') or []
alarm_code = (alarm_items[-1].get('x.com.samsung.da.code')
if alarm_items else None)
# Door LED — keys have no x.com.samsung.da. prefix on this appliance.
led_rep = links.get('/doorled/light/vs/0') or {}
led_brightness = led_rep.get('setBrightness')
led_night_light = led_rep.get('setNightLight')
sound_rep = links.get('/settings/sound/mode/vs/0') or {}
sound_mode = sound_rep.get('mode')
return {
'machine_state': machine_state,
'progress': progress,
'progress_percentage': _int(g('/operational/state/vs/0',
'x.com.samsung.da.progressPercentage')),
'completion_time': remaining,
'completion_minutes': rem_min,
'delay_start_time': delay_start,
'power_state': sam_power,
'power_state_binary': power_bin,
'child_lock': sam_kids,
'child_lock_binary': kids_bin,
'remote_control': sam_rc,
'remote_control_binary': rc_bin,
'power_watts': inst_w,
'energy_kwh': round(cum_wh / 1000.0, 2)
if cum_wh is not None else None,
'water_liters': cum_water_l,
'filter_usage': filter_usage,
'filter_status': filter_status,
'alarm_code': alarm_code,
'led_brightness': led_brightness,
'led_night_light': led_night_light,
'sound_mode': sound_mode,
}
# --- Remaining-time anchor + extrapolation --------------------------------
def on_observation(state, href, rep):
if href != '/operational/state/vs/0':
return
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 log_state_change(sensors):
return (f"machine={sensors.get('machine_state')} "
f"progress={sensors.get('progress')} "
f"power={sensors.get('power_watts')}W")
# --- HA discovery ---------------------------------------------------------
MODEL = 'OCF dishwasher (TizenRT-iotivity, DW9000F-class)'
_SENSORS = [
('machine_state', 'Machine state', {'icon': 'mdi:dishwasher'}),
('progress', 'Progress', {'icon': 'mdi:progress-wrench'}),
('progress_percentage', 'Progress percent',
{'unit_of_measurement': '%', 'state_class': 'measurement'}),
('completion_time', 'Completion time', {'icon': 'mdi:timer-sand'}),
('completion_minutes', 'Remaining minutes',
{'unit_of_measurement': 'min', 'device_class': 'duration',
'state_class': 'measurement'}),
('delay_start_time', 'Delay start time', {'icon': 'mdi:timer-pause'}),
('power_state', 'Power state', {}),
('power_watts', 'Power',
{'unit_of_measurement': 'W', 'device_class': 'power',
'state_class': 'measurement'}),
('energy_kwh', 'Energy',
{'unit_of_measurement': 'kWh', 'device_class': 'energy',
'state_class': 'total_increasing'}),
('water_liters', 'Water consumption',
{'unit_of_measurement': 'L', 'device_class': 'water',
'state_class': 'total_increasing', 'icon': 'mdi:water'}),
('filter_usage', 'Filter usage',
{'unit_of_measurement': '%', 'state_class': 'measurement',
'icon': 'mdi:filter'}),
('filter_status', 'Filter status', {'icon': 'mdi:filter-check'}),
('alarm_code', 'Alarm code',
{'icon': 'mdi:alert', 'entity_category': 'diagnostic'}),
('sound_mode', 'Sound mode', {'icon': 'mdi:volume-high'}),
]
_BINARY_SENSORS = [
('running', 'Running',
"{{ 'ON' if value_json.machine_state == 'active' else 'OFF' }}",
'running'),
('child_lock_active', 'Child lock',
"{{ 'ON' if value_json.child_lock_binary else 'OFF' }}",
'lock'),
('remote_control_enabled', 'Remote control',
"{{ 'ON' if value_json.remote_control_binary else 'OFF' }}",
'connectivity'),
]
CMD_OPERATIONAL = 'cmd/operational_state'
CMD_SOUND_MODE = 'cmd/sound_mode'
CMD_LED_BRIGHT = 'cmd/led_brightness'
CMD_LED_NIGHT = 'cmd/led_night_light'
_SOUND_MODES = ['voice', 'tone', 'mute']
_LED_LEVELS = ['Low', 'High']
def build_discovery(topic_prefix, ha_prefix, device_name):
state_topic = f"{topic_prefix}/state"
avail_topic = f"{topic_prefix}/availability"
remote_topic = f"{topic_prefix}/remote_available"
dev = device_block(topic_prefix, device_name, MODEL)
out = []
for key, name, extra in _SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': f"{{{{ value_json.{key} }}}}",
'availability': avail_base(avail_topic),
'device': dev,
}
cfg.update(extra)
out.append((f"{ha_prefix}/sensor/{topic_prefix}/{key}/config",
encode(cfg)))
for key, name, template, dclass in _BINARY_SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': template,
'payload_on': 'ON',
'payload_off': 'OFF',
'device_class': dclass,
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/binary_sensor/{topic_prefix}/{key}/config",
encode(cfg)))
# buttons: Start / Pause / Stop (RC-gated)
for key, name, payload, icon in [
('start', 'Start cycle', 'Run', 'mdi:play'),
('pause', 'Pause cycle', 'Pause', 'mdi:pause'),
('stop', 'Stop cycle', 'Ready', 'mdi:stop'),
]:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'command_topic': f"{topic_prefix}/{CMD_OPERATIONAL}",
'payload_press': payload,
'icon': icon,
'availability': avail_with_remote(avail_topic, remote_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/button/{topic_prefix}/{key}/config",
encode(cfg)))
# select: sound mode
cfg = {
'name': 'Sound mode',
'unique_id': f"{topic_prefix}_sound_mode_select",
'object_id': f"{topic_prefix}_sound_mode_select",
'state_topic': state_topic,
'value_template': '{{ value_json.sound_mode }}',
'command_topic': f"{topic_prefix}/{CMD_SOUND_MODE}",
'options': _SOUND_MODES,
'icon': 'mdi:volume-high',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/select/{topic_prefix}/sound_mode/config",
encode(cfg)))
# select: door LED brightness
cfg = {
'name': 'Door LED brightness',
'unique_id': f"{topic_prefix}_led_brightness_select",
'object_id': f"{topic_prefix}_led_brightness_select",
'state_topic': state_topic,
'value_template': '{{ value_json.led_brightness }}',
'command_topic': f"{topic_prefix}/{CMD_LED_BRIGHT}",
'options': _LED_LEVELS,
'icon': 'mdi:brightness-6',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/select/{topic_prefix}/led_brightness/config",
encode(cfg)))
# switch: door LED night light
cfg = {
'name': 'Door LED night light',
'unique_id': f"{topic_prefix}_led_night_light_switch",
'object_id': f"{topic_prefix}_led_night_light_switch",
'state_topic': state_topic,
'value_template': '{{ value_json.led_night_light }}',
'state_on': 'On',
'state_off': 'Off',
'command_topic': f"{topic_prefix}/{CMD_LED_NIGHT}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': 'mdi:weather-night',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/switch/{topic_prefix}/led_night_light/config",
encode(cfg)))
return out
# --- MQTT command handlers ------------------------------------------------
def command_handlers():
def _operational(p, _links):
if p not in ('Run', 'Pause', 'Ready'):
return None
return ['operational', 'state', 'vs', '0'], {
'x.com.samsung.da.state': p,
}
def _sound_mode(p, _links):
if p not in _SOUND_MODES:
return None
return ['settings', 'sound', 'mode', 'vs', '0'], {'mode': p}
def _led_brightness(p, _links):
if p not in _LED_LEVELS:
return None
return ['doorled', 'light', 'vs', '0'], {'setBrightness': p}
def _led_night(p, _links):
if p not in ('On', 'Off'):
return None
return ['doorled', 'light', 'vs', '0'], {'setNightLight': p}
return {
CMD_OPERATIONAL: _operational,
CMD_SOUND_MODE: _sound_mode,
CMD_LED_BRIGHT: _led_brightness,
CMD_LED_NIGHT: _led_night,
}
# --- Poll tiers -----------------------------------------------------------
# Same washer-family firmware as the dryer (~14 req/s empirical ceiling).
DISHWASHER_POLL_TIERS = [
PollTier(
name='hot',
interval_s=1.0,
active_interval_s=0.5,
paths=(
('operational', 'state', 'vs', '0'),
),
),
PollTier(
name='warm',
interval_s=15.0,
paths=(
('power', 'vs', '0'),
('kidslock', 'vs', '0'),
('remotectrl', 'vs', '0'),
('alarms', 'vs', '0'),
('energy', 'consumption', 'vs', '0'),
('water', 'consumption', 'vs', '0'),
('filter', 'waterfilter', 'vs', '0'),
('doorled', 'light', 'vs', '0'),
('settings', 'sound', 'mode', 'vs', '0'),
('wm', 'jobbeginingstatus', 'vs', '0'),
),
),
PollTier(
name='sweep',
interval_s=300.0,
paths=(('device', '0'),),
is_sweep=True,
),
]
def _is_active(links: dict) -> bool:
rep = links.get('/operational/state/vs/0') or {}
sam_state = rep.get('x.com.samsung.da.state')
return _SAMSUNG_STATE_TO_OCF.get(sam_state) == 'active'
# --- Descriptor -----------------------------------------------------------
DISHWASHER = ApplianceDescriptor(
name='dishwasher',
default_observe_port=49154,
observe_paths=OBSERVE_PATHS,
seed_path=['device', '0'],
flatten=flatten,
build_discovery=build_discovery,
command_handlers=command_handlers,
on_observation=on_observation,
project=project,
remote_available_field='remote_control_binary',
log_state_change=log_state_change,
poll_tiers=DISHWASHER_POLL_TIERS,
is_active=_is_active,
)
samsung_appliance/appliances/dryer.py
-503
View File
@@ -1,503 +0,0 @@
"""Dryer descriptor.
Lifts the dryer-specific OBSERVE paths, sensor flattening, HA discovery
inventory, and MQTT command handlers out of the original
samsung_dryer/{bridge,sensors,discovery}.py modules into one place.
"""
import time
from .base import (
ApplianceDescriptor,
avail_base,
avail_with_remote,
device_block,
encode,
)
from ..poll_scheduler import PollTier
# --- OBSERVE paths -----------------------------------------------------
# Only Samsung's `/<x>/vs/0` siblings actually push notifications; the
# OCF-standard `/<x>/0` paths accept registration silently but never
# fire. flatten() derives the OCF-shaped values from the live /vs/0
# strings.
OBSERVE_PATHS = [
['operational', 'state', 'vs', '0'], # state, remainingTime, progress
['power', 'vs', '0'], # power on/off
['kidslock', 'vs', '0'], # child lock
['remotectrl', 'vs', '0'], # remote control enabled
['energy', 'consumption', 'vs', '0'],
['course', 'vs', '0'],
['washer', 'vs', '0'], # dryLevel, dryTime, type
['diagnosis', 'vs', '0'],
['alarms', 'vs', '0'],
['st', 'dryercourse', 'vs', '0'],
['wm', 'jobbeginingstatus', 'vs', '0'],
]
# --- Course table ------------------------------------------------------
# Captured 2026-05-29 by dialing every course on a
# DA_WM_TP2_20_COMMON_DV5000T dryer. Other Samsung dryers may report a
# different Table_NN; capture a fresh table for them with
# local-tools/course_mapper.py.
COURSE_NAMES = {
'Table_03': {
0x16: 'Cotton',
0x18: 'Synthetics',
0x19: 'Delicates',
0x1A: 'Wool',
0x1B: 'Bedding',
0x1C: 'Shirts',
0x1D: 'Towels',
0x1E: 'Outdoor',
0x1F: 'Mixed Load',
0x20: 'Iron Dry',
0x23: 'Quick Dry 35',
0x24: 'Cool Air',
0x25: 'Warm Air',
0x27: 'Time Dry',
},
}
_COURSE_CODE_BY_NAME = {
name: code
for table_codes in COURSE_NAMES.values()
for code, name in table_codes.items()
}
def _decode_course(s):
"""`Table_03_Course_16` → `Cotton`. Pass through verbatim if the
table or code isn't in our lookup."""
if not isinstance(s, str) or '_Course_' not in s:
return s
table_part, _, code_str = s.partition('_Course_')
table = COURSE_NAMES.get(table_part)
if not table:
return s
try:
code = int(code_str, 16)
except ValueError:
return s
return table.get(code, s)
def _encode_course(name):
"""`Cotton` → `Course_16`. Returns None for unknown names so the
caller refuses rather than POST garbage."""
code = _COURSE_CODE_BY_NAME.get(name)
if code is None:
return None
return f"Course_{code:02X}"
def _course_options():
"""Stable-sorted human course names for the HA select dropdown."""
return sorted(_COURSE_CODE_BY_NAME.keys())
# --- Samsung-state → OCF currentMachineState ---------------------------
_SAMSUNG_STATE_TO_OCF = {
'Ready': 'idle',
'Run': 'active',
'Running': 'active',
'Pause': 'pause',
'Paused': 'pause',
'End': 'idle',
}
def _num(v):
try:
return float(v)
except (TypeError, ValueError):
return None
def _int(v):
try:
return int(v)
except (TypeError, ValueError):
return None
# --- flatten -----------------------------------------------------------
def flatten(links):
"""Map a /device/0 link dict to the flat sensor dict that's
published to MQTT. Every field reads from `/<x>/vs/0` paths so push
updates immediately drive every entity."""
g = lambda href, k, default=None: (links.get(href) or {}).get(k, default)
inst_w = _num(g('/energy/consumption/vs/0',
'x.com.samsung.da.instantaneousPower'))
cum_wh = _num(g('/energy/consumption/vs/0',
'x.com.samsung.da.cumulativePower'))
if inst_w is not None and inst_w < 0:
# The dryer reports a phantom -500W when idle; HA energy
# dashboard hates negatives.
inst_w = 0.0
sam_state = g('/operational/state/vs/0', 'x.com.samsung.da.state')
machine_state = (_SAMSUNG_STATE_TO_OCF.get(sam_state, sam_state)
if sam_state is not None
else g('/operational/state/0', 'currentMachineState'))
progress = g('/operational/state/vs/0', 'x.com.samsung.da.progress')
job_state = progress or g('/operational/state/0', 'currentJobState')
# HA's value_template treats the literal "None" as null (renders as
# "Unknown"). Substitute something we can render verbatim.
if job_state in (None, 'None'):
job_state = 'Idle'
if progress in (None, 'None'):
progress = 'Idle'
remaining = (g('/operational/state/vs/0',
'x.com.samsung.da.remainingTime')
or g('/operational/state/0', 'remainingTime'))
rem_min = None
if remaining:
try:
h, m, s = remaining.split(':')
rem_min = int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
except Exception:
pass
sam_power = g('/power/vs/0', 'x.com.samsung.da.power')
sam_kids = g('/kidslock/vs/0', 'x.com.samsung.da.kidsLock')
sam_rc = g('/remotectrl/vs/0',
'x.com.samsung.da.remoteControlEnabled')
power_bin = (sam_power == 'On') if sam_power is not None else None
kids_bin = (sam_kids != 'Ready') if sam_kids is not None else None
rc_bin = (str(sam_rc).lower() == 'true') if sam_rc is not None else None
return {
'machine_state': machine_state,
'job_state': job_state,
'progress': progress,
'progress_percentage': _int(g('/operational/state/vs/0',
'x.com.samsung.da.progressPercentage')
or g('/operational/state/0',
'progressPercentage')),
'completion_time': remaining,
'completion_minutes': rem_min,
'delay_end_time': g('/operational/state/vs/0',
'x.com.samsung.da.delayEndTime'),
'power_state': sam_power,
'power_state_binary': power_bin,
'child_lock': sam_kids,
'child_lock_binary': kids_bin,
'remote_control': sam_rc,
'remote_control_binary': rc_bin,
'power_watts': inst_w,
'energy_kwh': round(cum_wh / 1000.0, 2)
if cum_wh is not None else None,
'energy_wh_cumulative': int(cum_wh) if cum_wh is not None else None,
'dryer_mode': _decode_course(
g('/st/dryercourse/vs/0',
'x.com.samsung.da.st.dryerMode')),
'dry_level': _int(g('/washer/vs/0',
'x.com.samsung.da.dryLevel')),
'dry_time': g('/washer/vs/0',
'x.com.samsung.da.dryTime'),
'dryer_type': g('/washer/vs/0',
'x.com.samsung.da.dryerType'),
'wrinkle_prevent': g('/washer/vs/0',
'x.com.samsung.da.wrinklePrevent'),
'diagnosis': g('/diagnosis/vs/0',
'x.com.samsung.da.diagnosisStart'),
'country_code': g('/configuration/vs/0',
'x.com.samsung.da.countryCode'),
}
# --- Remaining-time anchor + extrapolation ----------------------------
# The dryer pushes /operational/state/vs/0 on state transitions but not
# on remainingTime ticks. Anchor = (timestamp, total_seconds) at last
# push; project() extrapolates downward while machine_state == 'active'.
def on_observation(state, href, rep):
if href != '/operational/state/vs/0':
return
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
# --- Log-line ----------------------------------------------------------
def log_state_change(sensors):
return (f"machine={sensors.get('machine_state')} "
f"power={sensors.get('power_watts')}W "
f"energy={sensors.get('energy_kwh')}kWh")
# --- HA discovery ------------------------------------------------------
MODEL = 'OCF dryer (TizenRT-iotivity)'
# (key, friendly name, extra-config-dict)
_SENSORS = [
('machine_state', 'Machine state', {'icon': 'mdi:tumble-dryer'}),
('job_state', 'Job state', {}),
('progress', 'Progress', {}),
('progress_percentage', 'Progress percent',
{'unit_of_measurement': '%', 'state_class': 'measurement'}),
('completion_time', 'Completion time', {'icon': 'mdi:timer-sand'}),
('completion_minutes', 'Remaining minutes',
{'unit_of_measurement': 'min', 'device_class': 'duration',
'state_class': 'measurement'}),
('delay_end_time', 'Delay end time', {'icon': 'mdi:timer'}),
('power_state', 'Power state', {}),
('power_watts', 'Power',
{'unit_of_measurement': 'W', 'device_class': 'power',
'state_class': 'measurement'}),
('energy_kwh', 'Energy',
{'unit_of_measurement': 'kWh', 'device_class': 'energy',
'state_class': 'total_increasing'}),
('dryer_mode', 'Dryer mode', {}),
('dry_level', 'Dry level', {}),
('dry_time', 'Dry time', {}),
('dryer_type', 'Dryer type', {}),
('wrinkle_prevent', 'Wrinkle prevent', {}),
('diagnosis', 'Diagnosis', {}),
('country_code', 'Country code', {}),
]
# (key, friendly name, value_template, device_class)
_BINARY_SENSORS = [
('running', 'Running',
"{{ 'ON' if value_json.machine_state == 'active' else 'OFF' }}",
'running'),
('power_switch', 'Power switch',
"{{ 'ON' if value_json.power_state_binary else 'OFF' }}",
'power'),
('child_lock_active', 'Child lock',
"{{ 'ON' if value_json.child_lock_binary else 'OFF' }}",
'lock'),
('remote_control_enabled', 'Remote control',
"{{ 'ON' if value_json.remote_control_binary else 'OFF' }}",
'connectivity'),
]
# MQTT command-topic suffixes. The bridge subscribes to <prefix>/cmd/#
# and dispatches by suffix.
CMD_WRINKLE_PREVENT = 'cmd/wrinkle_prevent'
CMD_OPERATIONAL = 'cmd/operational_state'
CMD_DRYER_MODE = 'cmd/dryer_mode'
def build_discovery(topic_prefix, ha_prefix, device_name):
"""Return list of (discovery_topic, payload_bytes) tuples ready to
publish (retained) on MQTT connect."""
state_topic = f"{topic_prefix}/state"
avail_topic = f"{topic_prefix}/availability"
remote_topic = f"{topic_prefix}/remote_available"
dev = device_block(topic_prefix, device_name, MODEL)
out = []
# read-only sensors
for key, name, extra in _SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': f"{{{{ value_json.{key} }}}}",
'availability': avail_base(avail_topic),
'device': dev,
}
cfg.update(extra)
out.append((f"{ha_prefix}/sensor/{topic_prefix}/{key}/config",
encode(cfg)))
for key, name, template, dclass in _BINARY_SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': template,
'payload_on': 'ON',
'payload_off': 'OFF',
'device_class': dclass,
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/binary_sensor/{topic_prefix}/{key}/config",
encode(cfg)))
# switch: wrinkle prevent (always available)
cfg = {
'name': 'Wrinkle prevent',
'unique_id': f"{topic_prefix}_wrinkle_prevent_switch",
'object_id': f"{topic_prefix}_wrinkle_prevent_switch",
'state_topic': state_topic,
'value_template': '{{ value_json.wrinkle_prevent }}',
'state_on': 'On',
'state_off': 'Off',
'command_topic': f"{topic_prefix}/{CMD_WRINKLE_PREVENT}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': 'mdi:iron',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/switch/{topic_prefix}/wrinkle_prevent/config",
encode(cfg)))
# buttons: Start / Pause / Stop (gated on remote control)
buttons = [
('start', 'Start cycle', 'Run', 'mdi:play'),
('pause', 'Pause cycle', 'Pause', 'mdi:pause'),
('stop', 'Stop cycle', 'Ready', 'mdi:stop'),
]
for key, name, payload_press, icon in buttons:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'command_topic': f"{topic_prefix}/{CMD_OPERATIONAL}",
'payload_press': payload_press,
'icon': icon,
'availability': avail_with_remote(avail_topic, remote_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/button/{topic_prefix}/{key}/config",
encode(cfg)))
# select: course (gated on remote control)
cfg = {
'name': 'Course',
'unique_id': f"{topic_prefix}_course_select",
'object_id': f"{topic_prefix}_course_select",
'state_topic': state_topic,
'value_template': '{{ value_json.dryer_mode }}',
'command_topic': f"{topic_prefix}/{CMD_DRYER_MODE}",
'options': _course_options(),
'icon': 'mdi:tumble-dryer',
'availability': avail_with_remote(avail_topic, remote_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/select/{topic_prefix}/course/config",
encode(cfg)))
return out
# --- MQTT command handlers --------------------------------------------
def command_handlers():
"""topic_suffix → fn(payload, links) → (path_segs, body_dict) | None.
`None` means refuse the command (caller logs & drops). Dryer
handlers don't need the links snapshot — they're all single-field
writes."""
def _wrinkle(p, _links):
if p not in ('On', 'Off'):
return None
return ['washer', 'vs', '0'], {'x.com.samsung.da.wrinklePrevent': p}
def _operational(p, _links):
if p not in ('Run', 'Pause', 'Ready'):
return None
return ['operational', 'state', 'vs', '0'], {'x.com.samsung.da.state': p}
def _course(p, _links):
code = _encode_course(p)
if code is None:
return None
return ['st', 'dryercourse', 'vs', '0'], {'x.com.samsung.da.st.dryerMode': code}
return {
CMD_WRINKLE_PREVENT: _wrinkle,
CMD_OPERATIONAL: _operational,
CMD_DRYER_MODE: _course,
}
# --- Poll tiers --------------------------------------------------------
# Empirical ceiling on this firmware is ~14 req/s (probe_poll_rate_combined.py
# 2026-06-03). The hot tier sits at 1s idle / 0.5s active — comfortably under
# the ceiling and leaves headroom for the warm + sweep budgets.
# Per-tier timeouts are scaled to cadence: hot tier retries every 1s, so
# a tight 2s ceiling caps the cascade damage from one wedged poll. Warm
# tier has more headroom; sweep is multi-block Block2 and tolerates ~15s.
DRYER_POLL_TIERS = [
PollTier(
name='hot',
interval_s=1.0,
active_interval_s=0.5,
timeout_s=2.0,
paths=(
('operational', 'state', 'vs', '0'),
),
),
PollTier(
name='warm',
interval_s=15.0,
timeout_s=4.0,
paths=(
('power', 'vs', '0'),
('kidslock', 'vs', '0'),
('remotectrl', 'vs', '0'),
('alarms', 'vs', '0'),
('course', 'vs', '0'),
('washer', 'vs', '0'),
('st', 'dryercourse', 'vs', '0'),
('wm', 'jobbeginingstatus', 'vs', '0'),
('energy', 'consumption', 'vs', '0'),
('diagnosis', 'vs', '0'),
),
),
PollTier(
name='sweep',
interval_s=300.0,
timeout_s=15.0,
paths=(('device', '0'),),
is_sweep=True,
),
]
def _is_active(links: dict) -> bool:
rep = links.get('/operational/state/vs/0') or {}
sam_state = rep.get('x.com.samsung.da.state')
return _SAMSUNG_STATE_TO_OCF.get(sam_state) == 'active'
# --- Descriptor --------------------------------------------------------
DRYER = ApplianceDescriptor(
name='dryer',
default_observe_port=49155,
observe_paths=OBSERVE_PATHS,
seed_path=['device', '0'],
flatten=flatten,
build_discovery=build_discovery,
command_handlers=command_handlers,
on_observation=on_observation,
project=project,
remote_available_field='remote_control_binary',
log_state_change=log_state_change,
poll_tiers=DRYER_POLL_TIERS,
is_active=_is_active,
)
samsung_appliance/appliances/oven.py
-862
View File
@@ -1,862 +0,0 @@
"""Oven descriptor (Samsung NV7000BS-class).
Resource map captured 2026-05-31 via DTLS-CoAP with the client cert.
See `local-tools/comparisons/oven-tree.md` for the full field reference.
Write surfaces this descriptor exposes:
proven:
* UpperLamp via /mode/vs/0 options RMW (probe_oven_lamp_toggle.py)
— works even with Remote Control off.
unproven (first HA use is also the test):
* Sound, FastPreheat — same RMW pattern as lamp.
* Setpoint via /temperatures/vs/0 items RMW. Mid-cook write may
or may not retune the element (plan §K-U #2).
* Mode select via /mode/vs/0 .modes — mid-cook acceptance unknown
(plan §K-U #3).
* Power on/off via /power/vs/0.
* Stop via /operational/state/vs/0 (dryer convention; oven may
use a different state value).
Untested writes are gated behind <prefix>/remote_available so HA
disables them in the UI when the oven's Remote Control switch is off."""
import time
from .base import (
ApplianceDescriptor,
avail_base,
avail_with_cycle,
avail_with_remote_and_cycle,
device_block,
encode,
)
from ..poll_scheduler import PollTier
# ---------------------------------------------------------------------
# OBSERVE paths — every push-eligible /<x>/vs/0 resource on the oven.
# Same wedge-safety story as the dryer: only `/<x>/vs/0` siblings push;
# OCF-standard `/<x>/0` paths accept registration but never fire.
# Security paths (/oic/sec/{doxm,pstat,acl,cred}) are deliberately
# EXCLUDED — those are the surfaces that nearly bricked the oven in
# prior sessions. The bridge has no reason to touch them.
# ---------------------------------------------------------------------
OBSERVE_PATHS = [
['operational', 'state', 'vs', '0'], # state, time, progress
['power', 'vs', '0'], # power On/Off
['oven', 'vs', '0'], # cavity state (Cooking, Idle, …)
['temperatures','vs', '0'], # current + desired temp
['doors', 'vs', '0'], # openState
['kidslock', 'vs', '0'], # child lock
['remotectrl', 'vs', '0'], # remote control enabled
['mode', 'vs', '0'], # cooking mode + options array
['alarms', 'vs', '0'], # alarm code (OV_E_OFF etc.)
['connected', 'vs', '0'], # cloud connectivity status
['otninformation', 'vs', '0'], # firmware-update flags
]
# Setpoint bounds — union across modeSpec entries on this oven. Per-mode
# bounds (e.g. PlateWarm 30–80) tighten this; the firmware will refuse
# out-of-range writes for the active mode and the HA UI will surface
# the resulting 4.xx in the bridge log.
SETPOINT_MIN_C = 30
SETPOINT_MAX_C = 270
SETPOINT_STEP_C = 5
# Samsung's operational state strings → OCF currentMachineState shape.
_SAMSUNG_STATE_TO_OCF = {
'Ready': 'idle',
'Run': 'active',
'Running': 'active',
'Pause': 'pause',
'Paused': 'pause',
'End': 'idle',
'Stop': 'idle',
}
def _num(v):
try: return float(v)
except (TypeError, ValueError): return None
def _int(v):
try: return int(v)
except (TypeError, ValueError): return None
def _option_value(options, prefix, default=None):
"""Find `<prefix>_<value>` in an options array and return <value>."""
for o in options:
if o.startswith(prefix + '_'):
return o.split('_', 1)[1]
return default
def _replace_in_options(options, prefix, new_value):
"""Return a new options array with any `<prefix>_*` entry replaced
by `<prefix>_<new_value>`. Caller must verify `options` is the live
options array first (Samsung uses replace-not-merge on this field)."""
return [f"{prefix}_{new_value}" if o.startswith(prefix + '_') else o
for o in options]
def _fmt_hms(seconds):
"""Format an integer second count as `H:MM:SS`. Returns None on
bad input so callers can leave the field null rather than emitting
a misleading `0:00:00`."""
try:
s = int(seconds)
except (TypeError, ValueError):
return None
if s < 0:
s = 0
h, rest = divmod(s, 3600)
m, sec = divmod(rest, 60)
return f"{h}:{m:02d}:{sec:02d}"
# ---------------------------------------------------------------------
# flatten — Samsung /device/0 links → HA-flavoured sensor dict.
# Every field reads from `/<x>/vs/0` paths so push updates immediately
# drive every entity. Where a field is settable (lamp, mode, setpoint),
# we publish it as a read-side sensor here AND as a writeable entity
# in build_discovery; the read side closes the HA UI feedback loop.
# ---------------------------------------------------------------------
def flatten(links):
g = lambda href, k, default=None: (links.get(href) or {}).get(k, default)
# Operational
sam_state = g('/operational/state/vs/0', 'x.com.samsung.da.state')
machine_state = (_SAMSUNG_STATE_TO_OCF.get(sam_state, sam_state)
if sam_state is not None else None)
operation_time = g('/operational/state/vs/0',
'x.com.samsung.da.operationTime')
remaining = g('/operational/state/vs/0',
'x.com.samsung.da.remainingTime')
rem_min = None
if remaining:
try:
h, m, s = remaining.split(':')
rem_min = int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
except Exception:
pass
# operationTime parsed as minutes — the source of truth for "Cook
# time" in HA (mid-cycle SmartThings updates land here, not in
# /mode/vs/0 UpperTimerSet which is vestigial).
op_min = None
if operation_time:
try:
h, m, s = operation_time.split(':')
op_min = int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
except Exception:
pass
# Cavity state — Cooking, Idle, Preheating, …
oven_state = g('/oven/vs/0', 'x.com.samsung.da.state')
# Temperatures
temps_items = (g('/temperatures/vs/0',
'x.com.samsung.da.items') or [])
cur_c = des_c = None
if temps_items:
cur_c = _int(temps_items[0].get('x.com.samsung.da.current'))
des_c = _int(temps_items[0].get('x.com.samsung.da.desired'))
# Door
doors_items = g('/doors/vs/0', 'x.com.samsung.da.items') or []
door = doors_items[0].get('x.com.samsung.da.openState') if doors_items else None
door_open = (door == 'Open') if door is not None else None
# Power
sam_power = g('/power/vs/0', 'x.com.samsung.da.power')
power_bin = (sam_power == 'On') if sam_power is not None else None
# Kidslock + Remote
sam_kids = g('/kidslock/vs/0', 'x.com.samsung.da.kidsLock')
kids_bin = (sam_kids != 'Ready') if sam_kids is not None else None
sam_rc = g('/remotectrl/vs/0',
'x.com.samsung.da.remoteControlEnabled')
rc_bin = (str(sam_rc).lower() == 'true') if sam_rc is not None else None
# Mode + options
modes = g('/mode/vs/0', 'x.com.samsung.da.modes') or []
current_mode = modes[0] if modes else None
options = g('/mode/vs/0', 'x.com.samsung.da.options') or []
lamp = _option_value(options, 'UpperLamp') # 'On' / 'Off'
# The door-coupling override (open → On, close → Off) lives in
# project() below — it needs descriptor_state to compare the
# latest door transition against the latest lamp value change so
# an HA-initiated optimistic write isn't clobbered by stale door
# state.
sound = _option_value(options, 'Sound') # 'On' / 'Off'
fastpreheat = _option_value(options, 'fastpreheat') # 'On' / 'Off'
# NaturalSteam only appears in the options array after it's been
# touched in the SmartThings app at least once. Until then it's
# absent, so `_option_value` returns None — surface that as None
# (HA renders as "Unknown") rather than "Off", which would falsely
# imply we know it's disabled.
natural_steam = _option_value(options, 'NaturalSteam') # 'On' / 'Off' / None
timer_state = _option_value(options, 'UpperTimerState') # 'Ready' / 'Running'
# UpperTimerCurrent/UpperTimerSet are integer seconds. Format as
# H:MM:SS for HA display so users see "1:10:00", not "4200".
timer_current_raw = _option_value(options, 'UpperTimerCurrent')
timer_set_raw = _option_value(options, 'UpperTimerSet')
timer_current = _fmt_hms(timer_current_raw)
timer_set = _fmt_hms(timer_set_raw)
timer_current_seconds = _int(timer_current_raw)
timer_set_seconds = _int(timer_set_raw)
# Alarms
alarm_items = g('/alarms/vs/0', 'x.com.samsung.da.items') or []
alarm_code = (alarm_items[0].get('x.com.samsung.da.code')
if alarm_items else None)
alarm_time = (alarm_items[0].get('x.com.samsung.da.triggeredTime')
if alarm_items else None)
# OV_E_OFF appears when the oven is off / no alarm; treat as inactive.
alarm_active = bool(alarm_code) and alarm_code != 'OV_E_OFF'
# Connectivity / firmware
sam_connected = g('/connected/vs/0', 'x.com.samsung.da.connected')
connected_bin = (sam_connected == 'On') if sam_connected is not None else None
fw_update_available = g('/otninformation/vs/0',
'x.com.samsung.da.newVersionAvailable')
fw_update_bin = (str(fw_update_available).lower() == 'true'
if fw_update_available is not None else None)
return {
'machine_state': machine_state,
# `cycle_active` gates the writable controls in HA. The oven
# only honours setpoint / cook-time / option writes (and Stop)
# while a cycle is active — outside an active cycle, writes
# return 2.04 but get rolled back within ~3s.
'cycle_active': machine_state == 'active',
'oven_state': oven_state,
'progress_percentage': _int(g('/operational/state/vs/0',
'x.com.samsung.da.progressPercentage')),
'operation_time': operation_time,
'operation_time_minutes': op_min,
'completion_time': remaining,
'completion_minutes': rem_min,
'current_temp_c': cur_c,
'target_temp_c': des_c,
'door': door,
'door_open': door_open,
'power_state': sam_power,
'power_state_binary': power_bin,
'child_lock': sam_kids,
'child_lock_binary': kids_bin,
'remote_control': sam_rc,
'remote_control_binary': rc_bin,
'mode': current_mode,
'lamp': lamp,
'sound': sound,
'fastpreheat': fastpreheat,
'natural_steam': natural_steam,
'timer_state': timer_state,
'timer_current': timer_current,
'timer_set': timer_set,
'timer_current_seconds': timer_current_seconds,
'timer_set_seconds': timer_set_seconds,
'alarm_code': alarm_code,
'alarm_time': alarm_time,
'alarm_active': alarm_active,
'connected': sam_connected,
'connected_binary': connected_bin,
'firmware_update_available': fw_update_bin,
}
# ---------------------------------------------------------------------
# Remaining-time anchor + projection. The oven pushes /operational/state
# on state transitions but probably not on remainingTime ticks (matches
# dryer behaviour). Capture (ts, total_seconds) at each push and
# extrapolate downward while machine_state == active.
# ---------------------------------------------------------------------
def on_observation(state, href, rep):
now = time.time()
if href == '/operational/state/vs/0':
rem = rep.get('x.com.samsung.da.remainingTime')
if isinstance(rem, str):
try:
h, m, s = rem.split(':')
state['remaining_anchor'] = (now,
int(h) * 3600 + int(m) * 60 + int(s))
except (ValueError, AttributeError):
pass
return
# Door + lamp tracking feeds the lamp/door coupling in project().
# Both timestamps bump only on value CHANGES so the comparison
# tells us which event happened more recently. /doors is hot-tier
# (1s) and would otherwise dominate; /mode is warm-tier (30s) and
# picks up HA optimistic writes via apply_optimistic → apply_rep.
if href == '/doors/vs/0':
items = rep.get('x.com.samsung.da.items') or []
door = items[0].get('x.com.samsung.da.openState') if items else None
if door != state.get('_door_last'):
state['_door_last'] = door
state['_door_change_ts'] = now
return
if href == '/mode/vs/0':
options = rep.get('x.com.samsung.da.options') or []
lamp = _option_value(options, 'UpperLamp')
if lamp != state.get('_lamp_last'):
state['_lamp_last'] = lamp
state['_lamp_change_ts'] = now
def project(state, sensors):
sensors = dict(sensors)
# Remaining-time projection: the oven pushes /operational/state on
# state transitions but not on remainingTime ticks. Extrapolate
# from the most recent anchor while the machine is active.
anchor = state.get('remaining_anchor')
if sensors.get('machine_state') == 'active' and anchor is not None:
ts, total = anchor
remaining = max(0, int(total - (time.time() - ts)))
h, rest = divmod(remaining, 3600)
m, s = divmod(rest, 60)
sensors['completion_time'] = f"{h}:{m:02d}:{s:02d}"
sensors['completion_minutes'] = h * 60 + m + (1 if s > 0 else 0)
# Lamp / door coupling. The oven hardware auto-drives the lamp from
# the door state, but /mode/vs/0 only polls every 30s. When a door
# TRANSITION is more recent than the last lamp VALUE change, derive
# lamp from door for sub-second freshness. When a lamp toggle is
# more recent (HA optimistic write, or panel-driven /mode diff),
# the cache value wins — preserves HA toggle responsiveness even
# while the door is closed.
door_ts = state.get('_door_change_ts')
lamp_ts = state.get('_lamp_change_ts')
door_open = sensors.get('door_open')
if door_ts is not None and (lamp_ts is None or door_ts > lamp_ts):
if door_open is True:
sensors['lamp'] = 'On'
elif door_open is False:
sensors['lamp'] = 'Off'
return sensors
def log_state_change(sensors):
return (f"machine={sensors.get('machine_state')} "
f"oven={sensors.get('oven_state')} "
f"temp={sensors.get('current_temp_c')}/"
f"{sensors.get('target_temp_c')}°C "
f"mode={sensors.get('mode')} "
f"timer_set={sensors.get('timer_set_seconds')} "
f"timer_cur={sensors.get('timer_current_seconds')}")
# ---------------------------------------------------------------------
# HA discovery inventory
# ---------------------------------------------------------------------
MODEL = 'OCF oven (TizenRT-iotivity, NV7000BS-class)'
# (key, friendly name, extra config)
#
# Only read-only sensors live here. Fields that ALSO have an
# interactive entity (light, switch, number, select) are removed —
# the interactive entity already surfaces the live state, so a
# duplicate read-only "Lamp state" / "Fast preheat state" / etc.
# sensor would just clutter the device card with the same value
# twice.
_SENSORS = [
('machine_state', 'Machine state', {'icon': 'mdi:stove'}),
('oven_state', 'Cavity state', {}),
# Cooking mode is read-only via local OCF — the oven owns the
# `modes` field once a cycle is active and rolls back any writes.
('mode', 'Cooking mode', {'icon': 'mdi:tune'}),
('progress_percentage', 'Progress percent',
{'unit_of_measurement': '%', 'state_class': 'measurement'}),
('operation_time', 'Elapsed time', {'icon': 'mdi:timer'}),
('completion_time', 'Completion time', {'icon': 'mdi:timer-sand'}),
('completion_minutes', 'Remaining minutes',
{'unit_of_measurement': 'min', 'device_class': 'duration',
'state_class': 'measurement'}),
('current_temp_c', 'Temperature',
{'unit_of_measurement': '°C', 'device_class': 'temperature',
'state_class': 'measurement'}),
# target_temp_c is also exposed as a Number entity for editing,
# but the Number is RC-gated. The sensor stays always-visible so
# the user can see the current setpoint even with Remote Control
# off at the oven.
('target_temp_c', 'Setpoint',
{'unit_of_measurement': '°C', 'device_class': 'temperature',
'state_class': 'measurement', 'icon': 'mdi:thermometer-chevron-up'}),
# power_state: read-only. The oven doesn't expose a meaningful
# POST /power/vs/0 from cold — turning the unit on at the panel
# is a physical action — so we don't ship a Power switch entity.
('power_state', 'Power state', {'icon': 'mdi:power'}),
('door', 'Door state', {}),
('child_lock', 'Child lock state', {}),
('remote_control', 'Remote control state', {}),
('timer_state', 'Timer state', {}),
('timer_current', 'Timer remaining', {'icon': 'mdi:timer-sand'}),
('timer_set', 'Timer set', {'icon': 'mdi:timer'}),
('alarm_code', 'Alarm code',
{'icon': 'mdi:alert', 'entity_category': 'diagnostic'}),
('alarm_time', 'Alarm time',
{'icon': 'mdi:clock-alert', 'entity_category': 'diagnostic'}),
('connected', 'Cloud connectivity',
{'entity_category': 'diagnostic'}),
]
# (key, friendly, value_template, device_class, extras)
_BINARY_SENSORS = [
('running', 'Running',
"{{ 'ON' if value_json.machine_state == 'active' else 'OFF' }}",
'running', {}),
('door_open', 'Door',
"{{ 'ON' if value_json.door_open else 'OFF' }}",
'door', {}),
# `power_switch` binary_sensor would duplicate the Power switch
# entity below; the switch already shows on/off state.
('child_lock_active', 'Child lock',
"{{ 'ON' if value_json.child_lock_binary else 'OFF' }}",
'lock', {}),
('remote_control_enabled', 'Remote control',
"{{ 'ON' if value_json.remote_control_binary else 'OFF' }}",
'connectivity', {}),
('alarm_active', 'Alarm active',
"{{ 'ON' if value_json.alarm_active else 'OFF' }}",
'problem', {}),
('connected_bin', 'Connected',
"{{ 'ON' if value_json.connected_binary else 'OFF' }}",
'connectivity', {'entity_category': 'diagnostic'}),
('firmware_update_available', 'Firmware update available',
"{{ 'ON' if value_json.firmware_update_available else 'OFF' }}",
'update', {'entity_category': 'diagnostic'}),
]
# MQTT command-topic suffixes (under <prefix>/cmd/…)
CMD_LAMP = 'cmd/lamp'
CMD_SOUND = 'cmd/sound'
CMD_FASTPREHEAT = 'cmd/fastpreheat'
CMD_NATURALSTEAM = 'cmd/naturalsteam'
CMD_POWER = 'cmd/power'
CMD_STOP = 'cmd/stop'
CMD_SETPOINT = 'cmd/setpoint'
CMD_COOK_TIME = 'cmd/cook_time'
# NOTE — no CMD_START or CMD_MODE. Reverse-engineered 2026-05-31:
# * `state='Run'` writes to /operational/state/vs/0 are accepted
# (2.04) and machine briefly goes active, but the oven cavity
# stays Ready (no Preheat) and the cycle self-cancels within
# ~3s. Tried every byte-level approximation of SmartThings's
# working start (matching all four fields on /operational/state,
# +operationTime, +remainingTime, +progressPercentage='1', plus
# /temperatures/vs/0 desired, with and without /mode/vs/0 modes,
# with PUT vs POST, paced 1s apart, with OCF-version-options
# 2049/2053, with Samsung vendor-option 65524=0xc0) — none of
# these engage the cavity. The differentiator must be something
# invisible at the OBSERVE-push level (likely a cloud-mediated
# auth path the SmartThings app uses). See project_oven_remote
# _start_open.md for full notes.
# * `modes=['Convection']` writes to /mode/vs/0 succeed (2.04)
# but the oven owns the field once a cycle is active and rolls
# local writes back to ['NoOperation']. mode is surfaced as a
# read-only sensor instead.
def build_discovery(topic_prefix, ha_prefix, device_name):
state_topic = f"{topic_prefix}/state"
avail_topic = f"{topic_prefix}/availability"
remote_topic = f"{topic_prefix}/remote_available"
cycle_topic = f"{topic_prefix}/cycle_active"
dev = device_block(topic_prefix, device_name, MODEL)
out = []
# --- read-only sensors -------------------------------------------
for key, name, extra in _SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': f"{{{{ value_json.{key} }}}}",
'availability': avail_base(avail_topic),
'device': dev,
}
cfg.update(extra)
out.append((f"{ha_prefix}/sensor/{topic_prefix}/{key}/config",
encode(cfg)))
for key, name, template, dclass, extra in _BINARY_SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': template,
'payload_on': 'ON',
'payload_off': 'OFF',
'device_class': dclass,
'availability': avail_base(avail_topic),
'device': dev,
}
cfg.update(extra)
out.append((f"{ha_prefix}/binary_sensor/{topic_prefix}/{key}/config",
encode(cfg)))
# --- light: oven lamp (proven via probe_oven_lamp_states.py;
# binary On/Off only — High/Low/Dim coerce back to previous
# state. Works regardless of Remote Control switch, so we only
# gate on base availability). For the MQTT light default schema,
# state_value_template's output must match payload_on/payload_off
# exactly (case-sensitive) for HA to recognise the state.
cfg = {
'name': 'Lamp',
'unique_id': f"{topic_prefix}_lamp_light",
'object_id': f"{topic_prefix}_lamp_light",
'state_topic': state_topic,
'state_value_template': "{{ value_json.lamp }}",
'command_topic': f"{topic_prefix}/{CMD_LAMP}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': 'mdi:track-light',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/light/{topic_prefix}/lamp/config",
encode(cfg)))
# --- switches. Sound is always-available — independent of cycle
# state, no RC required. Fast preheat + Natural steam are
# options-array writes the oven only honours mid-cycle, so they
# gate on RC + cycle_active. Power deliberately omitted: turning
# the oven on is a physical-panel action; read-only power_state
# sensor reflects its state.
cfg = {
'name': 'Sound',
'unique_id': f"{topic_prefix}_sound_switch",
'object_id': f"{topic_prefix}_sound_switch",
'state_topic': state_topic,
'value_template': '{{ value_json.sound }}',
'state_on': 'On',
'state_off': 'Off',
'command_topic': f"{topic_prefix}/{CMD_SOUND}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': 'mdi:volume-high',
'availability': avail_base(avail_topic),
'device': dev,
}
out.append((f"{ha_prefix}/switch/{topic_prefix}/sound/config",
encode(cfg)))
cycle_switches = [
('fastpreheat', 'Fast preheat', '{{ value_json.fastpreheat }}', CMD_FASTPREHEAT, 'mdi:fire'),
('natural_steam', 'Natural steam', '{{ value_json.natural_steam }}', CMD_NATURALSTEAM, 'mdi:kettle-steam'),
]
for key, name, tpl, cmd, icon in cycle_switches:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}_switch",
'object_id': f"{topic_prefix}_{key}_switch",
'state_topic': state_topic,
'value_template': tpl,
'state_on': 'On',
'state_off': 'Off',
'command_topic': f"{topic_prefix}/{cmd}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': icon,
'availability': avail_with_remote_and_cycle(
avail_topic, remote_topic, cycle_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/switch/{topic_prefix}/{key}/config",
encode(cfg)))
# --- number: setpoint (RC-gated, slider input) ------------------
cfg = {
'name': 'Setpoint',
'unique_id': f"{topic_prefix}_setpoint",
'object_id': f"{topic_prefix}_setpoint",
'state_topic': state_topic,
'value_template': '{{ value_json.target_temp_c }}',
'command_topic': f"{topic_prefix}/{CMD_SETPOINT}",
'min': SETPOINT_MIN_C,
'max': SETPOINT_MAX_C,
'step': SETPOINT_STEP_C,
'unit_of_measurement': '°C',
'device_class': 'temperature',
'mode': 'slider',
'icon': 'mdi:thermometer-chevron-up',
# RC + cycle_active gated — Samsung's local-OCF surface only
# honours setpoint changes while a cycle is actually running
# (idle writes get rolled back within ~3s).
'availability': avail_with_remote_and_cycle(
avail_topic, remote_topic, cycle_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/number/{topic_prefix}/setpoint/config",
encode(cfg)))
# --- button: Stop cycle ----------------------------------------
# Gated on cycle_active — there's nothing to stop when idle.
# There is no Start button: local-OCF cycle start is not
# reproducible on this firmware (see project_oven_remote_start
# _open.md memory note for the full investigation). Cooking mode
# is similarly omitted — read-only via local OCF, surfaced as a
# sensor.
cfg = {
'name': 'Stop cycle',
'unique_id': f"{topic_prefix}_stop",
'object_id': f"{topic_prefix}_stop",
'command_topic': f"{topic_prefix}/{CMD_STOP}",
'payload_press': 'Stop',
'icon': 'mdi:stop',
'availability': avail_with_cycle(avail_topic, cycle_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/button/{topic_prefix}/stop/config",
encode(cfg)))
# --- number: Cook time in minutes (RC + cycle gated; the oven
# only honours operationTime writes while running). Source of
# truth is `operationTime` on /operational/state/vs/0;
# SmartThings's mid-cycle time changes land in that same field.
cfg = {
'name': 'Cook time',
'unique_id': f"{topic_prefix}_cook_time",
'object_id': f"{topic_prefix}_cook_time",
'state_topic': state_topic,
'value_template': '{{ value_json.operation_time_minutes | int(0) }}',
'command_topic': f"{topic_prefix}/{CMD_COOK_TIME}",
'min': 0,
'max': 1439, # 23:59 — matches modeSpec timeMax
'step': 1,
'unit_of_measurement': 'min',
'mode': 'box',
'icon': 'mdi:timer',
'availability': avail_with_remote_and_cycle(
avail_topic, remote_topic, cycle_topic),
'availability_mode': 'all',
'device': dev,
}
out.append((f"{ha_prefix}/number/{topic_prefix}/cook_time/config",
encode(cfg)))
# --- removal: publish empty payload to the discovery topics of
# entities we used to expose. HA treats an empty retained payload
# on a discovery topic as "delete this entity", so previously-set
# up Start buttons and Cooking-mode selects disappear cleanly.
out.append((f"{ha_prefix}/button/{topic_prefix}/start/config", b''))
out.append((f"{ha_prefix}/select/{topic_prefix}/mode/config", b''))
return out
# ---------------------------------------------------------------------
# Command handlers — fn(payload, links) → (path_segs, body_dict) | None.
# Read-modify-write handlers (lamp/sound/fastpreheat) snapshot the
# `/mode/vs/0` options array and replace just their slot. /temperatures
# is also RMW because Samsung's write semantics on the items array are
# replace-not-merge.
# ---------------------------------------------------------------------
def _mode_options(links):
"""Return the live `/mode/vs/0` options array (a copy), or None
if /mode/vs/0 isn't seeded yet."""
rep = links.get('/mode/vs/0') or {}
opts = rep.get('x.com.samsung.da.options')
if not opts:
return None
return list(opts)
def _temps_items(links):
"""Return a deep-ish copy of the /temperatures/vs/0 items array."""
rep = links.get('/temperatures/vs/0') or {}
items = rep.get('x.com.samsung.da.items') or []
return [dict(it) for it in items] if items else None
def command_handlers():
def _lamp(p, links):
if p not in ('On', 'Off'):
return None
opts = _mode_options(links)
if opts is None:
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'UpperLamp', p),
}
def _sound(p, links):
if p not in ('On', 'Off'):
return None
opts = _mode_options(links)
if opts is None:
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(opts, 'Sound', p),
}
def _fastpreheat(p, links):
if p not in ('On', 'Off'):
return None
opts = _mode_options(links)
if opts is None:
return None
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': _replace_in_options(
opts, 'fastpreheat', p),
}
def _naturalsteam(p, links):
# NaturalSteam_* only appears in the options array after the
# SmartThings app has touched it once. If absent, append the
# slot — the oven creates it on first write, so the bridge
# doesn't need a "prime via app" dance.
if p not in ('On', 'Off'):
return None
opts = _mode_options(links)
if opts is None:
return None
if not any(o.startswith('NaturalSteam_') for o in opts):
opts = opts + [f'NaturalSteam_{p}']
else:
opts = _replace_in_options(opts, 'NaturalSteam', p)
return ['mode', 'vs', '0'], {
'x.com.samsung.da.options': opts,
}
def _power(p, _links):
if p not in ('On', 'Off'):
return None
return ['power', 'vs', '0'], {'x.com.samsung.da.power': p}
def _stop(_p, _links):
return ['operational', 'state', 'vs', '0'], {
'x.com.samsung.da.state': 'Ready',
}
def _setpoint(p, links):
try:
temp = float(p)
except (TypeError, ValueError):
return None
temp_i = int(round(temp / SETPOINT_STEP_C) * SETPOINT_STEP_C)
if not (SETPOINT_MIN_C <= temp_i <= SETPOINT_MAX_C):
return None
items = _temps_items(links)
if items is None:
return None
items[0]['x.com.samsung.da.desired'] = str(temp_i)
return ['temperatures', 'vs', '0'], {
'x.com.samsung.da.items': items,
}
def _cook_time(p, links):
# HA sends minutes; oven cycle duration lives in
# /operational/state/vs/0 as `operationTime` / `remainingTime`
# (H:MM:SS strings). Writing both — mirroring SmartThings's
# observed behaviour, which resets the live countdown to the
# new duration. Clamp to modeSpec's 0..23:59.
try:
minutes = int(round(float(p)))
except (TypeError, ValueError):
return None
if not (0 <= minutes <= 1439):
return None
h, m = divmod(minutes, 60)
hms = f"{h:02d}:{m:02d}:00"
return ['operational', 'state', 'vs', '0'], {
'x.com.samsung.da.operationTime': hms,
'x.com.samsung.da.remainingTime': hms,
}
return {
CMD_LAMP: _lamp,
CMD_SOUND: _sound,
CMD_FASTPREHEAT: _fastpreheat,
CMD_NATURALSTEAM: _naturalsteam,
CMD_POWER: _power,
CMD_STOP: _stop,
CMD_SETPOINT: _setpoint,
CMD_COOK_TIME: _cook_time,
}
# --- Poll tiers --------------------------------------------------------
# Empirical ceiling on this firmware is ~8 req/s (probe_poll_rate_combined.py
# 2026-06-03). Hot tier covers what changes mid-cook; doors get the tightest
# cadence because door open/close needs sub-second freshness in HA.
# Per-tier timeouts are scaled to cadence: hot tier retries every 1s, so
# a tight 2s ceiling caps the cascade damage from one wedged poll. Warm
# and cold tiers have more headroom; sweep is multi-block Block2 and
# tolerates ~15s.
OVEN_POLL_TIERS = [
PollTier(
name='hot',
interval_s=1.0,
active_interval_s=0.5,
timeout_s=2.0,
paths=(
('operational', 'state', 'vs', '0'),
('doors', 'vs', '0'),
('oven', 'vs', '0'),
('temperatures', 'vs', '0'),
),
),
PollTier(
name='warm',
interval_s=30.0,
timeout_s=4.0,
paths=(
('power', 'vs', '0'),
('kidslock', 'vs', '0'),
('remotectrl', 'vs', '0'),
('mode', 'vs', '0'),
('alarms', 'vs', '0'),
('connected', 'vs', '0'),
),
),
PollTier(
name='cold',
interval_s=600.0,
timeout_s=6.0,
paths=(
('otninformation', 'vs', '0'),
),
),
PollTier(
name='sweep',
interval_s=300.0,
timeout_s=15.0,
paths=(('device', '0'),),
is_sweep=True,
),
]
def _is_active(links: dict) -> bool:
rep = links.get('/operational/state/vs/0') or {}
sam_state = rep.get('x.com.samsung.da.state')
return _SAMSUNG_STATE_TO_OCF.get(sam_state) == 'active'
# ---------------------------------------------------------------------
OVEN = ApplianceDescriptor(
name='oven',
default_observe_port=49154,
observe_paths=OBSERVE_PATHS,
seed_path=['device', '0'],
flatten=flatten,
build_discovery=build_discovery,
command_handlers=command_handlers,
on_observation=on_observation,
project=project,
remote_available_field='remote_control_binary',
cycle_active_field='cycle_active',
log_state_change=log_state_change,
poll_tiers=OVEN_POLL_TIERS,
is_active=_is_active,
)
samsung_appliance/appliances/refrigerator.py
-520
View File
@@ -1,520 +0,0 @@
"""Refrigerator descriptor (Samsung RF9000B-class / TP1X_REF_21K).
Resource map captured 2026-06-27 via DTLS-CoAP with the ab0b0ac4 cert
from a unit at 10.0.0.254:49154.
See local-tools/dumps/10.0.0.254.json for the full field reference.
This appliance has no "cycle" concept — there is no machine_state and
no remote-control gate. All controls are always available when the bridge
is online.
Temperature units: this model reports in Fahrenheit.
Freezer setpoint range: -8°F to 5°F
Fridge setpoint range: 34°F to 44°F
Write surfaces (all unproven — modelled on oven/dryer write semantics,
first HA interaction is the live test):
* Freezer and fridge setpoints via /temperatures/vs/0 items RMW
(same pattern as oven /temperatures/vs/0)
* Rapid fridge / rapid freezing via /refrigeration/vs/0
* Ice maker 1 (cubed) on/off via /icemaker/one/vs/0
* Ice maker 2 (whiskey ball) type select via /icemaker/two/vs/0
* Autofill via /autofill/vs/0
* Cabinet light on/off via /cabinet/light/total/vs/0
* Beverage zone mode select via /specialzone/one/vs/0
"""
from .base import (
ApplianceDescriptor,
avail_base,
device_block,
encode,
)
from ..poll_scheduler import PollTier
# --- OBSERVE paths --------------------------------------------------------
OBSERVE_PATHS = [
['doors', 'vs', '0'], # door open states (2 doors)
['temperatures', 'vs', '0'], # current + desired temp, both zones
['icemaker', 'status', 'vs', '0'], # global ice maker enabled
['icemaker', 'one', 'vs', '0'], # cubed ice maker state + type
['icemaker', 'two', 'vs', '0'], # whiskey ball ice maker type
['refrigeration', 'vs', '0'], # rapidFridge, rapidFreezing
['energy', 'consumption', 'vs', '0'],
['alarms', 'vs', '0'],
['cabinet', 'light', 'total', 'vs', '0'],
['autofill', 'vs', '0'],
['filter', 'waterfilter', 'vs', '0'],
['sabbath', 'vs', '0'],
['status', 'lock', 'vs', '0'], # device sound + ADO control
]
# Setpoint bounds read from /temperatures/vs/0 items on 2026-06-27.
FREEZER_MIN_F = -8
FREEZER_MAX_F = 5
FRIDGE_MIN_F = 34
FRIDGE_MAX_F = 44
# Whiskey ball ice sizes available on this model.
ICE2_MODES = ['Off', 'WHISKEY_ICEBALL_3', 'WHISKEY_ICEBALL_6', 'WHISKEY_ICEBALL_9']
# Beverage zone flex modes on this model.
BZONE_MODES = ['SP_TTYPE_BEER_DRINKS', 'SP_TTYPE_WINE_DESSERT']
def _int(v):
try:
return int(v)
except (TypeError, ValueError):
return None
def _num(v):
try:
return float(v)
except (TypeError, ValueError):
return None
# --- flatten --------------------------------------------------------------
def flatten(links):
g = lambda href, k, default=None: (links.get(href) or {}).get(k, default)
# Temperatures — items[0]=Freezer, items[1]=Fridge.
temps_items = g('/temperatures/vs/0', 'x.com.samsung.da.items') or []
freezer_cur = freezer_des = fridge_cur = fridge_des = None
if len(temps_items) > 0:
freezer_cur = _int(temps_items[0].get('x.com.samsung.da.current'))
freezer_des = _int(temps_items[0].get('x.com.samsung.da.desired'))
if len(temps_items) > 1:
fridge_cur = _int(temps_items[1].get('x.com.samsung.da.current'))
fridge_des = _int(temps_items[1].get('x.com.samsung.da.desired'))
# Doors — items[0] and [1], openState 'Open'/'Close'.
doors_items = g('/doors/vs/0', 'x.com.samsung.da.items') or []
door0_open = door1_open = None
if len(doors_items) > 0:
s = doors_items[0].get('x.com.samsung.da.openState')
door0_open = (s == 'Open') if s is not None else None
if len(doors_items) > 1:
s = doors_items[1].get('x.com.samsung.da.openState')
door1_open = (s == 'Open') if s is not None else None
any_door_open = bool(door0_open or door1_open)
# Ice makers.
ice_enabled = g('/icemaker/status/vs/0', 'x.com.samsung.da.iceMaker')
ice1_state = g('/icemaker/one/vs/0', 'x.com.samsung.da.iceMaker.state')
ice1_status = g('/icemaker/one/vs/0', 'x.com.samsung.da.iceMaker.iceMakingStatus')
ice1_on = (ice1_state == 'On') if ice1_state is not None else None
ice2_type = g('/icemaker/two/vs/0', 'x.com.samsung.da.iceType.desired')
ice2_status = g('/icemaker/two/vs/0', 'x.com.samsung.da.iceMaker.iceMakingStatus')
# Modes.
rapid_fridge = g('/refrigeration/vs/0', 'x.com.samsung.da.rapidFridge')
rapid_freezing = g('/refrigeration/vs/0', 'x.com.samsung.da.rapidFreezing')
autofill = g('/autofill/vs/0', 'x.com.samsung.da.autofill')
sabbath = g('/sabbath/vs/0', 'x.com.samsung.da.sabbathMode')
# Cabinet light.
light_control = g('/cabinet/light/total/vs/0', 'x.com.samsung.da.lightControl')
light_on = (light_control == 'On') if light_control is not None else None
# Filter.
filter_usage = _int(g('/filter/waterfilter/vs/0', 'x.com.samsung.da.filterUsage'))
filter_status = g('/filter/waterfilter/vs/0', 'x.com.samsung.da.filterStatus')
# Beverage zone.
bzone_rep = links.get('/specialzone/one/vs/0') or {}
bzone_mode = bzone_rep.get('roomDesiredMode')
# Energy.
inst_w = _num(g('/energy/consumption/vs/0', 'x.com.samsung.da.instantaneousPower'))
cum_wh = _num(g('/energy/consumption/vs/0', 'x.com.samsung.da.cumulativePower'))
# Alarms — empty array when nothing is active on this model.
alarm_items = g('/alarms/vs/0', 'x.com.samsung.da.items') or []
alarm_code = (alarm_items[-1].get('x.com.samsung.da.code')
if alarm_items else None)
# Firmware update.
fw_update = g('/otninformation/vs/0', 'x.com.samsung.da.newVersionAvailable')
fw_update_bin = (str(fw_update).lower() == 'true'
if fw_update is not None else None)
return {
'freezer_temp_f': freezer_cur,
'freezer_setpoint_f': freezer_des,
'fridge_temp_f': fridge_cur,
'fridge_setpoint_f': fridge_des,
'door_freezer_open': door0_open,
'door_fridge_open': door1_open,
'any_door_open': any_door_open,
'ice_maker_enabled': ice_enabled,
'ice1_state': ice1_state,
'ice1_on': ice1_on,
'ice1_making_status': ice1_status,
'ice2_type': ice2_type,
'ice2_making_status': ice2_status,
'rapid_fridge': rapid_fridge,
'rapid_freezing': rapid_freezing,
'autofill': autofill,
'sabbath_mode': sabbath,
'cabinet_light': light_control,
'cabinet_light_on': light_on,
'filter_usage': filter_usage,
'filter_status': filter_status,
'beverage_zone_mode': bzone_mode,
'power_watts': max(0.0, inst_w) if inst_w is not None else None,
'energy_kwh': round(cum_wh / 1000.0, 2)
if cum_wh is not None else None,
'alarm_code': alarm_code,
'firmware_update': fw_update_bin,
}
def log_state_change(sensors):
return (f"freezer={sensors.get('freezer_temp_f')}°F "
f"fridge={sensors.get('fridge_temp_f')}°F "
f"door={'open' if sensors.get('any_door_open') else 'closed'} "
f"power={sensors.get('power_watts')}W")
# --- HA discovery ---------------------------------------------------------
MODEL = 'OCF refrigerator (TizenRT-iotivity, RF9000B-class)'
_SENSORS = [
('freezer_temp_f', 'Freezer temperature',
{'unit_of_measurement': '°F', 'device_class': 'temperature',
'state_class': 'measurement'}),
('freezer_setpoint_f', 'Freezer setpoint',
{'unit_of_measurement': '°F', 'device_class': 'temperature',
'icon': 'mdi:thermometer-chevron-down'}),
('fridge_temp_f', 'Fridge temperature',
{'unit_of_measurement': '°F', 'device_class': 'temperature',
'state_class': 'measurement'}),
('fridge_setpoint_f', 'Fridge setpoint',
{'unit_of_measurement': '°F', 'device_class': 'temperature',
'icon': 'mdi:thermometer-chevron-up'}),
('ice1_making_status', 'Ice maker (cubed) status', {'icon': 'mdi:cube'}),
('ice2_making_status', 'Ice maker (whiskey) status', {'icon': 'mdi:circle'}),
('filter_usage', 'Water filter usage',
{'unit_of_measurement': '%', 'state_class': 'measurement',
'icon': 'mdi:filter'}),
('filter_status', 'Water filter status', {'icon': 'mdi:filter-check'}),
('power_watts', 'Power',
{'unit_of_measurement': 'W', 'device_class': 'power',
'state_class': 'measurement'}),
('energy_kwh', 'Energy',
{'unit_of_measurement': 'kWh', 'device_class': 'energy',
'state_class': 'total_increasing'}),
('alarm_code', 'Alarm code',
{'icon': 'mdi:alert', 'entity_category': 'diagnostic'}),
]
_BINARY_SENSORS = [
('door_freezer_open', 'Freezer door',
"{{ 'ON' if value_json.door_freezer_open else 'OFF' }}", 'door'),
('door_fridge_open', 'Fridge door',
"{{ 'ON' if value_json.door_fridge_open else 'OFF' }}", 'door'),
('ice1_on', 'Ice maker (cubed)',
"{{ 'ON' if value_json.ice1_on else 'OFF' }}", 'running'),
('firmware_update', 'Firmware update available',
"{{ 'ON' if value_json.firmware_update else 'OFF' }}", 'update'),
]
CMD_FREEZER_SETPOINT = 'cmd/freezer_setpoint'
CMD_FRIDGE_SETPOINT = 'cmd/fridge_setpoint'
CMD_RAPID_FRIDGE = 'cmd/rapid_fridge'
CMD_RAPID_FREEZING = 'cmd/rapid_freezing'
CMD_ICE1 = 'cmd/ice1'
CMD_ICE2_TYPE = 'cmd/ice2_type'
CMD_AUTOFILL = 'cmd/autofill'
CMD_CABINET_LIGHT = 'cmd/cabinet_light'
CMD_BZONE_MODE = 'cmd/beverage_zone_mode'
def build_discovery(topic_prefix, ha_prefix, device_name):
state_topic = f"{topic_prefix}/state"
avail_topic = f"{topic_prefix}/availability"
dev = device_block(topic_prefix, device_name, MODEL)
avail = avail_base(avail_topic)
out = []
for key, name, extra in _SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': f"{{{{ value_json.{key} }}}}",
'availability': avail,
'device': dev,
}
cfg.update(extra)
out.append((f"{ha_prefix}/sensor/{topic_prefix}/{key}/config",
encode(cfg)))
for key, name, template, dclass in _BINARY_SENSORS:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}",
'object_id': f"{topic_prefix}_{key}",
'state_topic': state_topic,
'value_template': template,
'payload_on': 'ON',
'payload_off': 'OFF',
'device_class': dclass,
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/binary_sensor/{topic_prefix}/{key}/config",
encode(cfg)))
# number: freezer setpoint
cfg = {
'name': 'Freezer setpoint',
'unique_id': f"{topic_prefix}_freezer_setpoint_num",
'object_id': f"{topic_prefix}_freezer_setpoint_num",
'state_topic': state_topic,
'value_template': '{{ value_json.freezer_setpoint_f }}',
'command_topic': f"{topic_prefix}/{CMD_FREEZER_SETPOINT}",
'min': FREEZER_MIN_F,
'max': FREEZER_MAX_F,
'step': 1,
'unit_of_measurement': '°F',
'device_class': 'temperature',
'mode': 'slider',
'icon': 'mdi:thermometer-chevron-down',
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/number/{topic_prefix}/freezer_setpoint/config",
encode(cfg)))
# number: fridge setpoint
cfg = {
'name': 'Fridge setpoint',
'unique_id': f"{topic_prefix}_fridge_setpoint_num",
'object_id': f"{topic_prefix}_fridge_setpoint_num",
'state_topic': state_topic,
'value_template': '{{ value_json.fridge_setpoint_f }}',
'command_topic': f"{topic_prefix}/{CMD_FRIDGE_SETPOINT}",
'min': FRIDGE_MIN_F,
'max': FRIDGE_MAX_F,
'step': 1,
'unit_of_measurement': '°F',
'device_class': 'temperature',
'mode': 'slider',
'icon': 'mdi:thermometer-chevron-up',
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/number/{topic_prefix}/fridge_setpoint/config",
encode(cfg)))
# switches: rapid fridge, rapid freezing, autofill, cabinet light, ice maker 1
for key, name, tpl, cmd, icon in [
('rapid_fridge', 'Rapid fridge', '{{ value_json.rapid_fridge }}',
CMD_RAPID_FRIDGE, 'mdi:fridge-industrial'),
('rapid_freezing', 'Rapid freezing', '{{ value_json.rapid_freezing }}',
CMD_RAPID_FREEZING, 'mdi:snowflake'),
('autofill', 'Autofill', '{{ value_json.autofill }}',
CMD_AUTOFILL, 'mdi:cup-water'),
('cabinet_light', 'Cabinet light', '{{ value_json.cabinet_light }}',
CMD_CABINET_LIGHT, 'mdi:fridge-outline'),
('ice1_switch', 'Ice maker (cubed)', '{{ value_json.ice1_state }}',
CMD_ICE1, 'mdi:cube'),
]:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_{key}_switch",
'object_id': f"{topic_prefix}_{key}_switch",
'state_topic': state_topic,
'value_template': tpl,
'state_on': 'On',
'state_off': 'Off',
'command_topic': f"{topic_prefix}/{cmd}",
'payload_on': 'On',
'payload_off': 'Off',
'icon': icon,
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/switch/{topic_prefix}/{key}/config",
encode(cfg)))
# select: whiskey ball ice type
cfg = {
'name': 'Whiskey ball ice type',
'unique_id': f"{topic_prefix}_ice2_type_select",
'object_id': f"{topic_prefix}_ice2_type_select",
'state_topic': state_topic,
'value_template': '{{ value_json.ice2_type }}',
'command_topic': f"{topic_prefix}/{CMD_ICE2_TYPE}",
'options': ICE2_MODES,
'icon': 'mdi:circle',
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/select/{topic_prefix}/ice2_type/config",
encode(cfg)))
# select: beverage zone mode
cfg = {
'name': 'Beverage zone mode',
'unique_id': f"{topic_prefix}_bzone_mode_select",
'object_id': f"{topic_prefix}_bzone_mode_select",
'state_topic': state_topic,
'value_template': '{{ value_json.beverage_zone_mode }}',
'command_topic': f"{topic_prefix}/{CMD_BZONE_MODE}",
'options': BZONE_MODES,
'icon': 'mdi:glass-wine',
'availability': avail,
'device': dev,
}
out.append((f"{ha_prefix}/select/{topic_prefix}/bzone_mode/config",
encode(cfg)))
return out
# --- Command handlers -----------------------------------------------------
def _temps_items(links):
rep = links.get('/temperatures/vs/0') or {}
items = rep.get('x.com.samsung.da.items') or []
return [dict(it) for it in items] if items else None
def command_handlers():
def _freezer_setpoint(p, links):
try:
temp = int(round(float(p)))
except (TypeError, ValueError):
return None
if not (FREEZER_MIN_F <= temp <= FREEZER_MAX_F):
return None
items = _temps_items(links)
if not items:
return None
items[0]['x.com.samsung.da.desired'] = str(temp)
return ['temperatures', 'vs', '0'], {'x.com.samsung.da.items': items}
def _fridge_setpoint(p, links):
try:
temp = int(round(float(p)))
except (TypeError, ValueError):
return None
if not (FRIDGE_MIN_F <= temp <= FRIDGE_MAX_F):
return None
items = _temps_items(links)
if not items or len(items) < 2:
return None
items[1]['x.com.samsung.da.desired'] = str(temp)
return ['temperatures', 'vs', '0'], {'x.com.samsung.da.items': items}
def _on_off(key, path_segs):
def _handler(p, _links):
if p not in ('On', 'Off'):
return None
return path_segs, {key: p}
return _handler
def _ice1(p, _links):
if p not in ('On', 'Off'):
return None
return ['icemaker', 'one', 'vs', '0'], {
'x.com.samsung.da.iceMaker.state': p,
}
def _ice2_type(p, _links):
if p not in ICE2_MODES:
return None
return ['icemaker', 'two', 'vs', '0'], {
'x.com.samsung.da.iceType.desired': p,
}
def _bzone_mode(p, _links):
if p not in BZONE_MODES:
return None
return ['specialzone', 'one', 'vs', '0'], {'roomDesiredMode': p}
return {
CMD_FREEZER_SETPOINT: _freezer_setpoint,
CMD_FRIDGE_SETPOINT: _fridge_setpoint,
CMD_RAPID_FRIDGE: _on_off('x.com.samsung.da.rapidFridge',
['refrigeration', 'vs', '0']),
CMD_RAPID_FREEZING: _on_off('x.com.samsung.da.rapidFreezing',
['refrigeration', 'vs', '0']),
CMD_ICE1: _ice1,
CMD_ICE2_TYPE: _ice2_type,
CMD_AUTOFILL: _on_off('x.com.samsung.da.autofill',
['autofill', 'vs', '0']),
CMD_CABINET_LIGHT: _on_off('x.com.samsung.da.lightControl',
['cabinet', 'light', 'total', 'vs', '0']),
CMD_BZONE_MODE: _bzone_mode,
}
# --- Poll tiers -----------------------------------------------------------
# No empirical ceiling measurement on this device yet. Using a conservative
# ~8 req/s budget (matching oven) until a probe is run.
# Doors on the hot tier for sub-second open/close alerts.
FRIDGE_POLL_TIERS = [
PollTier(
name='hot',
interval_s=1.0,
paths=(
('doors', 'vs', '0'),
),
),
PollTier(
name='warm',
interval_s=30.0,
paths=(
('temperatures', 'vs', '0'),
('icemaker', 'status', 'vs', '0'),
('icemaker', 'one', 'vs', '0'),
('icemaker', 'two', 'vs', '0'),
('refrigeration', 'vs', '0'),
('energy', 'consumption', 'vs', '0'),
('filter', 'waterfilter', 'vs', '0'),
('autofill', 'vs', '0'),
('cabinet', 'light', 'total', 'vs', '0'),
('sabbath', 'vs', '0'),
('status', 'lock', 'vs', '0'),
('specialzone', 'one', 'vs', '0'),
('alarms', 'vs', '0'),
),
),
PollTier(
name='cold',
interval_s=600.0,
paths=(
('otninformation', 'vs', '0'),
),
),
PollTier(
name='sweep',
interval_s=600.0,
paths=(('device', '0'),),
is_sweep=True,
),
]
# --- Descriptor -----------------------------------------------------------
REFRIGERATOR = ApplianceDescriptor(
name='refrigerator',
default_observe_port=49154,
observe_paths=OBSERVE_PATHS,
seed_path=['device', '0'],
flatten=flatten,
build_discovery=build_discovery,
command_handlers=command_handlers,
log_state_change=log_state_change,
poll_tiers=FRIDGE_POLL_TIERS,
# No remote_available_field — fridge has no remote-control gate.
# No cycle_active_field — fridge has no cycle concept.
# No is_active — always uses normal poll interval.
)
samsung_appliance/bridge.py
+178 -48
View File
@@ -23,7 +23,6 @@ import time
import cbor2
from .appliances.base import ApplianceDescriptor, bridge_diagnostic_discovery
from .coap_dtls import DtlsCoapSession, fmt_code
from .config import ApplianceConfig, SharedConfig
from .keepalive import KeepaliveTask
@@ -36,6 +35,10 @@ from .state_cache import StateCache
DEBUG_BRIDGE = os.environ.get('DEBUG_BRIDGE') == '1'
# /device/0 is the OCF resource that contains the full link dict used
# to seed the state cache on every connect.
SEED_PATH = ['device', '0']
def _href_to_segs(href: str) -> list[str]:
return [s for s in href.split('/') if s]
@@ -60,29 +63,114 @@ UNREACHABLE_RECONNECT_S = 120.0
OBSERVE_REFRESH_INTERVAL_S = 6 * 3600.0
def _encode(cfg: dict) -> bytes:
return json.dumps(cfg).encode()
def _bridge_diagnostic_discovery(topic_prefix: str,
ha_discovery_prefix: str,
device_name: str,
model: str = 'Bridge') -> list[tuple[str, bytes]]:
"""HA-discovery payloads for the per-bridge diagnostic entities
(push state, polling RTT, freshness). Returned as a list of
(topic, payload-bytes) tuples to be merged with entity discovery."""
avail_topic = f"{topic_prefix}/availability"
health_topic = f"{topic_prefix}/bridge/health"
push_topic = f"{topic_prefix}/bridge/push_active"
device = {
'identifiers': [topic_prefix],
'name': device_name,
'manufacturer': 'Samsung',
'model': model,
}
avail = [{'topic': avail_topic,
'payload_available': 'online',
'payload_not_available': 'offline'}]
def sensor(slug: str, name: str, value_template: str,
unit: str | None = None, device_class: str | None = None,
icon: str | None = None) -> tuple[str, bytes]:
cfg = {
'name': name,
'unique_id': f"{topic_prefix}_bridge_{slug}",
'object_id': f"{topic_prefix}_bridge_{slug}",
'state_topic': health_topic,
'value_template': value_template,
'availability': avail,
'device': device,
'entity_category': 'diagnostic',
}
if unit is not None: cfg['unit_of_measurement'] = unit
if device_class is not None: cfg['device_class'] = device_class
if icon is not None: cfg['icon'] = icon
topic = (f"{ha_discovery_prefix}/sensor/"
f"{topic_prefix}/bridge_{slug}/config")
return topic, _encode(cfg)
push_active_cfg = {
'name': 'Push Active',
'unique_id': f"{topic_prefix}_bridge_push_active",
'object_id': f"{topic_prefix}_bridge_push_active",
'state_topic': push_topic,
'payload_on': 'online',
'payload_off': 'offline',
'availability': avail,
'device': device,
'entity_category': 'diagnostic',
'icon': 'mdi:rss',
}
push_active_topic = (f"{ha_discovery_prefix}/binary_sensor/"
f"{topic_prefix}/bridge_push_active/config")
return [
(push_active_topic, _encode(push_active_cfg)),
sensor('update_source', 'Last Update Source',
"{{ value_json.last_change_source | default('?') }}",
icon='mdi:transit-connection-variant'),
sensor('stalest_age_s', 'Stalest Resource Age',
"{{ value_json.stalest_age_s | default(0) }}",
unit='s', icon='mdi:clock-alert-outline'),
sensor('poll_max_rtt_ms', 'Poll Max RTT',
"{{ value_json.poll_window_max_rtt_ms | default(0) | int }}",
unit='ms', icon='mdi:speedometer'),
sensor('slow_polls', 'Slow Polls (window)',
"{{ value_json.poll_window_slow_count | default(0) }}",
icon='mdi:timer-sand'),
sensor('poll_errors', 'Poll Errors (window)',
"{{ value_json.poll_window_errors | default(0) }}",
icon='mdi:alert-circle-outline'),
sensor('observe_age_s', 'Last OBSERVE Age',
"{{ value_json.last_observe_age_s if value_json.last_observe_age_s is not none else 'never' }}",
icon='mdi:radar'),
]
class PushBridge:
def __init__(self,
shared: SharedConfig,
app: ApplianceConfig,
descriptor: ApplianceDescriptor,
mqtt_client):
self.shared = shared
self.app = app
self.descriptor = descriptor
self.mqtt = mqtt_client
self.log = bridge_logger(app.klass)
self._serial: str | None = None
self.port = app.ocf_port or descriptor.default_observe_port
self.port = app.ocf_port or 49154
self.session: DtlsCoapSession | None = None
self.scheduler: PollScheduler | None = None
self.keepalive: KeepaliveTask | None = None
self.observe_refresh: ObserveRefreshTask | None = None
self.cache = StateCache(descriptor)
# descriptor is None until _discover_and_publish completes after seed.
self.descriptor = None
self._discovered = False
self._entity_payloads: list = []
self.cache = StateCache()
self.cache.set_on_change(self._on_cache_change)
self.last_state_pub = None
@@ -132,16 +220,9 @@ class PushBridge:
self.cycle_topic = f"{p}/cycle_active"
self.health_topic = f"{p}/bridge/health"
self.push_active_topic = f"{p}/bridge/push_active"
self.cmd_handlers = descriptor.command_handlers()
self.cmd_handlers = {}
self.cmd_topic_prefix = f"{p}/cmd/"
self.discovery_payloads = (
descriptor.build_discovery(
app.topic_prefix, shared.HA_DISCOVERY_PREFIX, app.device_name)
+ bridge_diagnostic_discovery(
app.topic_prefix, shared.HA_DISCOVERY_PREFIX, app.device_name,
model=descriptor.name.title()))
# ---- cache plumbing ---------------------------------------------
def _on_cache_change(self, changed: bool, source: str) -> None:
@@ -248,8 +329,7 @@ class PushBridge:
self._unreachable_since = None
self._force_close_in_flight = False
self.log.info("DTLS connected — subscribing %d paths",
len(self.descriptor.observe_paths))
self.log.info("DTLS connected — port %d", self.port)
sess.start_reader()
@@ -273,19 +353,25 @@ class PushBridge:
session_ended.set()
def _run_session_inner(self, sess):
for path in self.descriptor.observe_paths:
sess.subscribe(path)
time.sleep(0.05)
# Inline seed so the publish gate opens before the scheduler's
# first tick. The scheduler's sweep tier will refresh /device/0
# on its own cadence afterwards.
# Seed first — populate cache before discovery can run.
# The publish gate stays closed so seed writes don't trigger
# per-resource publishes before the full snapshot is ready.
self._seed_from_device0(sess)
self._retag_logger_with_serial()
if DEBUG_BRIDGE:
self._debug_dump_links(sess)
# Discover entities from the seeded cache, build the runtime
# descriptor, and publish HA discovery payloads. Idempotent:
# on reconnect the descriptor is reused from the first session.
self._discover_and_publish(sess)
# Now subscribe to observe paths (descriptor populated above).
for path in self.descriptor.observe_paths:
sess.subscribe(path)
time.sleep(0.05)
self._publish_gate = True
self.maybe_publish_state(force=True)
self.set_availability(True)
@@ -344,15 +430,15 @@ class PushBridge:
self.observe_refresh = None
def _seed_from_device0(self, sess):
code, pl = sess.get(self.descriptor.seed_path, timeout=15.0)
code, pl = sess.get(SEED_PATH, timeout=15.0)
if code != 0x45:
raise RuntimeError(
f"/{'/'.join(self.descriptor.seed_path)} -> {fmt_code(code)}")
f"/{'/'.join(SEED_PATH)} -> {fmt_code(code)}")
try:
body = cbor2.loads(pl)
except Exception as e:
raise RuntimeError(
f"/{'/'.join(self.descriptor.seed_path)} cbor decode: {e}"
f"/{'/'.join(SEED_PATH)} cbor decode: {e}"
) from e
# During the seed we want the cache populated without triggering
# a publish per resource — gate the on_change callback off until
@@ -362,6 +448,40 @@ class PushBridge:
self.cache.apply_rep(href, rep, source='seed')
self.last_seed_ts = time.time()
def _discover_and_publish(self, sess):
"""Run entity discovery against the seeded cache and publish HA
discovery payloads. Idempotent — on reconnect the descriptor
built during the first session is reused; entity payloads are
re-published via reassert_availability instead."""
if self._discovered:
return
from .registry import CAPABILITIES, discover, build_runtime_descriptor
from .registry.identity import read_identity
snapshot = self.cache.snapshot()
bound = discover(snapshot, CAPABILITIES,
log=lambda m: self.log.debug("%s", m))
ident = read_identity(sess, self._serial)
model = ident.model or self.app.klass.title() or 'Samsung Appliance'
device_name = self.app.device_name or ident.name or model
self.descriptor = build_runtime_descriptor(
bound,
topic_prefix=self.app.topic_prefix,
ha_prefix=self.shared.HA_DISCOVERY_PREFIX,
device_name=device_name,
model=model,
name=(self.app.klass or 'appliance'),
default_port=self.port)
self.cache.set_on_observation(self.descriptor.on_observation)
self.cmd_handlers = self.descriptor.command_handlers()
diag_payloads = _bridge_diagnostic_discovery(
self.app.topic_prefix, self.shared.HA_DISCOVERY_PREFIX,
device_name, model=model)
self._entity_payloads = self.descriptor.discovery_payloads + diag_payloads
for topic, payload in self._entity_payloads:
self.mqtt.publish(topic, payload, qos=1, retain=True)
self._discovered = True
self.log.info("discovered %d entities", len(self.descriptor.discovery_payloads))
def _debug_dump_links(self, sess):
for href, rep in sorted(self.cache.links.items()):
if href == '/mode/vs/0':
@@ -417,6 +537,8 @@ class PushBridge:
# ---- MQTT publishing --------------------------------------------
def maybe_publish_state(self, force=False):
if self.descriptor is None:
return
if not force and not self._publish_gate:
return
snap = self.cache.snapshot()
@@ -479,14 +601,21 @@ class PushBridge:
if self.session is None or self.last_state_pub is None:
return
self.set_availability(True)
field = self.descriptor.remote_available_field
if field is not None:
self.publish_remote_available(
self.last_state_pub.get(field), force=True)
cycle_field = self.descriptor.cycle_active_field
if cycle_field is not None:
self.publish_cycle_active(
self.last_state_pub.get(cycle_field), force=True)
if self._discovered:
for topic, payload in self._entity_payloads:
try:
self.mqtt.publish(topic, payload, qos=1, retain=True)
except Exception as e:
self.log.warning("re-assert discovery: %s", e)
if self.descriptor is not None:
field = self.descriptor.remote_available_field
if field is not None:
self.publish_remote_available(
self.last_state_pub.get(field), force=True)
cycle_field = self.descriptor.cycle_active_field
if cycle_field is not None:
self.publish_cycle_active(
self.last_state_pub.get(cycle_field), force=True)
now = time.time()
active = (self._last_observe_change_ts is not None
and (now - self._last_observe_change_ts) <= self.push_active_window_s)
@@ -501,20 +630,21 @@ class PushBridge:
self.mqtt.publish(self.avail_topic, value, qos=1, retain=True)
except Exception as e:
self.log.warning("avail publish: %s", e)
if not online and self.descriptor.remote_available_field is not None:
self.last_remote_pub = None
try:
self.mqtt.publish(self.remote_topic, 'offline',
qos=1, retain=True)
except Exception:
pass
if not online and self.descriptor.cycle_active_field is not None:
self.last_cycle_pub = None
try:
self.mqtt.publish(self.cycle_topic, 'offline',
qos=1, retain=True)
except Exception:
pass
if not online and self.descriptor is not None:
if self.descriptor.remote_available_field is not None:
self.last_remote_pub = None
try:
self.mqtt.publish(self.remote_topic, 'offline',
qos=1, retain=True)
except Exception:
pass
if self.descriptor.cycle_active_field is not None:
self.last_cycle_pub = None
try:
self.mqtt.publish(self.cycle_topic, 'offline',
qos=1, retain=True)
except Exception:
pass
if not online:
self._last_push_active_pub = None
try:
@@ -593,7 +723,7 @@ class PushBridge:
h = {
'mode': 'poll+observe',
'device_class': self.descriptor.name,
'device_class': self.app.klass or 'appliance',
'serial': self._serial,
'connect_count': self.connect_count,
'error_count': self.error_count,
samsung_appliance/config.py
+5 -7
View File
@@ -83,25 +83,23 @@ class ApplianceConfig:
just a stable identifier for logs; it does not appear in MQTT
topics or HA discovery (those are keyed off `topic_prefix`)."""
index: int
klass: str # 'dryer', 'oven', …
klass: str # 'dryer', 'oven', … — empty string if unknown
ip: str
ocf_port: Optional[int] # None → descriptor.default_observe_port
ocf_port: Optional[int] # None → 49154 (default DTLS-CoAP port)
topic_prefix: str
device_name: str
@classmethod
def from_env(cls, index: int) -> 'ApplianceConfig':
prefix = f'APPLIANCE_{index}_'
klass = os.getenv(prefix + 'CLASS')
if not klass:
raise ValueError(f"{prefix}CLASS not set")
klass = os.getenv(prefix + 'CLASS') or ''
ip = os.getenv(prefix + 'IP')
if not ip:
raise ValueError(f"{prefix}IP not set")
port_env = os.getenv(prefix + 'OCF_PORT')
port = int(port_env) if port_env else None
topic = os.getenv(prefix + 'TOPIC') or f'samsung_{klass}'
name = os.getenv(prefix + 'NAME') or f'Samsung {klass.title()}'
topic = os.getenv(prefix + 'TOPIC') or (f'samsung_{klass}' if klass else 'samsung_appliance')
name = os.getenv(prefix + 'NAME') or (f'Samsung {klass.title()}' if klass else 'Samsung Appliance')
return cls(
index=index, klass=klass, ip=ip, ocf_port=port,
topic_prefix=topic, device_name=name,