Compare commits

...
Author SHA1 Message Date
Marc Billow 618443db7b Use generic names for shared start/stop/pause entities
The /operational/state/vs/0 href (and its start/pause/stop buttons and
"cycle active" sensor) is shared across the dryer/dishwasher/oven/washer
families, but the entity names hardcoded laundry vocabulary ("Start
cycle", "Cycle active") that doesn't fit an oven's bake/roast session.
Rename to generic "Start"/"Pause"/"Stop"/"Running".

Also fold oven.py's duplicate stop button into a single STOP_BUTTON
constant in operational.py -- both wrote the identical state='Ready'
RMW, so there was no reason for two copies to maintain.
2026-07-23 22:38:48 +00:00
Marc Billow 3ed277a33b Fix wall oven device-type detection (issue #55)
Ovens with modelNum like TP1X_DA-KS-OVEN-0107X report no oneUiVersion
and don't match any consumer-prefix or other fallback token in
for_device_by_model, so the device came back as "unknown" and every
resource fell through to the global capability registry instead of
oven.py's own -- explaining the unbound /connected/vs/0 href, the
unrelated-looking entities, and the non-functional controls reported
in the issue. Add a '-OVEN-' modelNum token fallback, mirroring the
existing '-RANGE-' fallback from issue #44.
2026-07-23 22:38:44 +00:00
Marc Billow 30e814f242 Merge pull request #23 from themaanda/contrib/cooktop-range-hood
Add cooktop and range hood support
2026-07-22 22:41:10 -05:00
Marc Billow 13f1214790 Merge branch 'main' into contrib/cooktop-range-hood 2026-07-22 22:39:19 -05:00
Marc Billow 4d21f3b8f2 Merge pull request #51 from mbillow/claude/release-notes-version-bump-1obhfb
Bump version to 0.10.1
2026-07-22 22:31:42 -05:00
Marc Billow 760d797c12 Bump version to 0.10.1 2026-07-23 03:30:16 +00:00
Marc Billow b944549f55 Merge pull request #50 from mbillow/claude/release-notes-version-bump-1obhfb
Add range/cooktop-oven combo to supported appliance types
2026-07-22 22:28:51 -05:00
Marc Billow 94551da4a0 fix: apply writes optimistically before the settle guard (issue #27)
mark_write_pending's settle window was dropping every update for a
just-written href, including the coordinator's own post-write refresh,
because nothing ever wrote the optimistic value into the cache for it
to protect. The write reflected on the device immediately but reverted
in HA until the next 30s summary sweep.
2026-07-23 03:26:00 +00:00
themaanda f7db568a43 Merge remote-tracking branch 'upstream/main' into contrib/cooktop-range-hood
# Conflicts:
#	custom_components/localthings/coordinator.py
#	custom_components/localthings/registry/by_type/__init__.py
2026-07-22 21:59:23 -05:00
themaanda 62caf769f2 Address cooktop and range hood review 2026-07-22 21:54:01 -05:00
themaanda e7daa21931 Merge remote-tracking branch 'upstream/main' into contrib/cooktop-range-hood
# Conflicts:
#	README.md
#	custom_components/localthings/registry/by_type/__init__.py
#	tests/test_by_type.py
#	tests/test_golden_regression.py
2026-07-22 21:53:45 -05:00
Marc Billow f783aa72b9 Add range/cooktop-oven combo to supported appliance types 2026-07-23 02:47:12 +00:00
Marc Billow ed461d82cd Merge pull request #49 from mbillow/claude/release-notes-version-bump-1obhfb
Bump version to 0.10.0
2026-07-22 21:45:15 -05:00
Marc Billow 62d4ed673b Bump version to 0.10.0 2026-07-23 02:44:22 +00:00
Marc Billow e74089a545 Merge pull request #48 from mbillow/claude/issue-44-device-fixes-ejmedf
Add range/cooktop-oven combo support (issue #44)
2026-07-22 21:35:10 -05:00
Marc Billow b2f3e501e5 Merge pull request #47 from mbillow/claude/remote-control-write-validation-axyqqs
feat: reject writes when a device's remote control is disabled
2026-07-22 21:32:43 -05:00
Marc Billow 2f6b7d54e3 refactor: share remote-control read between sensor and write guard, poll it warm
Move the on/off interpretation into a single remote_control_enabled()
in registry/capabilities/common.py so the write-guard added in the
previous commit can't silently drift from the Smart Control binary
sensor's own reading of the same hrefs. Also promotes both
/remotectrl hrefs to poll_tier='warm' so the coordinator's cached
state backing that write guard doesn't lag up to a full 30s cold
summary poll behind the device's actual toggle state.
2026-07-23 02:29:03 +00:00
Marc Billow 6e5fee32b8 feat: reject writes when a device's remote control is disabled
Devices with a /remotectrl href already surface it as a read-only
"Smart Control" binary sensor, but writes weren't checking it before
now. async_send_command now blocks every write (any platform) with a
ServiceValidationError telling the user to enable remote control via
the appliance's manual, ahead of any per-description validate_fn.
2026-07-23 02:15:40 +00:00
themaanda c64f58aaa5 Add cooktop and range hood support 2026-07-21 16:57:45 -05:00
33 changed files with 2094 additions and 36 deletions

No files matched your search

+7 -3
View File
@@ -19,6 +19,9 @@ Your state stays on your LAN: HA talks to the appliance over a direct DTLS sessi
| Air conditioner | `by_type/airconditioner.py` |
| Dryer | `by_type/dryer.py` |
| Oven | `by_type/oven.py` |
| Cooktop (read-only burner status) | `by_type/cooktop.py` |
| Range hood | `by_type/range_hood.py` |
| Range | `by_type/range.py` |
| Dishwasher | `by_type/dishwasher.py` |
| Refrigerator | `by_type/refrigerator.py` |
| Washer | `by_type/washer.py` |
@@ -100,7 +103,7 @@ custom_components/localthings/
diagnostics.py Redacted diagnostics download (device state + coverage metadata)
const.py Domain, config keys, probe ports
entity.py Base entity wiring capability registry -> HA entity
sensor.py / binary_sensor.py / switch.py / number.py / select.py / button.py / time.py / climate.py
sensor.py / binary_sensor.py / switch.py / number.py / select.py / button.py / time.py / fan.py / climate.py
One module per HA platform
strings.json / translations/ Config-flow copy + entity state translations
registry/
@@ -112,7 +115,8 @@ custom_components/localthings/
identity.py Reads device identity for type detection
redact.py Strips account/identity data before diagnostics leave HA
capabilities/ Shared + per-family Capability definitions (common, airconditioner,
dryer, oven, dishwasher, fridge, washer, laundry, operational, ignored)
cooktop, range_hood, dryer, oven, dishwasher, fridge, washer,
laundry, operational, ignored)
by_type/ One DeviceRegistry per appliance type, composed from capabilities/
tests/ Registry composition, discovery, entity descriptors, coordinator/observe
behavior, and golden-file regression against captured device dumps
@@ -138,7 +142,7 @@ support for hardware the maintainers don't have.
1. Get a capture of the appliance's `/device/0` response. The easiest way: add the device to HA (type detection failing is fine) and pull its Diagnostics download from Settings > Devices & Services > the device > the menu > Download diagnostics — it already contains a redacted dump of the device's resources.
2. Reuse existing `Capability` objects from `registry/capabilities/` wherever the resource matches one already declared. Most `common.py` capabilities (power, kids lock, remote control, alarms, energy/water meters) are shared verbatim across families; add new ones only for resources unique to the new type.
3. Create `registry/by_type/<name>.py` with a `DeviceRegistry(name=..., capabilities=_build([...]))`. Use `pattern_capabilities` instead of `capabilities` for any resource whose `href` isn't fixed (for example per-compartment fridge resources); see `refrigerator.py` for the pattern.
4. Register it in `_REGISTRY_BY_KEY` in `registry/by_type/__init__.py`, keyed on the lowercased, space/hyphen-to-underscore-converted suffix of the device's `oneUiVersion` string (see `_type_key()` in that file for the exact transform). If the device never reports `oneUiVersion` — confirmed true for washers — add its consumer-model prefix to `_CONSUMER_PREFIX_TO_KEY` instead, so `for_device_by_model()` can route it.
4. Register it in `_REGISTRY_BY_KEY` in `registry/by_type/__init__.py`, keyed on the lowercased, space/hyphen-to-underscore-converted suffix of the device's `oneUiVersion` string (see `_type_key()` in that file for the exact transform). If the device never reports `oneUiVersion`, add its consumer-model prefix to `_CONSUMER_PREFIX_TO_KEY` so `for_device_by_model()` can route it. If it also omits `/information/vs/0` (as the verified NA9300K cooktop does), add a distinctive, conservative resource-signature rule to `for_device_by_resources()`.
5. Add golden-file coverage in `tests/` against a captured `/device/0` dump for the new type.
No config-flow changes are needed. Device-type detection and entity wiring are fully driven by the registry.
+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"
+29 -5
View File
@@ -21,7 +21,8 @@ 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.capabilities.common import remote_control_enabled
from .registry.discovery import discover, BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
@@ -37,6 +38,12 @@ _LOGGER = logging.getLogger(__name__)
_SEED_PATH = ['device', '0']
_REMOTE_CONTROL_DISABLED_MESSAGE = (
"Remote control is turned off on this device. Check your appliance's "
"manual for how to enable remote control before Home Assistant can "
"control it."
)
class _NoOpDescriptor:
"""StateCache requires a descriptor with an on_observation hook. This
@@ -288,6 +295,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)
@@ -307,7 +316,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}"
@@ -515,16 +524,21 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
A description-level validate_fn (currently SwitchDesc only) runs
here rather than per-platform, so rejecting a write with a
user-facing message -- as opposed to write_fn's silent no-op below
-- is available to every platform for free."""
-- is available to every platform for free. The remote-control
check runs first and applies to every platform unconditionally,
ahead of any description-specific validate_fn."""
desc = bound_entity.desc
write_fn = getattr(desc, 'write_fn', None)
if write_fn is None:
return
href = bound_entity.href
rep = self._cache.get(href or '') or {}
resources = self._cache.snapshot()
if not remote_control_enabled(resources):
raise ServiceValidationError(_REMOTE_CONTROL_DISABLED_MESSAGE)
validate_fn = getattr(desc, 'validate_fn', None)
if validate_fn is not None:
error = validate_fn(payload, rep, self._cache.snapshot())
error = validate_fn(payload, rep, resources)
if error:
raise ServiceValidationError(error)
result = write_fn(payload, rep, href)
@@ -533,11 +547,21 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
return
path_segs, body = result
# Apply the write optimistically before starting the settle guard,
# not after -- mark_write_pending gates every source (poll, sweep,
# observe) through the same apply(), itself included, so flipping
# this order would have the guard drop the one update it exists to
# protect. Without an optimistic value in the cache for it to hold
# onto, the settle window was just delaying the real device
# confirmation for a few seconds on every write, which read exactly
# like the write being silently reverted (issue #27).
self._observe.apply(href, body, source='optimistic')
self._observe.mark_write_pending(href)
def _do_put():
sess = self._session
if sess is None:
raise RuntimeError("no session")
self._observe.mark_write_pending(href)
code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0)
self._log.info("PUT %s → code %#04x", href, code)
+122
View File
@@ -0,0 +1,122 @@
"""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()
def _power_payload(self, enabled: bool) -> tuple[str, bool, str]:
"""Target whichever power resource this hood actually exposes."""
resources = self.coordinator.last_resources
target = POWER_HREF if POWER_HREF in resources else POWER_VS_HREF
return 'power', enabled, target
@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, self._power_payload(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, self._power_payload(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, self._power_payload(True),
)
code = percentage_to_ordered_list_item(codes, percentage)
await self.coordinator.async_send_command(self._bound, ('speed', code))
+1 -1
View File
@@ -12,5 +12,5 @@
"pyOpenSSL>=23.0",
"smartthings-local>=0.1.0"
],
"version": "0.9.0"
"version": "0.10.1"
}
@@ -2,18 +2,27 @@
from typing import Optional
from ._base import DeviceRegistry
from . import airconditioner, dishwasher, dryer, oven, range as _range, refrigerator, washer
from . import (
airconditioner, cooktop, dishwasher, dryer, oven, range as _range,
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': _range.REGISTRY,
'range_hood': range_hood.REGISTRY,
'refrigerator': refrigerator.REGISTRY,
'washer': washer.REGISTRY,
}
@@ -102,9 +111,48 @@ def for_device_by_model(model_num: str, description: str) -> Optional[DeviceRegi
# 'P' sits between the underscore and 'RAC' in that token).
if key is None and '_RAC_' 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'
# Range/cooktop-oven combos (e.g. TP1X_DA-KS-RANGE-0102X, issue #44) --
# like the RAC/PRAC air conditioners above, these report no oneUiVersion
# and don't match the washer/dryer/dishwasher consumer-prefix map either.
if key is None and '-RANGE-' in (model_num or '').upper():
key = 'range'
# Wall ovens (e.g. TP1X_DA-KS-OVEN-0107X, issue #55) -- same board-family
# naming as the range combo above, minus the burners; also reports no
# oneUiVersion and doesn't match the washer/dryer/dishwasher prefix map.
if key is None and '-OVEN-' in (model_num or '').upper():
key = 'oven'
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 common, cooktop, 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,
common.FIRMWARE_UPDATE,
]),
)
@@ -0,0 +1,22 @@
"""Range-hood device registry."""
from ..capabilities import common, ignored, range_hood
from ._base import DeviceRegistry, _build
REGISTRY = DeviceRegistry(
name='range_hood',
capabilities=_build([
*ignored.IGNORED,
range_hood.HOOD_ALARMS,
common.ENERGY_METER,
common.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
@@ -119,8 +119,30 @@ KIDS_LOCK_VS_FALLBACK = Capability(
),
)
def remote_control_enabled(resources: dict) -> bool:
"""Single source of truth for the /remotectrl on/off signal, mirroring
REMOTE_CONTROL_GENERIC/_VS_FALLBACK's href/field pair and precedence
below. Used both to render the read-only Smart Control binary_sensor
(via those two descriptors) and, from coordinator.async_send_command,
to block writes outright when remote control is off. Both hrefs are
poll_tier='warm' below so that gate reads recent state (subscribed
when observe is live, subpolled every ~6s otherwise) rather than a
once-per-30s cold summary poll. True (assume enabled) when neither
href is present -- most device types don't report this capability
at all."""
generic = resources.get('/remotectrl/0')
if generic is not None:
return bool(generic.get('value'))
fallback = resources.get('/remotectrl/vs/0')
if fallback is not None:
return str(fallback.get('x.com.samsung.da.remoteControlEnabled')).lower() == 'true'
return True
REMOTE_CONTROL_GENERIC = Capability(
href='/remotectrl/0',
poll_tier='warm',
entities=(
BinarySensorDesc(key='remote_control', field='value',
name='Smart Control', device_class='connectivity',
@@ -131,6 +153,7 @@ REMOTE_CONTROL_GENERIC = Capability(
REMOTE_CONTROL_VS_FALLBACK = Capability(
href='/remotectrl/vs/0',
match_fn=lambda rep, resources: '/remotectrl/0' not in resources,
poll_tier='warm',
entities=(
BinarySensorDesc(key='remote_control',
field='x.com.samsung.da.remoteControlEnabled',
@@ -0,0 +1,195 @@
"""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=(
BinarySensorDesc(
key='power_state',
field='x.com.samsung.da.power',
name='Power state',
device_class='power',
icon='mdi:stove',
value_fn=lambda value: str(value).lower() == 'on',
),
),
)
# The verified NA9300K exposes physical slots 0, 1, 3, 4, and 5. Declare a
# generous static superset so variants with other layouts are not silently
# omitted; exists_fn hides every slot the live options array does not report.
# Replace this bound with data-driven entity generation when #31 lands.
_SUPPORTED_OPERATION_SLOTS = tuple(range(8))
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 _SUPPORTED_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,
),
),
)
@@ -70,6 +70,15 @@ def _finish_time(remaining_str):
return None
# Shared by dryer/dishwasher/oven/washer -- oven.py imports this directly
# rather than keeping its own copy, since both wrote the identical
# state='Ready' RMW.
STOP_BUTTON = ButtonDesc(key='stop', field='', name='Stop', payload='Ready',
icon='mdi:stop',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p}))
OPERATIONAL_STATE = Capability(
href='/operational/state/vs/0',
poll_tier='hot',
@@ -83,8 +92,12 @@ OPERATIONAL_STATE = Capability(
# Harmless for non-oven appliances — just an extra bool in state.
# Samsung firmware keeps state='Run' after progress reaches 'Finish',
# so we also gate on progress to avoid a stuck 'Running' indication.
# Named 'Running' rather than 'Cycle active' -- this href (and the
# start/pause/stop buttons below) is shared across the dryer/
# dishwasher/oven/washer families, and 'cycle' is laundry-specific
# vocabulary that doesn't fit an oven's bake/roast/etc.
BinarySensorDesc(key='cycle_active', device_class='running',
name='Cycle active',
name='Running',
rep_fn=lambda rep: (
_SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
and rep.get('x.com.samsung.da.progress') != 'Finish'
@@ -119,20 +132,16 @@ OPERATIONAL_STATE = Capability(
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{_delay_field(rep): _format_delay(p)})),
ButtonDesc(key='start', field='', name='Start cycle', payload='Run',
ButtonDesc(key='start', field='', name='Start', payload='Run',
icon='mdi:play',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p})),
ButtonDesc(key='pause', field='', name='Pause cycle', payload='Pause',
ButtonDesc(key='pause', field='', name='Pause', payload='Pause',
icon='mdi:pause',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p})),
ButtonDesc(key='stop', field='', name='Stop cycle', payload='Ready',
icon='mdi:stop',
write_fn=lambda p, rep, href=None: (
['operational', 'state', 'vs', '0'],
{'x.com.samsung.da.state': p})),
STOP_BUTTON,
),
)
@@ -26,10 +26,10 @@ from datetime import datetime, timezone, timedelta
from ..capability import Capability
from ..entities import (
BinarySensorDesc, ButtonDesc, NumberDesc, SelectDesc, SensorDesc,
SwitchDesc,
BinarySensorDesc, NumberDesc, SelectDesc, SensorDesc, SwitchDesc,
)
from .common import normalize_temp_unit
from .operational import STOP_BUTTON
# ---------------------------------------------------------------------------
# Constants
@@ -166,12 +166,6 @@ def _cook_time_write(p, rep, href=None):
}
def _stop_write(p, rep, href=None):
return ['operational', 'state', 'vs', '0'], {
'x.com.samsung.da.state': 'Ready',
}
def _oven_mode_write(p, rep, href=None):
if p not in _OVEN_MODES:
return None
@@ -237,7 +231,7 @@ OVEN_OPERATIONAL_STATE = Capability(
device_class='enum', options=('idle', 'active', 'pause'),
translation_key='machine_state', value_fn=_to_ocf),
BinarySensorDesc(key='cycle_active', field='x.com.samsung.da.state',
name='Cycle active', device_class='running',
name='Running', device_class='running',
value_fn=lambda v: _SAMSUNG_STATE_TO_OCF.get(v) == 'active'),
SensorDesc(key='progress_percentage',
field='x.com.samsung.da.progressPercentage',
@@ -254,8 +248,7 @@ OVEN_OPERATIONAL_STATE = Capability(
name='Cook time', unit='min', native_min=0, native_max=1439,
step=1.0, icon='mdi:timer', value_fn=_op_minutes,
write_fn=_cook_time_write),
ButtonDesc(key='stop', field='', name='Stop cycle', icon='mdi:stop',
payload='Stop', write_fn=_stop_write),
STOP_BUTTON,
),
)
@@ -0,0 +1,306 @@
"""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.
Unlike ``common._alarm_codes``, the hood retains a deleted alarm row in
its live representation, so this family-specific helper also checks state.
"""
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, *args = payload
if kind == 'power':
power_href = args[0] if args else '/power/0'
if power_href == '/power/0':
return ['power', '0'], {'value': bool(value)}
if power_href == '/power/vs/0':
return ['power', 'vs', '0'], {
'x.com.samsung.da.power': 'On' if value else 'Off',
}
return None
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',
)
]
@@ -111,6 +111,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',
@@ -120,4 +127,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",
+73
View File
@@ -0,0 +1,73 @@
{
"device0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/otninformation/vs/0",
"rep": {
"x.com.samsung.da.target": "",
"x.com.samsung.da.newVersionAvailable": "false"
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "On"
}
},
{
"href": "/mode/vs/0",
"rep": {
"rt": [
"x.com.samsung.da.mode"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"x.com.samsung.da.options": [
"DeviceType_NA9300K-/AA1",
"MainTimerCurrent_0",
"MainTimerSet_0",
"MainTimerState_Ready",
"OperationState0_Ready",
"OperationState1_Ready",
"OperationState3_Ready",
"OperationState4_Ready",
"OperationState5_Ready"
]
}
},
{
"href": "/connected/vs/0",
"rep": {
"x.com.samsung.da.connected": "On"
}
},
{
"href": "/configuration/vs/0",
"rep": {}
},
{
"href": "/bluetooth/hood/status/vs/0",
"rep": {
"connectionState": "disconnected",
"micomModelId": "",
"firmwareVersion": "",
"power": "off",
"fanSpeed": 0,
"lampState": "off",
"timer": {}
}
}
]
}
+21
View File
@@ -0,0 +1,21 @@
{
"state_keys": [
"any_burner_active",
"burner_0_state",
"burner_1_state",
"burner_3_state",
"burner_4_state",
"burner_5_state",
"cloud_connected",
"firmware_update",
"main_timer_current",
"main_timer_state",
"paired_hood_connected",
"paired_hood_fan_speed",
"paired_hood_firmware",
"paired_hood_light",
"paired_hood_model",
"paired_hood_power",
"power_state"
]
}
+25
View File
@@ -0,0 +1,25 @@
{
"state_keys": [
"alarm_code",
"child_lock",
"cloud_connected",
"cook_time",
"current_temp_c",
"cycle_active",
"door_open",
"fast_preheat",
"finish_time",
"firmware_update",
"lamp",
"machine_state",
"natural_steam",
"operation_time_minutes",
"oven_mode",
"oven_setpoint",
"oven_state",
"power_switch",
"progress_percentage",
"remote_control",
"sound"
]
}
+25
View File
@@ -0,0 +1,25 @@
{
"state_keys": [
"air_sensing_state",
"alarm_code",
"auto_ventilation_action",
"automatic_operation",
"automatic_ventilation_state",
"clean_level",
"energy_kwh",
"energy_saved_kwh",
"fan",
"firmware_update",
"hood_filter_capacity",
"hood_filter_status",
"hood_filter_usage",
"lamp",
"lamp_brightness",
"last_air_sensing_level",
"last_air_sensing_time",
"periodic_air_sensing",
"dust",
"fine_dust",
"super_fine_dust"
]
}
+289
View File
@@ -0,0 +1,289 @@
{
"meta": {
"model": "TP1X_DA-KS-OVEN-0107X (NV7000BS/ET5)",
"device_type": "oven",
"source": "issue #55 diagnostics (scrubbed)",
"note": "Wall oven reports no oneUiVersion; resolved via the '-OVEN-' modelNum token fallback in for_device_by_model, mirroring the '-RANGE-' fallback added for issue #44."
},
"device0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/otninformation/vs/0",
"rep": {
"x.com.samsung.da.target": "",
"x.com.samsung.da.newVersionAvailable": "false",
"flashingProgress": "",
"otnStatus": "None",
"otnList": [
{
"type": "WIFI",
"modelId": "AKS-WW-TP1-22-OVEN-1",
"versions": [
"40250221"
],
"visVersion": "250221"
},
{
"type": "Micom",
"modelId": "07134046004140472641",
"versions": [
"23011700",
"23122000"
],
"visVersion": "231220"
},
{
"type": "Micom",
"modelId": "071360127441FFFFFFFF",
"versions": [
"21123000",
"FFFFFFFF"
],
"visVersion": "211230"
}
]
}
},
{
"href": "/connected/vs/0",
"rep": {
"x.com.samsung.da.connected": "On"
}
},
{
"href": "/doors/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Door",
"x.com.samsung.da.openState": "Close"
}
],
"rt": [
"x.com.samsung.da.doors"
],
"if": [
"oic.if.baseline",
"oic.if.s"
]
}
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "TP1X_DA-KS-OVEN-0107X|40460041|50030018001611020A00000000000000",
"x.com.samsung.da.description": "NV7000BS/ET5",
"x.com.samsung.da.serialNum": "**REDACTED**",
"x.com.samsung.da.otnDUID": "**REDACTED**",
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Software",
"x.com.samsung.da.number": "25022100",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "1",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "04600A23011700, 04726A23122000",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "01274A21123000",
"x.com.samsung.da.newVersionAvailable": "0"
}
],
"x.com.samsung.da.diagProtocolType": "BLE_OCF",
"x.com.samsung.da.diagLogType": [
"errCode",
"dump"
],
"x.com.samsung.da.diagDumpType": "file",
"x.com.samsung.da.diagEndPoint": "SSM",
"x.com.samsung.da.diagMnid": "0AJT",
"x.com.samsung.da.diagSetupid": "KV1",
"x.com.samsung.da.diagMinVersion": "3.0",
"x.com.samsung.da.diagTsId": "DA01"
}
},
{
"href": "/mode/vs/0",
"rep": {
"x.com.samsung.da.supportedModes": [
"Autocook",
"Convection",
"TopHeatPluseConvection",
"Conventional",
"LargeGrill",
"SmallGrill",
"BottomHeatPluseConvection",
"PlateWarm",
"KeepWarm",
"Bottom",
"EcoConvection",
"FanGrill",
"Defrost",
"SteamClean"
],
"x.com.samsung.da.modes": [
"NoOperation"
],
"x.com.samsung.da.options": [
"DeviceType_NV7000BS/ET5",
"UpperTimerCurrent_0",
"UpperTimerSet_0",
"UpperTimerState_Ready",
"UpperLamp_Off",
"Sound_On"
],
"x.com.samsung.da.defaultMode": "Convection",
"rt": [
"x.com.samsung.da.mode"
],
"if": [
"oic.if.baseline",
"oic.if.a"
]
}
},
{
"href": "/operational/state/vs/0",
"rep": {
"x.com.samsung.da.state": "Ready",
"x.com.samsung.da.operationTime": "00:00:00",
"x.com.samsung.da.remainingTime": "00:00:00",
"x.com.samsung.da.progressPercentage": "0",
"rt": [
"x.com.samsung.da.operation"
],
"if": [
"oic.if.baseline",
"oic.if.a"
]
}
},
{
"href": "/alarms/vs/0",
"rep": {
"rt": [
"x.com.samsung.da.alarms"
],
"if": [
"oic.if.baseline",
"oic.if.s"
],
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "OV_E_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-23T15:26:07"
}
]
}
},
{
"href": "/oven/vs/0",
"rep": {
"x.com.samsung.da.state": "Ready",
"x.com.samsung.da.recipe": "00000000000000",
"rt": [
"x.com.samsung.da.oven"
],
"if": [
"oic.if.baseline",
"oic.if.s"
]
}
},
{
"href": "/temperatures/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Temperature",
"x.com.samsung.da.desired": "0",
"x.com.samsung.da.current": "0",
"x.com.samsung.da.increment": "5",
"x.com.samsung.da.unit": "Celsius"
}
],
"rt": [
"x.com.samsung.da.temperatures"
],
"if": [
"oic.if.baseline",
"oic.if.a"
]
}
},
{
"href": "/timezone/vs/0",
"rep": {
"timezoneid": "Europe/Rome",
"offset": "+02:00",
"DST": "ON"
}
},
{
"href": "/configuration/vs/0",
"rep": {}
},
{
"href": "/kidslock/vs/0",
"rep": {
"x.com.samsung.da.kidsLock": "Ready"
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "On"
}
},
{
"href": "/remotectrl/vs/0",
"rep": {
"x.com.samsung.da.remoteControlEnabled": "true",
"rt": [
"x.com.samsung.da.configuration"
],
"if": [
"oic.if.baseline",
"oic.if.s"
]
}
},
{
"href": "/wirelessinfo/vs/0",
"rep": {
"macaddressWiFi": "**REDACTED**",
"macaddressBLE": "**REDACTED**"
}
},
{
"href": "/quickcontrol/info/vs/0",
"rep": {
"supportedVersion": "1.0"
}
}
]
}
+199
View File
@@ -0,0 +1,199 @@
{
"device0": [
{
"rt": ["x.com.samsung.devcol", "oic.wk.col"],
"if": ["oic.if.baseline", "oic.if.ll", "oic.if.b"]
},
{
"href": "/personality/presence/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "",
"x.com.samsung.da.deviceId": "TEST-DEVICE-ID",
"x.com.samsung.da.value": ""
}
]
}
},
{
"href": "/realtimenotiforclient/vs/0",
"rep": {
"x.com.samsung.da.timeforshortnoti": "0",
"x.com.samsung.da.longnotisubscription": "false",
"x.com.samsung.da.periodicnotisubscription": "true"
}
},
{
"href": "/airlevelcheck/vs/0",
"rep": {
"x.com.samsung.da.periodicSensingActivationState": "Off",
"x.com.samsung.da.periodicSensingInterval": "3600",
"x.com.samsung.da.startSensingOnce": "Off",
"x.com.samsung.da.sensingState": "NonProcessing",
"x.com.samsung.da.lastSensingTime": "1784651378",
"x.com.samsung.da.lastSensingLevel": "Kr2",
"x.com.samsung.da.autoExeState": "Sensing",
"x.com.samsung.da.supportedAutoExeState": ["Airpurify", "Alarm"],
"x.com.samsung.da.periodicSensingSkipStatus": "Off",
"x.com.samsung.da.periodicSensingSkipTime": "00000000"
}
},
{
"href": "/alarms/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "ErrorCode_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-21T20:18:51",
"x.com.samsung.da.state": "Deleted"
}
]
}
},
{
"href": "/energy/consumption/vs/0",
"rep": {
"x.com.samsung.da.cumulativePower": "165620",
"x.com.samsung.da.cumulativeSavedPower": "0",
"x.com.samsung.da.cumulativeDate": "1784647130",
"x.com.samsung.da.cumulativeDateUTC": "1784665130",
"x.com.samsung.da.cumulativeUnit": "Wh"
}
},
{"href": "/energy/consumption/0", "rep": {}},
{
"href": "/mode/vs/0",
"rep": {
"x.com.samsung.da.options": ["OptionCode_19322", "UpdateAllow_NotAllowed"]
}
},
{
"href": "/power/vs/0",
"rep": {"x.com.samsung.da.power": "Off"}
},
{"href": "/power/0", "rep": {"value": false}},
{
"href": "/sensors/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Sensor for CleanLevel",
"x.com.samsung.da.type": "CleanLevel",
"x.com.samsung.da.value": ["2"]
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Sensor for Dust",
"x.com.samsung.da.type": "Dust",
"x.com.samsung.da.value": ["18", "1"]
},
{
"x.com.samsung.da.id": "3",
"x.com.samsung.da.description": "Sensor for FineDust",
"x.com.samsung.da.type": "FineDust",
"x.com.samsung.da.value": ["14", "1"]
},
{
"x.com.samsung.da.id": "4",
"x.com.samsung.da.description": "Sensor for SuperFineDust",
"x.com.samsung.da.type": "SuperFineDust",
"x.com.samsung.da.value": ["9", "1"]
}
]
}
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "AHD-WW-TP1-22-COMMON|20136141|7800006B001713C44D00090001030000",
"x.com.samsung.da.description": "AHD-WW-TP1-22-COMMON",
"x.com.samsung.da.serialNum": "TEST-HOOD-SERIAL",
"x.com.samsung.da.diagProtocolType": "WIFI_HTTPS",
"x.com.samsung.da.diagSetupid": "KH0"
}
},
{
"href": "/file/information/vs/0",
"rep": {"x.com.samsung.timeoffset": "-05:00"}
},
{
"href": "/configuration/vs/0",
"rep": {
"x.com.samsung.da.region": "0000000000",
"x.com.samsung.da.airconOptionList": ["DR"],
"x.com.samsung.da.countryCode": ""
}
},
{
"href": "/drlc/0",
"rep": {
"DRLevel": 0,
"start": "1970-01-01T00:00:00Z",
"durationminutes": 0,
"override": false
}
},
{
"href": "/drlc/vs/0",
"rep": {
"x.com.samsung.da.drlcLevel": "0",
"x.com.samsung.da.durationminutes": "0",
"x.com.samsung.da.start": "1970-01-01T00:00:00Z",
"x.com.samsung.da.override": "Off",
"x.com.samsung.da.realSaving": "Off"
}
},
{
"href": "/availablecontrolsets/vs/0",
"rep": {
"x.com.samsung.da.sets": "0000000000012000000000000000",
"x.com.samsung.da.id": "HOOD",
"x.com.samsung.da.version": "1.0"
}
},
{
"href": "/da/softreset/vs/0",
"rep": {"x.com.samsung.da.softwarereset": "false"}
},
{
"href": "/filter/hoodfilter/vs/0",
"rep": {
"x.com.samsung.da.filterUsage": "100",
"x.com.samsung.da.filterUsageResolution": "1",
"x.com.samsung.da.filterStatus": "wash",
"x.com.samsung.da.filterCapacity": "60",
"x.com.samsung.da.filterCapacityUnit": "Hour",
"x.com.samsung.da.filterResetType": ["replaceable", "washable"]
}
},
{
"href": "/hood/fanspeed/vs/0",
"rep": {
"x.com.samsung.da.hood.autoOperation": "Off",
"x.com.samsung.da.hood.fanSpeed": "14",
"x.com.samsung.da.hood.supportedFanSpeed": ["14", "15", "16", "17"]
}
},
{
"href": "/hood/lamp/vs/0",
"rep": {
"x.com.samsung.lamp.power": "Off",
"x.com.samsung.lamp.current": "2",
"x.com.samsung.lamp.range": ["1", "2"]
}
},
{"href": "/autoventilation/vs/0", "rep": {"action": "Start"}},
{
"href": "/otninformation/vs/0",
"rep": {
"x.com.samsung.da.target": "",
"x.com.samsung.da.newVersionAvailable": "false"
}
}
]
}
+201
View File
@@ -7,12 +7,16 @@ from unittest.mock import AsyncMock, patch
import cbor2
import pytest
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import issue_registry as ir
from custom_components.localthings.const import (
CONF_HOST, DOMAIN, SUMMARY_INTERVAL_S,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capabilities.common import (
remote_control_enabled,
)
from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL, PUSH_HEALTH_WINDOW_S
from .conftest import ENTRY_DATA, MOCK_SERIAL
@@ -132,6 +136,31 @@ def test_run_discovery_detects_washer_via_model_fallback(
assert coordinator.device_type_name == 'washer'
def test_run_discovery_detects_cooktop_via_resource_signature(
hass: HomeAssistant, mock_entry
) -> None:
"""NA9300K omits oneUiVersion and information but has burner options."""
from tests.conftest import _load_device
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._run_discovery(_load_device('cooktop'))
assert coordinator.device_type_name == 'cooktop'
assert coordinator._unbound_hrefs == []
def test_run_discovery_detects_range_hood_via_model(
hass: HomeAssistant, mock_entry
) -> None:
from tests.conftest import _load_device
coordinator = LocalThingsCoordinator(hass, mock_entry)
coordinator._run_discovery(_load_device('range_hood'))
assert coordinator.device_type_name == 'range_hood'
assert coordinator._unbound_hrefs == []
def test_update_coverage_gap_issue_creates_issue_for_unknown_type(
hass: HomeAssistant, mock_entry
) -> None:
@@ -634,3 +663,175 @@ async def test_write_marks_href_pending_before_post(
await coordinator.async_send_command(bound, 5)
assert coordinator._observe._settle_until.get('/test/vs/0') is not None
async def test_send_command_applies_write_optimistically_before_settling(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""The cache must reflect a write immediately, and stay put against a
stale echo for the rest of the settle window -- otherwise
mark_write_pending has nothing to protect and just delays the real
device confirmation instead, which reads to a user as the write being
silently reverted (issue #27)."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import NumberDesc
fake = mock_coordinator_observe_session
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
def _write_fn(payload, rep, href):
return (['some', 'path'], {'value': payload})
desc = NumberDesc(key='test', field='value', write_fn=_write_fn)
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
with patch.object(fake, 'subscribe'):
fake.post = lambda *a, **k: (0x44, b'')
await coordinator.async_send_command(bound, 5)
# The optimistic value is visible right away, without waiting for a
# device response.
assert coordinator._cache.get('/test/vs/0') == {'value': 5}
# A stale update racing in behind the write (e.g. a poll/notify that
# was already in flight before the PUT) must not clobber it while the
# settle window is open.
applied = coordinator._observe.apply('/test/vs/0', {'value': 0}, source='observe')
assert applied is False
assert coordinator._cache.get('/test/vs/0') == {'value': 5}
class TestRemoteControlEnabled:
"""remote_control_enabled (registry/capabilities/common.py) is the
single source of truth for the /remotectrl on/off signal, shared by
the Smart Control binary_sensor descriptors and the coordinator's
write guard alike."""
def test_vs_fallback_href_true_is_enabled(self):
resources = {'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'true'}}
assert remote_control_enabled(resources) is True
def test_vs_fallback_href_false_is_disabled(self):
resources = {'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'false'}}
assert remote_control_enabled(resources) is False
def test_generic_href_true_is_enabled(self):
resources = {'/remotectrl/0': {'value': True}}
assert remote_control_enabled(resources) is True
def test_generic_href_false_is_disabled(self):
resources = {'/remotectrl/0': {'value': False}}
assert remote_control_enabled(resources) is False
def test_generic_href_wins_when_both_present(self):
resources = {
'/remotectrl/0': {'value': True},
'/remotectrl/vs/0': {'x.com.samsung.da.remoteControlEnabled': 'false'},
}
assert remote_control_enabled(resources) is True
def test_assumes_enabled_when_capability_absent(self):
assert remote_control_enabled({}) is True
async def test_send_command_blocked_when_remote_control_disabled(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""A device reporting remote control off must reject every write with
a user-facing message, ahead of write_fn's silent-no-op path and any
description-level validate_fn."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import NumberDesc
fake = mock_coordinator_observe_session
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
coordinator._cache.apply_rep(
'/remotectrl/vs/0',
{'x.com.samsung.da.remoteControlEnabled': 'false'},
source='test',
)
def _write_fn(payload, rep, href):
return (['some', 'path'], {'value': payload})
desc = NumberDesc(key='test', field='value', write_fn=_write_fn)
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
posted = False
def _post(*a, **k):
nonlocal posted
posted = True
return (0x44, b'')
with patch.object(fake, 'subscribe'):
fake.post = _post
with pytest.raises(ServiceValidationError, match="Remote control is turned off"):
await coordinator.async_send_command(bound, 5)
assert posted is False
async def test_send_command_allowed_when_remote_control_enabled(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""The same write goes through once remote control reports enabled."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import NumberDesc
fake = mock_coordinator_observe_session
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
coordinator._cache.apply_rep(
'/remotectrl/vs/0',
{'x.com.samsung.da.remoteControlEnabled': 'true'},
source='test',
)
def _write_fn(payload, rep, href):
return (['some', 'path'], {'value': payload})
desc = NumberDesc(key='test', field='value', write_fn=_write_fn)
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
with patch.object(fake, 'subscribe'):
fake.post = lambda *a, **k: (0x44, b'')
await coordinator.async_send_command(bound, 5)
assert coordinator._observe._settle_until.get('/test/vs/0') is not None
async def test_send_command_remote_control_check_precedes_validate_fn(
hass: HomeAssistant, mock_entry, mock_coordinator_observe_session
) -> None:
"""When both a disabled remote-control state and a failing validate_fn
apply to the same write, the remote-control message wins -- it's the
more actionable of the two and is checked first."""
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import SwitchDesc
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
coordinator._cache.apply_rep(
'/remotectrl/vs/0',
{'x.com.samsung.da.remoteControlEnabled': 'false'},
source='test',
)
def _write_fn(payload, rep, href=None):
return (['some', 'path'], {'value': payload})
def _validate_fn(payload, rep, resources):
return "always rejected"
desc = SwitchDesc(key='test', field='value', write_fn=_write_fn, validate_fn=_validate_fn)
bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc)
with pytest.raises(ServiceValidationError, match="Remote control is turned off"):
await coordinator.async_send_command(bound, 'On')
+76
View File
@@ -41,6 +41,16 @@ class TestForDevice:
assert isinstance(registry, DeviceRegistry)
assert registry.name == 'refrigerator'
def test_for_device_returns_cooktop_registry(self):
registry = for_device('7.0 Cooktop')
assert registry is not None
assert registry.name == 'cooktop'
def test_for_device_returns_range_hood_registry(self):
registry = for_device('7.0 Range Hood')
assert registry is not None
assert registry.name == 'range_hood'
class TestDeviceRegistries:
"""Tests for device registries themselves."""
@@ -180,6 +190,25 @@ class TestForDeviceByModel:
assert reg is not None
assert reg.name == 'airconditioner'
def test_cooktop_via_legacy_model_description(self):
"""Older cooktops identify themselves as ARTIK051_GLOBAL_COOKTOP."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'ARTIK051_GB_CT_001|40424141|50000204001211000200000000000000',
'ARTIK051_GLOBAL_COOKTOP',
)
assert reg is not None
assert reg.name == 'cooktop'
def test_range_hood_via_ahd_model(self):
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'AHD-WW-TP1-22-COMMON|20136141|7800006B001713C44D00090001030000',
'AHD-WW-TP1-22-COMMON',
)
assert reg is not None
assert reg.name == 'range_hood'
def test_washer_wf_prefix(self):
"""US front-load washers use the WF consumer-model prefix."""
from custom_components.localthings.registry.by_type import for_device_by_model
@@ -200,6 +229,18 @@ class TestForDeviceByModel:
assert reg is not None
assert reg.name == 'washer'
def test_oven_via_oven_token(self):
"""Issue #55: a wall oven (NV7000BS/ET5) reports no oneUiVersion and
an unrecognized consumer token ('NV'); it falls back to the '-OVEN-'
token in modelNum, mirroring the '-RANGE-' fallback for issue #44."""
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model(
'TP1X_DA-KS-OVEN-0107X|40460041|50030018001611020A00000000000000',
'NV7000BS/ET5',
)
assert reg is not None
assert reg.name == 'oven'
def test_unknown_model_returns_none(self):
from custom_components.localthings.registry.by_type import for_device_by_model
reg = for_device_by_model('SOME-UNKNOWN-BOARD', 'SOME-UNKNOWN-BOARD')
@@ -208,3 +249,38 @@ class TestForDeviceByModel:
def test_empty_inputs_return_none(self):
from custom_components.localthings.registry.by_type import for_device_by_model
assert for_device_by_model('', '') is None
class TestForDeviceByResources:
def test_na9300k_without_one_ui_or_information_is_cooktop(self):
from custom_components.localthings.registry.by_type import for_device_by_resources
from tests.conftest import _load_device
reg = for_device_by_resources(_load_device('cooktop'))
assert reg is not None
assert reg.name == 'cooktop'
def test_unrelated_mode_options_are_not_cooktop(self):
from custom_components.localthings.registry.by_type import for_device_by_resources
resources = {
'/mode/vs/0': {
'x.com.samsung.da.options': [
'DeviceType_SOME_OVEN',
'UpperLamp_Off',
],
},
}
assert for_device_by_resources(resources) is None
def test_hood_resource_signature(self):
from custom_components.localthings.registry.by_type import for_device_by_resources
resources = {
'/hood/fanspeed/vs/0': {},
'/hood/lamp/vs/0': {},
}
reg = for_device_by_resources(resources)
assert reg is not None
assert reg.name == 'range_hood'
+8
View File
@@ -84,6 +84,14 @@ class TestRemoteControlFallback:
assert common.REMOTE_CONTROL_VS_FALLBACK.match_fn(
{}, {'/remotectrl/0': {}, '/remotectrl/vs/0': {}}) is False
def test_polled_warm_so_write_gating_stays_fresh(self):
"""coordinator.async_send_command blocks writes on this signal, so
it can't sit in the default 'cold' tier (refreshed only once per
30s summary poll) -- it needs the subscribe/subpoll cadence 'warm'
and 'hot' hrefs get instead."""
assert common.REMOTE_CONTROL_GENERIC.poll_tier == 'warm'
assert common.REMOTE_CONTROL_VS_FALLBACK.poll_tier == 'warm'
# ---------------------------------------------------------------------------
# Energy meter. instantaneousPower clamps negatives to 0, but the constant
+77
View File
@@ -0,0 +1,77 @@
"""Tests for the read-only cooktop capability profile."""
from custom_components.localthings.registry.adapter import flatten
from custom_components.localthings.registry.by_type import cooktop
from custom_components.localthings.registry.capabilities.cooktop import (
COOKTOP_MODE,
)
from custom_components.localthings.registry.discovery import discover
from tests.conftest import _load_device
def _state(resources):
bound = discover(
resources,
cooktop.REGISTRY.capabilities,
cooktop.REGISTRY.pattern_capabilities,
)
return flatten(bound, resources)
def test_real_cooktop_fixture_has_expected_idle_state():
state = _state(_load_device('cooktop'))
assert state['power_state'] is True
assert state['any_burner_active'] is False
assert state['burner_0_state'] == 'Ready'
assert state['burner_5_state'] == 'Ready'
assert state['main_timer_state'] == 'Ready'
assert state['main_timer_current'] == 0
assert state['cloud_connected'] is True
assert state['paired_hood_connected'] is False
assert state['paired_hood_power'] is False
assert state['paired_hood_fan_speed'] == 0
assert state['paired_hood_light'] is False
def test_run_operation_marks_any_burner_active():
resources = _load_device('cooktop')
options = resources['/mode/vs/0']['x.com.samsung.da.options']
options[options.index('OperationState3_Ready')] = 'OperationState3_Run'
state = _state(resources)
assert state['burner_3_state'] == 'Run'
assert state['any_burner_active'] is True
def test_cooktop_profile_has_no_write_functions():
"""A local integration must never expose unverified remote ignition."""
descriptions = [
desc
for capabilities in cooktop.REGISTRY.capabilities.values()
for capability in capabilities
for desc in capability.entities
]
assert descriptions
assert all(not hasattr(desc, 'write_fn') or desc.write_fn is None
for desc in descriptions)
def test_static_slot_superset_has_headroom_for_other_layouts():
keys = {desc.key for desc in COOKTOP_MODE.entities}
assert {f'burner_{slot}_state' for slot in range(8)} <= keys
def test_variant_slot_outside_na9300k_layout_is_discovered():
resources = _load_device('cooktop')
resources['/mode/vs/0']['x.com.samsung.da.options'].append(
'OperationState7_Ready',
)
state = _state(resources)
assert state['burner_7_state'] == 'Ready'
assert 'burner_2_state' not in state
+2 -1
View File
@@ -1,6 +1,6 @@
from custom_components.localthings.registry.entities import (
SensorDesc, BinarySensorDesc, SelectDesc, SwitchDesc, ButtonDesc,
NumberDesc, TimeDesc, ClimateDesc, PLATFORM_OF,
NumberDesc, TimeDesc, ClimateDesc, FanDesc, PLATFORM_OF,
)
@@ -28,6 +28,7 @@ def test_platform_mapping_covers_all_subclasses():
assert PLATFORM_OF[NumberDesc] == 'number'
assert PLATFORM_OF[TimeDesc] == 'time'
assert PLATFORM_OF[ClimateDesc] == 'climate'
assert PLATFORM_OF[FanDesc] == 'fan'
def test_select_carries_options_and_write_fn():
+47 -1
View File
@@ -7,7 +7,9 @@ GOLDEN = Path(__file__).parent / 'fixtures' / 'golden'
def _new_state_keys(name, resources):
from custom_components.localthings.registry.by_type import for_device, for_device_by_model
from custom_components.localthings.registry.by_type import (
for_device, for_device_by_model, for_device_by_resources,
)
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.adapter import flatten
otn = resources.get('/otninformation/vs/0', {})
@@ -19,6 +21,8 @@ def _new_state_keys(name, resources):
info.get('x.com.samsung.da.modelNum', ''),
info.get('x.com.samsung.da.description', ''),
)
if reg is None:
reg = for_device_by_resources(resources)
if reg is None:
from custom_components.localthings.registry.registry import CAPABILITIES
caps, pats = CAPABILITIES, []
@@ -81,6 +85,18 @@ def test_registry_reproduces_golden_state_keys_for_airconditioner():
)
def test_registry_reproduces_golden_state_keys_for_cooktop():
from tests.conftest import _load_device
resources = _load_device('cooktop')
golden = json.loads((GOLDEN / 'cooktop.json').read_text())
state_keys = _new_state_keys('cooktop', resources)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_registry_reproduces_golden_state_keys_for_tp1x_ref_21k_us():
from tests.conftest import _load_device
resources = _load_device('refrigerator_tp1x_ref_21k_us')
@@ -95,6 +111,18 @@ def test_registry_reproduces_golden_state_keys_for_tp1x_ref_21k_us():
)
def test_registry_reproduces_golden_state_keys_for_range_hood():
from tests.conftest import _load_device
resources = _load_device('range_hood')
golden = json.loads((GOLDEN / 'range_hood.json').read_text())
state_keys = _new_state_keys('range_hood', resources)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_registry_reproduces_golden_state_keys_for_washer_flexwash():
"""FlexWash twin washers (WV-prefix consumer model, e.g. WV55M9600AW)
report no oneUiVersion and previously fell through for_device_by_model's
@@ -216,6 +244,24 @@ def test_registry_reproduces_golden_state_keys_for_range():
)
def test_registry_reproduces_golden_state_keys_for_oven():
"""Wall oven (model TP1X_DA-KS-OVEN-0107X, issue #55) -- reports no
oneUiVersion; resolved via the '-OVEN-' modelNum token fallback in
for_device_by_model, mirroring the '-RANGE-' fallback added for
issue #44. Before that fallback existed the device type came back
'unknown' and every href fell through to the global CAPABILITIES
registry instead of the oven family's own."""
from tests.conftest import _load_device
resources = _load_device('oven')
golden = json.loads((GOLDEN / 'oven.json').read_text())
state_keys = _new_state_keys('oven', resources)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_resources_from_batch_preferred_over_flat():
from tests.conftest import _resources_from_dump
dump = {
+22
View File
@@ -1,8 +1,30 @@
"""Unit tests for oven-family capabilities."""
from custom_components.localthings.registry.by_type import for_device_by_model
from custom_components.localthings.registry.capabilities import oven
from custom_components.localthings.registry.discovery import discover
# ---------------------------------------------------------------------------
# Device-type detection + full-dump coverage (issue #55)
# ---------------------------------------------------------------------------
def test_oven_fixture_resolves_and_has_no_unbound_hrefs():
"""The issue #55 dump previously came back device_type='unknown' with
/connected/vs/0 unbound -- resolving via the '-OVEN-' modelNum token
fallback must leave every href in the oven registry bound or ignored."""
from tests.conftest import _load_device
resources = _load_device('oven')
info = resources['/information/vs/0']
reg = for_device_by_model(
info['x.com.samsung.da.modelNum'], info['x.com.samsung.da.description'])
assert reg is not None
assert reg.name == 'oven'
unbound = []
discover(resources, reg.capabilities, reg.pattern_capabilities, log=unbound.append)
assert unbound == []
# ---------------------------------------------------------------------------
# OVEN_SETPOINT — NumberDesc with RMW write semantics
# ---------------------------------------------------------------------------
+116
View File
@@ -0,0 +1,116 @@
"""Tests for the AHD-WW-TP1-22 range-hood profile."""
from custom_components.localthings.registry.adapter import flatten
from custom_components.localthings.registry.by_type import for_device_by_model
from custom_components.localthings.registry.capabilities import range_hood
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.entities import FanDesc
from tests.conftest import _load_device
def _hood():
resources = _load_device('range_hood')
info = resources['/information/vs/0']
registry = for_device_by_model(
info['x.com.samsung.da.modelNum'],
info['x.com.samsung.da.description'],
)
return registry, resources
def _state():
registry, resources = _hood()
bound = discover(
resources, registry.capabilities, registry.pattern_capabilities,
)
return flatten(bound, resources)
def test_ahd_model_resolves_to_range_hood_registry():
registry, _ = _hood()
assert registry is not None
assert registry.name == 'range_hood'
def test_range_hood_has_no_unbound_hrefs():
registry, resources = _hood()
unbound = []
discover(
resources,
registry.capabilities,
registry.pattern_capabilities,
log=unbound.append,
)
assert unbound == []
def test_range_hood_fixture_values():
state = _state()
assert state['fan'] == '14'
assert state['alarm_code'] == 'none'
assert state['automatic_operation'] is False
assert state['lamp'] is False
assert state['lamp_brightness'] == '2'
assert state['hood_filter_usage'] == 100
assert state['hood_filter_status'] == 'wash'
assert state['energy_kwh'] == 165.62
assert state['clean_level'] == 2
assert state['dust'] == 18
assert state['fine_dust'] == 14
assert state['super_fine_dust'] == 9
assert state['periodic_air_sensing'] is False
assert state['air_sensing_state'] == 'NonProcessing'
assert state['last_air_sensing_level'] == 'Kr2'
def test_one_composite_fan_is_bound():
registry, resources = _hood()
bound = discover(
resources, registry.capabilities, registry.pattern_capabilities,
)
fans = [entity for entity in bound if isinstance(entity.desc, FanDesc)]
assert len(fans) == 1
assert fans[0].href == '/hood/fanspeed/vs/0'
def test_fan_write_contract():
desc = range_hood.HOOD_FAN.entities[0]
rep = {
'x.com.samsung.da.hood.supportedFanSpeed': ['14', '15', '16', '17'],
}
assert desc.write_fn(('power', True), rep) == (
['power', '0'], {'value': True},
)
assert desc.write_fn(('power', False), rep) == (
['power', '0'], {'value': False},
)
assert desc.write_fn(('power', True, '/power/vs/0'), rep) == (
['power', 'vs', '0'], {'x.com.samsung.da.power': 'On'},
)
assert desc.write_fn(('power', False, '/power/vs/0'), rep) == (
['power', 'vs', '0'], {'x.com.samsung.da.power': 'Off'},
)
assert desc.write_fn(('power', True, '/not-a-power-resource'), rep) is None
assert desc.write_fn(('speed', '16'), rep) == (
['hood', 'fanspeed', 'vs', '0'],
{'x.com.samsung.da.hood.fanSpeed': '16'},
)
assert desc.write_fn(('speed', '99'), rep) is None
def test_lamp_write_contract():
lamp, brightness = range_hood.HOOD_LAMP.entities
rep = {'x.com.samsung.lamp.range': ['1', '2']}
assert lamp.write_fn('On', rep) == (
['hood', 'lamp', 'vs', '0'], {'x.com.samsung.lamp.power': 'On'},
)
assert lamp.write_fn('Off', rep) == (
['hood', 'lamp', 'vs', '0'], {'x.com.samsung.lamp.power': 'Off'},
)
assert brightness.write_fn('1', rep) == (
['hood', 'lamp', 'vs', '0'], {'x.com.samsung.lamp.current': '1'},
)
assert brightness.write_fn('2', rep) == (
['hood', 'lamp', 'vs', '0'], {'x.com.samsung.lamp.current': '2'},
)
assert brightness.write_fn('Unsupported', rep) is None
+88
View File
@@ -0,0 +1,88 @@
"""HA fan-entity mapping tests for the range hood."""
from custom_components.localthings.fan import LocalThingsRangeHoodFan
from custom_components.localthings.registry.by_type import range_hood
from custom_components.localthings.registry.discovery import discover
from custom_components.localthings.registry.entities import FanDesc
from tests.conftest import _load_device
class _FakeCoordinator:
device_serial = 'TEST-HOOD-SERIAL'
device_info = {}
data = {}
def __init__(self, resources):
self.last_resources = resources
self.commands = []
def resource(self, href):
return self.last_resources.get(href, {})
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
def _entity(resources, coordinator=None):
bound = discover(
resources,
range_hood.REGISTRY.capabilities,
range_hood.REGISTRY.pattern_capabilities,
)
fan_bound = next(item for item in bound if isinstance(item.desc, FanDesc))
return LocalThingsRangeHoodFan(
coordinator or _FakeCoordinator(resources), fan_bound,
)
def test_power_off_maps_to_zero_percent_and_four_retained_speeds():
entity = _entity(_load_device('range_hood'))
assert entity.is_on is False
assert entity.speed_count == 4
assert entity.percentage == 0
def test_active_codes_map_to_ordered_percentages():
resources = _load_device('range_hood')
resources['/power/0']['value'] = True
resources['/hood/fanspeed/vs/0'][
'x.com.samsung.da.hood.fanSpeed'
] = '14'
assert _entity(resources).percentage == 25
resources['/hood/fanspeed/vs/0'][
'x.com.samsung.da.hood.fanSpeed'
] = '15'
assert _entity(resources).percentage == 50
resources['/hood/fanspeed/vs/0'][
'x.com.samsung.da.hood.fanSpeed'
] = '16'
assert _entity(resources).percentage == 75
resources['/hood/fanspeed/vs/0'][
'x.com.samsung.da.hood.fanSpeed'
] = '17'
assert _entity(resources).percentage == 100
async def test_power_write_prefers_standard_resource_when_both_exist():
resources = _load_device('range_hood')
coordinator = _FakeCoordinator(resources)
entity = _entity(resources, coordinator)
await entity.async_turn_on()
assert coordinator.commands[-1][1] == ('power', True, '/power/0')
async def test_power_write_falls_back_to_vendor_resource():
resources = _load_device('range_hood')
resources.pop('/power/0')
coordinator = _FakeCoordinator(resources)
entity = _entity(resources, coordinator)
await entity.async_turn_off()
assert coordinator.commands[-1][1] == ('power', False, '/power/vs/0')