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

fix: log each unknown rt individually in discover()

Browse Source
This commit is contained in:
Marc Billow
2026-06-30 16:32:37 -05:00
parent 8d63610f82
commit 73fff9c2cf
6 changed files with 1033 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TODO_SECURITY.md
+41
View File
@@ -0,0 +1,41 @@
# Security TODOs
Findings from initial security review (2026-06-27). These are hardening issues,
not malicious code — the repo is safe to run. Fix before exposing to untrusted networks.
---
## 1. DTLS Certificate Verification Disabled
**File:** `samsung_appliance/coap_dtls.py:228`
**Severity:** High | **Category:** Authentication Bypass / Crypto
The DTLS context uses `SSL.VERIFY_NONE` with a lambda that accepts any certificate.
The bridge does not verify the appliance's identity.
**Exploit:** LAN attacker ARP-spoofs the appliance IP, presents a self-signed cert,
and either poisons state payloads (e.g. fake oven temp) or intercepts/drops commands.
**Fix:** Use `SSL.VERIFY_PEER` + `ctx.load_verify_locations()` pointing to the Samsung
CA cert already on disk. The `@SECLEVEL=0` SHA-1 workaround can coexist with peer
verification — `VERIFY_NONE` is not required for it.
---
## 2. MQTT — No TLS, No Publisher Authentication
**Files:** `main.py:137`, `samsung_appliance/config.py:68`
**Severity:** High | **Category:** Missing Encryption / Unauthenticated Commands
MQTT connects on plaintext port 1883. `tls_set()` is never called. The bridge
executes any payload published to `<prefix>/cmd/#` — only value range is checked,
not who sent it. Credentials (if set) are sent in cleartext CONNECT packets.
**Exploit:**
- Any LAN host publishes `270` to `samsung_oven/cmd/setpoint` → oven raised to max
- LAN attacker sniffs TCP stream → captures `MQTT_USER`/`MQTT_PASS`
- All appliance state (energy, temp, run state) readable by any subscriber
**Fix:** Call `cli.tls_set()` with the broker CA before connecting. Add
`MQTT_TLS_CA` / `MQTT_TLS_CERT` / `MQTT_TLS_KEY` to the `.env` schema.
Enforce broker-side ACLs so only the bridge credential can publish to `cmd/#`.
samsung_appliance/appliances/__init__.py
+6 -2
View File
@@ -10,13 +10,17 @@ 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] = {
DRYER.name: DRYER,
OVEN.name: OVEN,
DISHWASHER.name: DISHWASHER,
DRYER.name: DRYER,
OVEN.name: OVEN,
REFRIGERATOR.name: REFRIGERATOR,
}
samsung_appliance/appliances/dishwasher.py
+452
View File
@@ -0,0 +1,452 @@
"""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/refrigerator.py
+520
View File
@@ -0,0 +1,520 @@
"""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/registry/discovery.py
+3 -5
View File
@@ -6,7 +6,7 @@ bound to that resource href. Unknown rt is a coverage gap, logged and skipped.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import Callable, Optional
from .capability import Capability
@@ -40,13 +40,11 @@ def discover(resources: dict[str, dict],
for rt in rts:
cap = registry.get(rt)
if cap is None:
if log is not None:
log(f"unknown capability {rt} at {href}")
continue
inst = instance_suffix(href)
for desc in cap.entities:
out.append(BoundEntity(href=href, capability=cap,
desc=desc, instance=inst))
# Coverage logging: a resource none of whose rts are known.
if log is not None and not any(rt in registry for rt in rts):
for rt in rts:
log(f"unknown capability {rt} at {href}")
return out
tests/test_discovery.py
+11
View File
@@ -45,3 +45,14 @@ def test_discover_multi_instance_suffixes():
bound = discover(resources, {cap.rt: cap})
insts = sorted(b.instance for b in bound)
assert insts == ['', '_1']
def test_discover_logs_only_unknown_rt_in_mixed_resource():
seen = []
cap = Capability(rt='x.com.samsung.da.kidsLock',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),))
resources = {'/kidslock/vs/0': {'rt': ['x.com.samsung.da.kidsLock', 'x.com.samsung.da.mystery']}}
bound = discover(resources, {cap.rt: cap}, log=seen.append)
assert len(bound) == 1 # known rt still produces entity
assert any('mystery' in m for m in seen) # unknown rt is logged
assert not any('kidsLock' in m for m in seen) # known rt is NOT logged