Add cooktop and range hood support

This commit is contained in:
themaanda committed 2026-07-21 16:57:45 -05:00
1 parent bedc662b47
commit c64f58aaa5
25 files changed
+1399 -14

No files matched your search

+4 -2
View File
@@ -195,7 +195,9 @@ def _probe_and_validate(host: str, ca_cert_pem: str, ca_key_pem: str) -> dict:
import cbor2
from smartthings_local.protocol.dtls_session import DtlsCoapSession
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device, for_device_by_model
from .registry.by_type import (
for_device, for_device_by_model, for_device_by_resources,
)
_LOGGER.debug("Fetching Samsung cloud UUID from %s", _SAMSUNG_CLOUD_HOST)
try:
@@ -266,7 +268,7 @@ def _probe_and_validate(host: str, ca_cert_pem: str, ca_key_pem: str) -> dict:
) or for_device_by_model(
info_resource.get('x.com.samsung.da.modelNum', ''),
info_resource.get('x.com.samsung.da.description', ''),
)
) or for_device_by_resources(resources)
return {
"port": port,
"serial": serial,
+4 -1
View File
@@ -1,6 +1,9 @@
DOMAIN = "localthings"
PLATFORMS = ["sensor", "binary_sensor", "switch", "number", "select", "button", "time", "climate"]
PLATFORMS = [
"sensor", "binary_sensor", "switch", "number", "select", "button",
"time", "climate", "fan",
]
CONF_HOST = "host"
CONF_PORT = "port"
+4 -2
View File
@@ -20,7 +20,7 @@ from smartthings_local.protocol.dtls_session import DtlsCoapSession
from smartthings_local.ocf.state_cache import StateCache
from .registry.batch import parse_device0_batch
from .registry.by_type import for_device, for_device_by_model
from .registry.by_type import for_device, for_device_by_model, for_device_by_resources
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -287,6 +287,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
info.get('x.com.samsung.da.modelNum', ''),
info.get('x.com.samsung.da.description', ''),
)
if reg is None:
reg = for_device_by_resources(resources)
unbound: list[str] = []
if reg is not None:
self._log.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui)
@@ -306,7 +308,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self.device_serial = serial
ident = self._identity
device_type = reg.name.title() if reg else 'Appliance'
device_type = reg.name.replace('_', ' ').title() if reg else 'Appliance'
model_num = info.get('x.com.samsung.da.modelNum', '')
model = model_num.split('|', 1)[0] if model_num else (ident.model if ident else '')
name = f"Samsung {device_type} ({model})" if model else f"Samsung {device_type}"
+110
View File
@@ -0,0 +1,110 @@
"""Fan platform for Samsung range hoods."""
from __future__ import annotations
from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from .const import DOMAIN
from .coordinator import LocalThingsCoordinator
from .entity import LocalThingsEntity, _is_included
from .registry.entities import FanDesc
POWER_HREF = '/power/0'
POWER_VS_HREF = '/power/vs/0'
_FAN_SPEED_FIELD = 'x.com.samsung.da.hood.fanSpeed'
_SUPPORTED_FAN_SPEED_FIELD = 'x.com.samsung.da.hood.supportedFanSpeed'
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
LocalThingsRangeHoodFan(coordinator, bound)
for bound in coordinator.bound
if isinstance(bound.desc, FanDesc) and _is_included(bound, coordinator)
)
class LocalThingsRangeHoodFan(LocalThingsEntity, FanEntity):
"""A hood fan combining sibling power and fan-speed resources."""
_enable_turn_on_off_backwards_compatibility = False
_attr_supported_features = (
FanEntityFeature.SET_SPEED
| FanEntityFeature.TURN_ON
| FanEntityFeature.TURN_OFF
)
def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None:
super().__init__(coordinator, bound)
self._attr_name = None
def _rep(self, href: str) -> dict:
return self.coordinator.resource(href) or {}
def _all_speed_codes(self) -> list[str]:
rep = self._rep(self._bound.href)
return [str(value) for value in rep.get(_SUPPORTED_FAN_SPEED_FIELD, ())]
def _active_speed_codes(self) -> list[str]:
# Power is carried by the separate /power resource. fanSpeed retains
# the selected setting while power is off (as the lamp's `current`
# field does), so every advertised code is an active ordered speed.
return self._all_speed_codes()
@property
def is_on(self) -> bool:
rep = self._rep(POWER_HREF)
if 'value' in rep:
return bool(rep.get('value'))
return str(
self._rep(POWER_VS_HREF).get('x.com.samsung.da.power', '')
).lower() == 'on'
@property
def speed_count(self) -> int:
return len(self._active_speed_codes())
@property
def percentage(self) -> int | None:
if not self.is_on:
return 0
codes = self._active_speed_codes()
current = str(self._rep(self._bound.href).get(_FAN_SPEED_FIELD, ''))
if not codes or current not in codes:
return None
return ordered_list_item_to_percentage(codes, current)
async def async_turn_on(
self, percentage: int | None = None, preset_mode: str | None = None,
**kwargs,
) -> None:
await self.coordinator.async_send_command(self._bound, ('power', True))
if percentage is not None:
await self.async_set_percentage(percentage)
async def async_turn_off(self, **kwargs) -> None:
await self.coordinator.async_send_command(self._bound, ('power', False))
async def async_set_percentage(self, percentage: int) -> None:
if percentage <= 0:
await self.async_turn_off()
return
codes = self._active_speed_codes()
if not codes:
return
if not self.is_on:
await self.coordinator.async_send_command(self._bound, ('power', True))
code = percentage_to_ordered_list_item(codes, percentage)
await self.coordinator.async_send_command(self._bound, ('speed', code))
@@ -2,17 +2,26 @@
from typing import Optional
from ._base import DeviceRegistry
from . import airconditioner, dishwasher, dryer, oven, refrigerator, washer
from . import (
airconditioner, cooktop, dishwasher, dryer, oven, range_hood, refrigerator,
washer,
)
__all__ = ['DeviceRegistry', '_type_key', 'for_device', 'for_device_by_model']
__all__ = [
'DeviceRegistry', '_type_key', 'for_device', 'for_device_by_model',
'for_device_by_resources',
]
_REGISTRY_BY_KEY: dict[str, DeviceRegistry] = {
'airconditioner': airconditioner.REGISTRY,
'air_conditioner': airconditioner.REGISTRY,
'cooktop': cooktop.REGISTRY,
'dishwasher': dishwasher.REGISTRY,
'dryer': dryer.REGISTRY,
'oven': oven.REGISTRY,
'hood': range_hood.REGISTRY,
'range_hood': range_hood.REGISTRY,
'refrigerator': refrigerator.REGISTRY,
'washer': washer.REGISTRY,
}
@@ -94,4 +103,38 @@ def for_device_by_model(model_num: str, description: str) -> Optional[DeviceRegi
# a modelNum carrying the '_PRAC_' (Package Room Air Conditioner) token.
if key is None and '_PRAC_' in (model_num or ''):
key = 'airconditioner'
model_identity = f'{model_num} {description}'.upper()
if key is None and ('_COOKTOP' in model_identity or '_GB_CT_' in model_identity):
key = 'cooktop'
if key is None and model_identity.startswith('AHD-'):
key = 'range_hood'
return _REGISTRY_BY_KEY.get(key) if key else None
def for_device_by_resources(resources: dict[str, dict]) -> Optional[DeviceRegistry]:
"""Detect a device family from a distinctive local-resource signature.
Some newer cooktops omit both ``oneUiVersion`` and
``/information/vs/0``. Their mode resource still identifies them: it
contains a DeviceType option and multiple per-burner OperationState
options. Require both shapes so an oven's unrelated ``/mode/vs/0`` is
not misclassified.
"""
mode = resources.get('/mode/vs/0', {})
options = mode.get('x.com.samsung.da.options') or ()
has_device_type = any(
isinstance(option, str) and option.startswith('DeviceType_')
for option in options
)
operation_states = sum(
1 for option in options
if isinstance(option, str) and option.startswith('OperationState')
)
if has_device_type and operation_states >= 2:
return _REGISTRY_BY_KEY['cooktop']
if (
'/hood/fanspeed/vs/0' in resources
and '/hood/lamp/vs/0' in resources
):
return _REGISTRY_BY_KEY['range_hood']
return None
@@ -0,0 +1,17 @@
"""Cooktop device registry."""
from ..capabilities import cooktop, fridge, ignored
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='cooktop',
capabilities=_build([
*ignored.IGNORED,
cooktop.COOKTOP_POWER,
cooktop.COOKTOP_MODE,
cooktop.COOKTOP_CONNECTED,
cooktop.PAIRED_HOOD_STATUS,
fridge.FIRMWARE_UPDATE,
]),
)
@@ -0,0 +1,22 @@
"""Range-hood device registry."""
from ..capabilities import common, fridge, ignored, range_hood
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='range_hood',
capabilities=_build([
*ignored.IGNORED,
range_hood.HOOD_ALARMS,
common.ENERGY_METER,
fridge.FIRMWARE_UPDATE,
range_hood.HOOD_FAN,
range_hood.HOOD_LAMP,
range_hood.HOOD_FILTER,
range_hood.AIR_QUALITY,
range_hood.AIR_LEVEL_CHECK,
range_hood.AUTO_VENTILATION,
*range_hood.COVERAGE,
]),
)
@@ -1,4 +1,7 @@
from . import common, dishwasher, fridge, ignored, laundry, operational, oven
from . import (
common, cooktop, dishwasher, fridge, ignored, laundry, operational, oven,
range_hood,
)
from ..capability import Capability
@@ -0,0 +1,192 @@
"""Read-only capabilities for Samsung cooktops.
The first verified device is an NA9300K-class five-burner gas cooktop. Its
local OCF API reports burner state as strings embedded in the
``x.com.samsung.da.options`` array on ``/mode/vs/0``. Heat-producing controls
are intentionally not exposed: the local write contract is unverified and a
cooktop must not be remotely ignited by an automation.
"""
import re
from ..capability import Capability
from ..entities import BinarySensorDesc, SensorDesc
_INACTIVE_OPERATION_STATES = {'Off', 'Ready'}
def _option_value(options, prefix):
"""Return the value from the first ``<prefix>_<value>`` option."""
marker = prefix + '_'
for option in options or ():
if isinstance(option, str) and option.startswith(marker):
return option[len(marker):]
return None
def _operation_slots(options) -> tuple[int, ...]:
"""Return the numeric burner slots advertised in an options array."""
slots = set()
for option in options or ():
if not isinstance(option, str):
continue
match = re.match(r'^OperationState(\d+)_', option)
if match:
slots.add(int(match.group(1)))
return tuple(sorted(slots))
def _any_burner_active(options):
"""True when any advertised burner slot is not idle."""
states = [
_option_value(options, f'OperationState{slot}')
for slot in _operation_slots(options)
]
states = [state for state in states if state is not None]
return any(state not in _INACTIVE_OPERATION_STATES for state in states)
def _int_or_none(value):
try:
return int(value)
except (TypeError, ValueError):
return None
COOKTOP_POWER = Capability(
href='/power/vs/0',
poll_tier='hot',
entities=(
SensorDesc(
key='power_state',
field='x.com.samsung.da.power',
name='Power state',
icon='mdi:stove',
),
),
)
# NA9300K exposes physical slots 0, 1, 3, 4, and 5. Preserve Samsung's slot
# numbers instead of guessing at front/back/left/right placement. A user can
# safely identify and rename each entity by briefly lighting one burner.
_NA9300K_OPERATION_SLOTS = (0, 1, 3, 4, 5)
COOKTOP_MODE = Capability(
href='/mode/vs/0',
poll_tier='hot',
entities=(
BinarySensorDesc(
key='any_burner_active',
field='x.com.samsung.da.options',
name='Any burner active',
device_class='running',
icon='mdi:fire',
value_fn=_any_burner_active,
),
*(
SensorDesc(
key=f'burner_{slot}_state',
field='x.com.samsung.da.options',
name=f'Burner {slot} state',
icon='mdi:gas-burner',
value_fn=lambda options, slot=slot: _option_value(
options, f'OperationState{slot}'
),
exists_fn=lambda rep, resources, slot=slot: (
not rep or _option_value(
rep.get('x.com.samsung.da.options'),
f'OperationState{slot}',
) is not None
),
)
for slot in _NA9300K_OPERATION_SLOTS
),
SensorDesc(
key='main_timer_state',
field='x.com.samsung.da.options',
name='Timer state',
icon='mdi:timer-outline',
value_fn=lambda options: _option_value(options, 'MainTimerState'),
),
SensorDesc(
key='main_timer_current',
field='x.com.samsung.da.options',
name='Timer current value',
icon='mdi:timer-sand',
enabled_default=False,
value_fn=lambda options: _int_or_none(
_option_value(options, 'MainTimerCurrent')
),
),
),
)
COOKTOP_CONNECTED = Capability(
href='/connected/vs/0',
poll_tier='warm',
entities=(
BinarySensorDesc(
key='cloud_connected',
field='x.com.samsung.da.connected',
name='Cloud connected',
device_class='connectivity',
entity_category='diagnostic',
value_fn=lambda value: str(value).lower() == 'on',
),
),
)
PAIRED_HOOD_STATUS = Capability(
href='/bluetooth/hood/status/vs/0',
poll_tier='hot',
entities=(
BinarySensorDesc(
key='paired_hood_connected',
field='connectionState',
name='Paired hood connected',
device_class='connectivity',
value_fn=lambda value: str(value).lower() == 'connected',
),
BinarySensorDesc(
key='paired_hood_power',
field='power',
name='Paired hood power',
device_class='running',
value_fn=lambda value: str(value).lower() == 'on',
),
SensorDesc(
key='paired_hood_fan_speed',
field='fanSpeed',
name='Paired hood fan speed',
icon='mdi:fan',
value_fn=_int_or_none,
),
BinarySensorDesc(
key='paired_hood_light',
field='lampState',
name='Paired hood light',
device_class='light',
value_fn=lambda value: str(value).lower() == 'on',
),
SensorDesc(
key='paired_hood_model',
field='micomModelId',
name='Paired hood model',
icon='mdi:information-outline',
entity_category='diagnostic',
enabled_default=False,
),
SensorDesc(
key='paired_hood_firmware',
field='firmwareVersion',
name='Paired hood firmware',
icon='mdi:chip',
entity_category='diagnostic',
enabled_default=False,
),
),
)
@@ -0,0 +1,295 @@
"""Capabilities for Samsung AHD-WW-TP1-22 range hoods.
The verified device exposes its fan, two-level work lamp, washable-filter
status, and particulate sensors as distinct local OCF resources. Fan power
and speed are combined into one HA fan entity by ``fan.py``; lamp power and
brightness remain separate controls because the device advertises them as two
independent fields.
"""
from datetime import datetime, timezone
from ..capability import Capability
from ..entities import (
BinarySensorDesc,
FanDesc,
SelectDesc,
SensorDesc,
SwitchDesc,
)
def _int_or_none(value):
try:
return int(value)
except (TypeError, ValueError):
return None
def _timestamp(value):
try:
return datetime.fromtimestamp(float(value), tz=timezone.utc)
except (TypeError, ValueError, OSError):
return None
def _item_value(items, sensor_type, index=0):
for item in items or ():
if not isinstance(item, dict):
continue
if item.get('x.com.samsung.da.type') != sensor_type:
continue
values = item.get('x.com.samsung.da.value') or ()
if index < len(values):
return _int_or_none(values[index])
return None
def _active_alarm_codes(items):
"""Discard the hood firmware's retained/deleted ``ErrorCode_OFF`` row."""
codes = []
for item in items or ():
if not isinstance(item, dict):
continue
if str(item.get('x.com.samsung.da.state', '')).lower() == 'deleted':
continue
code = item.get('x.com.samsung.da.code')
if code and str(code).lower() != 'errorcode_off':
codes.append(code)
return ', '.join(codes) if codes else 'none'
HOOD_ALARMS = Capability(
href='/alarms/vs/0',
poll_tier='hot',
entities=(
SensorDesc(
key='alarm_code',
field='x.com.samsung.da.items',
name='Alarm code',
icon='mdi:alert',
entity_category='diagnostic',
value_fn=_active_alarm_codes,
),
),
)
def _hood_fan_write(payload, rep, href=None):
kind, value = payload
if kind == 'power':
return ['power', '0'], {'value': bool(value)}
if kind == 'speed':
value = str(value)
supported = [
str(code)
for code in rep.get('x.com.samsung.da.hood.supportedFanSpeed', ())
]
if value not in supported:
return None
return ['hood', 'fanspeed', 'vs', '0'], {
'x.com.samsung.da.hood.fanSpeed': value,
}
return None
HOOD_FAN = Capability(
href='/hood/fanspeed/vs/0',
poll_tier='hot',
entities=(
FanDesc(
key='fan',
field='x.com.samsung.da.hood.fanSpeed',
name=None,
write_fn=_hood_fan_write,
),
BinarySensorDesc(
key='automatic_operation',
field='x.com.samsung.da.hood.autoOperation',
name='Automatic operation',
icon='mdi:fan-auto',
entity_category='diagnostic',
value_fn=lambda value: str(value).lower() == 'on',
),
),
)
def _lamp_level_write(value, rep, href=None):
code = str(value)
supported = [str(level) for level in rep.get('x.com.samsung.lamp.range', ())]
if code not in supported:
return None
return ['hood', 'lamp', 'vs', '0'], {
'x.com.samsung.lamp.current': code,
}
HOOD_LAMP = Capability(
href='/hood/lamp/vs/0',
poll_tier='hot',
entities=(
SwitchDesc(
key='lamp',
field='x.com.samsung.lamp.power',
name='Lamp',
icon='mdi:range-hood',
value_fn=lambda value: str(value).lower() == 'on',
write_fn=lambda payload, rep, href=None: (
['hood', 'lamp', 'vs', '0'],
{'x.com.samsung.lamp.power': 'On' if payload == 'On' else 'Off'},
),
),
SelectDesc(
key='lamp_brightness',
field='x.com.samsung.lamp.current',
name='Lamp brightness',
icon='mdi:brightness-6',
translation_key='range_hood_lamp_brightness',
options_field='x.com.samsung.lamp.range',
write_fn=_lamp_level_write,
),
),
)
HOOD_FILTER = Capability(
href='/filter/hoodfilter/vs/0',
poll_tier='cold',
entities=(
SensorDesc(
key='hood_filter_usage',
field='x.com.samsung.da.filterUsage',
name='Filter usage',
unit='%',
state_class='measurement',
icon='mdi:air-filter',
entity_category='diagnostic',
value_fn=_int_or_none,
),
SensorDesc(
key='hood_filter_status',
field='x.com.samsung.da.filterStatus',
name='Filter status',
icon='mdi:air-filter',
entity_category='diagnostic',
),
SensorDesc(
key='hood_filter_capacity',
field='x.com.samsung.da.filterCapacity',
name='Filter capacity',
unit='h',
icon='mdi:timer-outline',
entity_category='diagnostic',
enabled_default=False,
value_fn=_int_or_none,
),
),
)
AIR_QUALITY = Capability(
href='/sensors/vs/0',
poll_tier='warm',
entities=(
SensorDesc(
key='clean_level',
field='x.com.samsung.da.items',
name='Clean level',
icon='mdi:air-filter',
value_fn=lambda items: _item_value(items, 'CleanLevel'),
),
SensorDesc(
key='dust',
field='x.com.samsung.da.items',
name='Dust',
value_fn=lambda items: _item_value(items, 'Dust'),
),
SensorDesc(
key='fine_dust',
field='x.com.samsung.da.items',
name='Fine dust',
value_fn=lambda items: _item_value(items, 'FineDust'),
),
SensorDesc(
key='super_fine_dust',
field='x.com.samsung.da.items',
name='Super fine dust',
value_fn=lambda items: _item_value(items, 'SuperFineDust'),
),
),
)
AIR_LEVEL_CHECK = Capability(
href='/airlevelcheck/vs/0',
poll_tier='warm',
entities=(
BinarySensorDesc(
key='periodic_air_sensing',
field='x.com.samsung.da.periodicSensingActivationState',
name='Periodic air sensing',
icon='mdi:radar',
entity_category='diagnostic',
value_fn=lambda value: str(value).lower() == 'on',
),
SensorDesc(
key='air_sensing_state',
field='x.com.samsung.da.sensingState',
name='Air sensing state',
icon='mdi:radar',
entity_category='diagnostic',
),
SensorDesc(
key='last_air_sensing_time',
field='x.com.samsung.da.lastSensingTime',
name='Last air sensing time',
device_class='timestamp',
entity_category='diagnostic',
value_fn=_timestamp,
),
SensorDesc(
key='last_air_sensing_level',
field='x.com.samsung.da.lastSensingLevel',
name='Last air sensing level',
icon='mdi:air-filter',
entity_category='diagnostic',
),
SensorDesc(
key='automatic_ventilation_state',
field='x.com.samsung.da.autoExeState',
name='Automatic ventilation state',
icon='mdi:fan-auto',
entity_category='diagnostic',
),
),
)
AUTO_VENTILATION = Capability(
href='/autoventilation/vs/0',
poll_tier='warm',
entities=(
SensorDesc(
key='auto_ventilation_action',
field='action',
name='Auto ventilation action',
icon='mdi:fan-auto',
),
),
)
# Resource plumbing and opaque feature-negotiation fields that are specific to
# this family. Bare capabilities mark them covered without creating entities.
COVERAGE = [
Capability(href=href)
for href in (
'/power/0',
'/power/vs/0',
'/mode/vs/0',
'/personality/presence/vs/0',
'/availablecontrolsets/vs/0',
'/da/softreset/vs/0',
)
]
@@ -96,6 +96,13 @@ class ClimateDesc(SamsungEntityDescription):
write_fn: WriteFn = None
@dataclass(frozen=True, kw_only=True)
class FanDesc(SamsungEntityDescription):
# Composite fan entity: reads power from /power/0 and speed/support data
# from its bound href. Payloads are (kind, value), like ClimateDesc.
write_fn: WriteFn = None
PLATFORM_OF: dict[type, str] = {
SensorDesc: 'sensor',
BinarySensorDesc: 'binary_sensor',
@@ -105,4 +112,5 @@ PLATFORM_OF: dict[type, str] = {
NumberDesc: 'number',
TimeDesc: 'time',
ClimateDesc: 'climate',
FanDesc: 'fan',
}
@@ -16,6 +16,12 @@
"100": "High"
}
},
"range_hood_lamp_brightness": {
"state": {
"1": "Level 1",
"2": "Level 2"
}
},
"ice_type": {
"state": {
"off": "Off",
@@ -16,6 +16,12 @@
"100": "High"
}
},
"range_hood_lamp_brightness": {
"state": {
"1": "Level 1",
"2": "Level 2"
}
},
"ice_type": {
"state": {
"off": "Off",