diff --git a/custom_components/localthings/__init__.py b/custom_components/localthings/__init__.py new file mode 100644 index 0000000..dfbc4c3 --- /dev/null +++ b/custom_components/localthings/__init__.py @@ -0,0 +1,33 @@ +"""Local Things — Samsung appliance local control integration.""" +from __future__ import annotations + +import logging + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady + +from .const import DOMAIN, PLATFORMS +from .coordinator import LocalThingsCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + hass.data.setdefault(DOMAIN, {}) + coordinator = LocalThingsCoordinator(hass, entry) + try: + await coordinator.async_config_entry_first_refresh() + except Exception as err: + raise ConfigEntryNotReady(f"Cannot connect to device: {err}") from err + hass.data[DOMAIN][entry.entry_id] = coordinator + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + unloaded = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + if unloaded: + coordinator: LocalThingsCoordinator = hass.data[DOMAIN].pop(entry.entry_id) + await hass.async_add_executor_job(coordinator._close_session) + return unloaded diff --git a/custom_components/localthings/binary_sensor.py b/custom_components/localthings/binary_sensor.py new file mode 100644 index 0000000..88c38f8 --- /dev/null +++ b/custom_components/localthings/binary_sensor.py @@ -0,0 +1,38 @@ +"""Binary sensor platform for Local Things.""" +from __future__ import annotations + +from homeassistant.components.binary_sensor import BinarySensorEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from samsung_appliance.registry.entities import BinarySensorDesc + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator +from .entity import LocalThingsEntity + + +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( + LocalThingsBinarySensor(coordinator, b) + for b in coordinator.bound + if isinstance(b.desc, BinarySensorDesc) + ) + + +class LocalThingsBinarySensor(LocalThingsEntity, BinarySensorEntity): + + def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None: + super().__init__(coordinator, bound) + desc: BinarySensorDesc = bound.desc + self._attr_device_class = desc.device_class + + @property + def is_on(self): + return (self.coordinator.data or {}).get(self._state_key) diff --git a/custom_components/localthings/config_flow.py b/custom_components/localthings/config_flow.py new file mode 100644 index 0000000..804470e --- /dev/null +++ b/custom_components/localthings/config_flow.py @@ -0,0 +1,169 @@ +"""Config flow for Local Things integration.""" +from __future__ import annotations + +import logging +import os +import tempfile +from typing import Any + +import voluptuous as vol +from homeassistant import config_entries +from homeassistant.data_entry_flow import FlowResult + +from .const import DOMAIN, CONF_HOST, CONF_PORT, CONF_CERT_PEM, CONF_KEY_PEM, PROBE_PORTS + +_LOGGER = logging.getLogger(__name__) + + +class CannotConnect(Exception): + pass + + +def _probe_and_validate(host: str, cert_pem: str, key_pem: str) -> dict: + """Try each known port; return {port, serial} on first success. + + DtlsCoapSession only accepts file paths, so PEM strings are written + to a temp-file pair and cleaned up in the finally block. + """ + import cbor2 + from samsung_appliance.coap_dtls import DtlsCoapSession + from samsung_appliance.batch import parse_device0_batch + + # Write PEM strings to temporary files — DtlsCoapSession requires + # cert_path / key_path rather than in-memory PEM strings. + cert_file = tempfile.NamedTemporaryFile( + mode='w', suffix='.pem', delete=False) + key_file = tempfile.NamedTemporaryFile( + mode='w', suffix='.pem', delete=False) + try: + cert_file.write(cert_pem) + cert_file.flush() + cert_file.close() + key_file.write(key_pem) + key_file.flush() + key_file.close() + + last_exc = None + for port in PROBE_PORTS: + sess = None + try: + sess = DtlsCoapSession( + host, port, + cert_path=cert_file.name, + key_path=key_file.name, + ) + sess.connect() + sess.start_reader() + code, payload = sess.get(['device', '0'], timeout=15.0) + if code != 0x45 or not payload: + raise CannotConnect( + f"port {port}: unexpected code {code:#04x}") + body = cbor2.loads(payload) + resources = ( + parse_device0_batch(body) if isinstance(body, list) + else {} + ) + serial = ( + resources + .get('/information/vs/0', {}) + .get('x.com.samsung.da.serialNum', '') + ) + if not serial: + serial = ( + resources + .get('/otninformation/vs/0', {}) + .get('swVersionInfo', {}) + .get('firmwareVersion', '') + ) + if not serial: + serial = f"{host}:{port}" + return {"port": port, "serial": serial} + except CannotConnect: + raise + except Exception as exc: + last_exc = exc + _LOGGER.debug("port %d failed: %s", port, exc) + finally: + if sess is not None: + try: + sess.close() + except Exception: + pass + raise CannotConnect(f"no port responded on {host}: {last_exc}") + finally: + for path in (cert_file.name, key_file.name): + try: + os.unlink(path) + except OSError: + pass + + +class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + VERSION = 1 + + def __init__(self) -> None: + self._host: str = "" + self._cert_pem: str = "" + self._key_pem: str = "" + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + existing = self.hass.config_entries.async_entries(DOMAIN) + has_creds = bool(existing) + + errors: dict[str, str] = {} + + if user_input is not None: + self._host = user_input[CONF_HOST].strip() + if has_creds: + self._cert_pem = existing[0].data[CONF_CERT_PEM] + self._key_pem = existing[0].data[CONF_KEY_PEM] + else: + self._cert_pem = user_input[CONF_CERT_PEM].strip() + self._key_pem = user_input[CONF_KEY_PEM].strip() + + try: + info = await self.hass.async_add_executor_job( + _probe_and_validate, + self._host, + self._cert_pem, + self._key_pem, + ) + except CannotConnect: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Unexpected error during device probe") + errors["base"] = "unknown" + else: + await self.async_set_unique_id( + f"localthings_{info['serial']}" + ) + self._abort_if_unique_id_configured() + return self.async_create_entry( + title=f"Samsung Appliance ({self._host})", + data={ + CONF_HOST: self._host, + CONF_PORT: info["port"], + CONF_CERT_PEM: self._cert_pem, + CONF_KEY_PEM: self._key_pem, + }, + ) + + if has_creds: + schema = vol.Schema({vol.Required(CONF_HOST): str}) + reuse_note = "Certificate and key will be reused from the existing device." + else: + schema = vol.Schema({ + vol.Required(CONF_HOST): str, + vol.Required(CONF_CERT_PEM): str, + vol.Required(CONF_KEY_PEM): str, + }) + reuse_note = "" + + return self.async_show_form( + step_id="user", + data_schema=schema, + errors=errors, + description_placeholders={"reuse_note": reuse_note}, + ) diff --git a/custom_components/localthings/const.py b/custom_components/localthings/const.py new file mode 100644 index 0000000..06717da --- /dev/null +++ b/custom_components/localthings/const.py @@ -0,0 +1,13 @@ +DOMAIN = "localthings" + +PLATFORMS = ["sensor", "binary_sensor", "switch", "number", "select"] + +CONF_HOST = "host" +CONF_PORT = "port" +CONF_CERT_PEM = "cert_pem" +CONF_KEY_PEM = "key_pem" + +PROBE_PORTS = [49154, 49155] + +ACTIVE_INTERVAL_S = 5.0 +IDLE_INTERVAL_S = 30.0 diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py new file mode 100644 index 0000000..ad65b64 --- /dev/null +++ b/custom_components/localthings/coordinator.py @@ -0,0 +1,221 @@ +"""Coordinator for Local Things integration.""" +from __future__ import annotations + +import logging +import tempfile +import os +from datetime import timedelta +from typing import Any + +import cbor2 + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from homeassistant.helpers.device_registry import DeviceInfo + +from samsung_appliance.batch import parse_device0_batch +from samsung_appliance.coap_dtls import DtlsCoapSession +from samsung_appliance.registry.by_type import for_device +from samsung_appliance.registry.discovery import discover, BoundEntity +from samsung_appliance.registry import CAPABILITIES +from samsung_appliance.registry.adapter import flatten, is_active +from samsung_appliance.registry.identity import read_identity, DeviceIdentity + +from .const import ( + DOMAIN, CONF_HOST, CONF_PORT, CONF_CERT_PEM, CONF_KEY_PEM, + ACTIVE_INTERVAL_S, IDLE_INTERVAL_S, +) + +_LOGGER = logging.getLogger(__name__) + +_SEED_PATH = ['device', '0'] + + +class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): + """Manages one Samsung appliance: session, discovery, polling.""" + + bound: list[BoundEntity] + device_info: DeviceInfo + device_serial: str + + def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: + super().__init__( + hass, + _LOGGER, + name=f"{DOMAIN}_{entry.data[CONF_HOST]}", + update_interval=timedelta(seconds=IDLE_INTERVAL_S), + ) + self._entry = entry + self._session: DtlsCoapSession | None = None + self._identity: DeviceIdentity | None = None + self._discovered = False + self.bound = [] + self._last_resources: dict[str, dict] = {} + self.device_serial = entry.data[CONF_HOST] # placeholder until first poll + self.device_info = DeviceInfo( + identifiers={(DOMAIN, entry.data[CONF_HOST])}, + name=f"Samsung Appliance ({entry.data[CONF_HOST]})", + manufacturer="Samsung", + ) + + # ------------------------------------------------------------------ + # Session management (all blocking — must run in executor) + # ------------------------------------------------------------------ + + def _connect_session(self) -> None: + host = self._entry.data[CONF_HOST] + port = self._entry.data[CONF_PORT] + cert_pem: str = self._entry.data[CONF_CERT_PEM] + key_pem: str = self._entry.data[CONF_KEY_PEM] + + cert_f = tempfile.NamedTemporaryFile(suffix='.pem', delete=False, mode='w') + key_f = tempfile.NamedTemporaryFile(suffix='.pem', delete=False, mode='w') + try: + cert_f.write(cert_pem); cert_f.flush(); cert_f.close() + key_f.write(key_pem); key_f.flush(); key_f.close() + sess = DtlsCoapSession(host, port, + cert_path=cert_f.name, + key_path=key_f.name) + sess.connect() + sess.start_reader() + self._session = sess + _LOGGER.debug("DTLS connected to %s:%d", host, port) + try: + self._identity = read_identity(sess, None) + except Exception as e: + _LOGGER.debug("read_identity failed: %s", e) + self._identity = None + finally: + for f in (cert_f.name, key_f.name): + try: + os.unlink(f) + except OSError: + pass + + def _close_session(self) -> None: + sess = self._session + self._session = None + if sess is not None: + try: + sess.close() + except Exception: + pass + + def _poll_once(self) -> dict[str, dict]: + """GET /device/0, return parsed resources. Blocking.""" + if self._session is None: + self._connect_session() + sess = self._session + try: + code, payload = sess.get(_SEED_PATH, timeout=15.0) + except Exception as e: + self._close_session() + raise RuntimeError(f"poll GET failed: {e}") from e + if code != 0x45 or not payload: + self._close_session() + raise RuntimeError(f"poll: unexpected code {code:#04x}") + try: + body = cbor2.loads(payload) + except Exception as e: + raise RuntimeError(f"poll cbor decode: {e}") from e + return parse_device0_batch(body) if isinstance(body, list) else {} + + # ------------------------------------------------------------------ + # Discovery (runs once on first successful poll) + # ------------------------------------------------------------------ + + def _run_discovery(self, resources: dict[str, dict]) -> None: + one_ui = (resources.get('/otninformation/vs/0', {}) + .get('swVersionInfo', {}) + .get('oneUiVersion', '')) + reg = for_device(one_ui) if one_ui else None + if reg is not None: + _LOGGER.debug("device type: %s (oneUiVersion=%r)", reg.name, one_ui) + bound = discover(resources, reg.capabilities, reg.pattern_capabilities) + else: + _LOGGER.warning("unknown device type oneUiVersion=%r; using common caps", one_ui) + bound = discover(resources, CAPABILITIES) + self.bound = bound + + serial = (resources.get('/information/vs/0', {}) + .get('x.com.samsung.da.serialNum', '')) + if not serial: + serial = self._entry.data[CONF_HOST] + self.device_serial = serial + + ident = self._identity + model = (ident.model if ident else '') or (reg.name.title() if reg else 'Appliance') + name = (ident.name if ident else '') or f"Samsung {model}" + mfr = (ident.manufacturer if ident else '') or 'Samsung' + + self.device_info = DeviceInfo( + identifiers={(DOMAIN, serial)}, + name=name, + manufacturer=mfr, + model=model, + ) + self._discovered = True + _LOGGER.info("discovered %d entities (serial=%s)", len(bound), serial) + + # ------------------------------------------------------------------ + # DataUpdateCoordinator hook + # ------------------------------------------------------------------ + + async def _async_update_data(self) -> dict[str, Any]: + try: + resources = await self.hass.async_add_executor_job(self._poll_once) + except Exception as e: + # One reconnect attempt + _LOGGER.warning("poll failed, reconnecting: %s", e) + try: + await self.hass.async_add_executor_job(self._close_session) + resources = await self.hass.async_add_executor_job(self._poll_once) + except Exception as e2: + raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 + + self._last_resources = resources + + if not self._discovered: + self._run_discovery(resources) + + # Adjust polling interval dynamically + active = is_active(self.bound, resources) + self.update_interval = timedelta( + seconds=ACTIVE_INTERVAL_S if active else IDLE_INTERVAL_S + ) + + return flatten(self.bound, resources) + + # ------------------------------------------------------------------ + # Command dispatch (called by entity platforms in Task 5) + # ------------------------------------------------------------------ + + async def async_send_command(self, bound_entity: BoundEntity, + payload: Any) -> None: + """Write a value to the device. Fire-and-forget style.""" + desc = bound_entity.desc + write_fn = getattr(desc, 'write_fn', None) + if write_fn is None: + return + href = bound_entity.href + rep = self._last_resources.get(href or '', {}) + result = write_fn(payload, rep, href) + if result is None: + _LOGGER.warning("write_fn rejected payload %r for %s", payload, href) + return + path_segs, body = result + + def _do_put(): + sess = self._session + if sess is None: + raise RuntimeError("no session") + code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0) + _LOGGER.info("PUT %s → code %#04x", href, code) + + try: + await self.hass.async_add_executor_job(_do_put) + except Exception as e: + _LOGGER.error("command failed for %s: %s", href, e) + else: + await self.async_request_refresh() diff --git a/custom_components/localthings/entity.py b/custom_components/localthings/entity.py new file mode 100644 index 0000000..326ff51 --- /dev/null +++ b/custom_components/localthings/entity.py @@ -0,0 +1,31 @@ +"""Base entity for Local Things.""" +from __future__ import annotations + +from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.helpers.device_registry import DeviceInfo + +from samsung_appliance.registry.adapter import _key +from samsung_appliance.registry.discovery import BoundEntity + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator + + +class LocalThingsEntity(CoordinatorEntity[LocalThingsCoordinator]): + """Base class for all Local Things entities.""" + + _attr_has_entity_name = True + + def __init__(self, coordinator: LocalThingsCoordinator, bound: BoundEntity) -> None: + super().__init__(coordinator) + self._bound = bound + self._state_key = _key(bound) + self._attr_unique_id = f"{DOMAIN}_{coordinator.device_serial}_{self._state_key}" + self._attr_name = bound.desc.name + self._attr_icon = bound.desc.icon + self._attr_entity_category = bound.desc.entity_category + self._attr_entity_registry_enabled_default = bound.desc.enabled_default + + @property + def device_info(self) -> DeviceInfo: + return self.coordinator.device_info diff --git a/custom_components/localthings/manifest.json b/custom_components/localthings/manifest.json new file mode 100644 index 0000000..d62e8ff --- /dev/null +++ b/custom_components/localthings/manifest.json @@ -0,0 +1,10 @@ +{ + "domain": "localthings", + "name": "Local Things", + "version": "0.1.0", + "config_flow": true, + "iot_class": "local_polling", + "requirements": ["cbor2>=5.4.6", "pyOpenSSL>=22.0.0"], + "dependencies": [], + "codeowners": [] +} diff --git a/custom_components/localthings/number.py b/custom_components/localthings/number.py new file mode 100644 index 0000000..73bda23 --- /dev/null +++ b/custom_components/localthings/number.py @@ -0,0 +1,50 @@ +"""Number platform for Local Things.""" +from __future__ import annotations + +from homeassistant.components.number import NumberEntity, NumberMode +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from samsung_appliance.registry.entities import NumberDesc + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator +from .entity import LocalThingsEntity + + +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( + LocalThingsNumber(coordinator, b) + for b in coordinator.bound + if isinstance(b.desc, NumberDesc) + ) + + +class LocalThingsNumber(LocalThingsEntity, NumberEntity): + + _attr_mode = NumberMode.BOX + + def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None: + super().__init__(coordinator, bound) + desc: NumberDesc = bound.desc + self._attr_native_unit_of_measurement = desc.unit + self._attr_device_class = desc.device_class + if desc.native_min is not None: + self._attr_native_min_value = desc.native_min + if desc.native_max is not None: + self._attr_native_max_value = desc.native_max + if desc.step is not None: + self._attr_native_step = desc.step + + @property + def native_value(self): + return (self.coordinator.data or {}).get(self._state_key) + + async def async_set_native_value(self, value: float) -> None: + await self.coordinator.async_send_command(self._bound, value) diff --git a/custom_components/localthings/select.py b/custom_components/localthings/select.py new file mode 100644 index 0000000..e7e31b8 --- /dev/null +++ b/custom_components/localthings/select.py @@ -0,0 +1,42 @@ +"""Select platform for Local Things.""" +from __future__ import annotations + +from homeassistant.components.select import SelectEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from samsung_appliance.registry.entities import SelectDesc + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator +from .entity import LocalThingsEntity + + +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( + LocalThingsSelect(coordinator, b) + for b in coordinator.bound + if isinstance(b.desc, SelectDesc) + ) + + +class LocalThingsSelect(LocalThingsEntity, SelectEntity): + + def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None: + super().__init__(coordinator, bound) + desc: SelectDesc = bound.desc + opts = desc.options + self._attr_options = list(opts() if callable(opts) else opts) + + @property + def current_option(self): + return (self.coordinator.data or {}).get(self._state_key) + + async def async_select_option(self, option: str) -> None: + await self.coordinator.async_send_command(self._bound, option) diff --git a/custom_components/localthings/sensor.py b/custom_components/localthings/sensor.py new file mode 100644 index 0000000..fed4f74 --- /dev/null +++ b/custom_components/localthings/sensor.py @@ -0,0 +1,40 @@ +"""Sensor platform for Local Things.""" +from __future__ import annotations + +from homeassistant.components.sensor import SensorEntity, SensorDeviceClass, SensorStateClass +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from samsung_appliance.registry.entities import SensorDesc + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator +from .entity import LocalThingsEntity + + +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( + LocalThingsSensor(coordinator, b) + for b in coordinator.bound + if isinstance(b.desc, SensorDesc) + ) + + +class LocalThingsSensor(LocalThingsEntity, SensorEntity): + + def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None: + super().__init__(coordinator, bound) + desc: SensorDesc = bound.desc + self._attr_native_unit_of_measurement = desc.unit + self._attr_device_class = desc.device_class + self._attr_state_class = desc.state_class + + @property + def native_value(self): + return (self.coordinator.data or {}).get(self._state_key) diff --git a/custom_components/localthings/strings.json b/custom_components/localthings/strings.json new file mode 100644 index 0000000..f53dcc1 --- /dev/null +++ b/custom_components/localthings/strings.json @@ -0,0 +1,22 @@ +{ + "config": { + "step": { + "user": { + "title": "Add Samsung Appliance", + "description": "Enter the IP address of your Samsung appliance. {reuse_note}", + "data": { + "host": "IP Address", + "cert_pem": "Certificate (PEM)", + "key_pem": "Private Key (PEM)" + } + } + }, + "error": { + "cannot_connect": "Cannot connect to the device. Check the IP address and credentials.", + "unknown": "Unexpected error." + }, + "abort": { + "already_configured": "This device is already configured." + } + } +} diff --git a/custom_components/localthings/switch.py b/custom_components/localthings/switch.py new file mode 100644 index 0000000..927dcf8 --- /dev/null +++ b/custom_components/localthings/switch.py @@ -0,0 +1,44 @@ +"""Switch platform for Local Things.""" +from __future__ import annotations + +from homeassistant.components.switch import SwitchEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from samsung_appliance.registry.entities import SwitchDesc + +from .const import DOMAIN +from .coordinator import LocalThingsCoordinator +from .entity import LocalThingsEntity + + +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( + LocalThingsSwitch(coordinator, b) + for b in coordinator.bound + if isinstance(b.desc, SwitchDesc) + ) + + +class LocalThingsSwitch(LocalThingsEntity, SwitchEntity): + + def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None: + super().__init__(coordinator, bound) + desc: SwitchDesc = bound.desc + self._attr_device_class = desc.device_class + + @property + def is_on(self): + return (self.coordinator.data or {}).get(self._state_key) + + async def async_turn_on(self, **kwargs) -> None: + await self.coordinator.async_send_command(self._bound, 'On') + + async def async_turn_off(self, **kwargs) -> None: + await self.coordinator.async_send_command(self._bound, 'Off') diff --git a/main.py b/main.py deleted file mode 100644 index 2ff6b3f..0000000 --- a/main.py +++ /dev/null @@ -1,179 +0,0 @@ -#!/usr/bin/env python3 -"""SmartThings-Local Bridge — entry point. - -One process supervises N Samsung appliances over their OCF CoAP-DTLS -local APIs, publishing state + HA discovery to MQTT. Each appliance -runs its own DTLS session in its own thread, with an in-session -PollScheduler driving state freshness and a KeepaliveTask driving -DTLS-layer liveness. MQTT is shared. - -Config is env-var driven: - * APPLIANCE_COUNT plus APPLIANCE__{CLASS,IP,OCF_PORT,TOPIC,NAME} - define the appliances to bridge. - * Shared keys (MQTT_*, HA_DISCOVERY_PREFIX, CERT_PATH, KEY_PATH, - HEALTH_INTERVAL_S, PING_INTERVAL_S) apply to all. - -Reconnects on session errors per-appliance; shuts down cleanly on -SIGINT / SIGTERM.""" -import logging -import os -import signal -import sys -import threading - -import paho.mqtt.client as mqtt - -from samsung_appliance.bridge import PushBridge -from samsung_appliance.config import SharedConfig, load_appliances -from samsung_appliance.logger import logger - - -def main(): - shared = SharedConfig.from_env() - - try: - appliances = load_appliances() - except ValueError as e: - logger.error("config: %s", e) - return 2 - - if not shared.MQTT_BROKER: - logger.error("config: MQTT_BROKER not set") - return 2 - for path in (shared.CERT_PATH, shared.KEY_PATH): - if not path.exists(): - logger.error("client cert/key not found: %s", path) - return 2 - - logger.info("SmartThings-Local Bridge starting (%d appliance%s)", - len(appliances), '' if len(appliances) == 1 else 's') - logger.info(" broker = %s:%d (user=%s)", - shared.MQTT_BROKER, shared.MQTT_PORT, - shared.MQTT_USER or '') - for app in appliances: - port = app.ocf_port or 49154 - logger.info(" [%d] %s @ %s:%d (DTLS) → topic %s/*", - app.index, app.klass or '?', app.ip, port, app.topic_prefix) - - # --- MQTT client (shared) --- - cli = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, - client_id='smartthings_local_bridge') - if shared.MQTT_USER: - cli.username_pw_set(shared.MQTT_USER, shared.MQTT_PASS) - # Headroom for the on_connect burst across all appliances (each - # publishes ~30 retained QoS-1 messages: discovery + state + avail). - # 100 + 50*N keeps us comfortably ahead of paho's default 20. - cli.max_inflight_messages_set(100 + 50 * len(appliances)) - - # Use the FIRST appliance's availability topic for LWT — paho only - # supports one will message. If a second appliance is added later, - # its availability is managed via explicit publishes on disconnect - # rather than LWT. For Phase 1 (dryer only) this is exact. - first_app = appliances[0] - cli.will_set(f"{first_app.topic_prefix}/availability", - payload='offline', qos=1, retain=True) - - # Build bridges. Each will discover entities at first connect. - bridges: list[PushBridge] = [PushBridge(shared, app, cli) - for app in appliances] - by_prefix = {b.cmd_topic_prefix.rstrip('/'): b for b in bridges} - - def on_connect(client, userdata, flags, rc, props=None): - if rc != 0: - logger.warning("MQTT connect rc=%s", rc) - return - logger.info("MQTT connected → %s:%d", - shared.MQTT_BROKER, shared.MQTT_PORT) - for b in bridges: - cmd_wildcard = f"{b.app.topic_prefix}/cmd/#" - client.subscribe(cmd_wildcard, qos=1) - b.reassert_availability() - logger.info("subscribed to %d cmd wildcards", len(bridges)) - - def on_disconnect(client, userdata, flags, rc, props=None): - logger.warning("MQTT disconnected rc=%s", rc) - - def on_message(client, userdata, msg): - # Route by topic prefix. Each appliance owns a distinct - # `/cmd/*` namespace, so the prefix-match is unambiguous. - for prefix, bridge in by_prefix.items(): - if msg.topic.startswith(prefix + '/'): - try: - payload = msg.payload.decode('utf-8', - errors='replace').strip() - except Exception: - return - bridge.handle_command(msg.topic, payload) - return - - cli.on_connect = on_connect - cli.on_disconnect = on_disconnect - cli.on_message = on_message - - if os.getenv('PAHO_DEBUG'): - paho_logger = logging.getLogger('paho.mqtt.client') - paho_logger.setLevel(logging.DEBUG) - paho_handler = logging.StreamHandler(sys.stdout) - paho_handler.setFormatter(logging.Formatter( - '%(asctime)s PAHO %(message)s', datefmt='%H:%M:%S')) - paho_logger.addHandler(paho_handler) - paho_logger.propagate = False - cli.enable_logger(paho_logger) - - cli.connect_async(shared.MQTT_BROKER, shared.MQTT_PORT, keepalive=60) - cli.loop_start() - - # Per-bridge runner / health / heartbeat threads. - threads: list[threading.Thread] = [] - - def make_health(b: PushBridge): - def loop(): - while not b.stop.is_set(): - b.publish_health() - if b.stop.wait(shared.HEALTH_INTERVAL_S): - break - return loop - - for b in bridges: - tag = b.app.klass - threads.append(threading.Thread( - target=b.run_forever, daemon=True, name=f'{tag}-session')) - threads.append(threading.Thread( - target=make_health(b), daemon=True, name=f'{tag}-health')) - - stopping = threading.Event() - - def shutdown(*_): - if stopping.is_set(): - return - stopping.set() - logger.info("shutting down…") - for b in bridges: - b.stop.set() - try: b.set_availability(False) - except Exception: pass - - signal.signal(signal.SIGINT, shutdown) - signal.signal(signal.SIGTERM, shutdown) - - for t in threads: - t.start() - - # Wait for the session threads (the only non-daemon-equivalent - # loops). They exit when their bridge's stop event is set. - try: - for t in threads: - if t.name.endswith('-session'): - t.join() - finally: - try: - cli.loop_stop() - cli.disconnect() - except Exception: - pass - logger.info("stopped") - return 0 - - -if __name__ == '__main__': - sys.exit(main()) diff --git a/samsung_appliance/__init__.py b/samsung_appliance/__init__.py index 5f4f4ee..debc576 100644 --- a/samsung_appliance/__init__.py +++ b/samsung_appliance/__init__.py @@ -1,5 +1,5 @@ -"""Samsung appliance local-API → MQTT bridge with HA discovery. +"""Samsung appliance local-API library. -Multi-device: dryer, oven, etc. Entities are discovered at connect-time -from the live device resources via samsung_appliance.registry.""" +CoAP/DTLS session management, OCF batch resource parsing, and a +capability registry for discovering and reading Samsung appliance state.""" __version__ = "2.0.0" diff --git a/samsung_appliance/batch.py b/samsung_appliance/batch.py new file mode 100644 index 0000000..6a90c2e --- /dev/null +++ b/samsung_appliance/batch.py @@ -0,0 +1,19 @@ +"""OCF /device/0 batch response parser.""" +from __future__ import annotations + + +def parse_device0_batch(device0: list) -> dict[str, dict]: + """Extract {href: rep} from a /device/0 CBOR list response.""" + out = {} + for entry in device0[1:]: # skip [0] (device-level rep) + if not isinstance(entry, dict): + continue + href = entry.get('href') + rep = entry.get('rep') + if not href: + continue + # rep == {"href": "..."} is a stub (resource present, no current data). + # Include it as {} so capabilities still bind and the entity exists. + if isinstance(rep, dict): + out[href] = {} if set(rep.keys()) == {'href'} else rep + return out diff --git a/samsung_appliance/bridge.py b/samsung_appliance/bridge.py deleted file mode 100644 index c5f2973..0000000 --- a/samsung_appliance/bridge.py +++ /dev/null @@ -1,644 +0,0 @@ -"""Bridge: OCF CoAP-DTLS appliance → MQTT, batch-poll architecture. - - Appliance ──CoAP DTLS─► PushBridge ──MQTT──► Home Assistant - -State freshness comes from a periodic batch GET to /device/0 over a -persistent DTLS session. The poll interval switches between -active_interval_s and idle_interval_s based on whether the device -is in an active state. - -DTLS-layer liveness is a separate KeepaliveTask (CoAP empty-CON ping -every PING_INTERVAL_S). Three consecutive failures publish MQTT -availability=offline. - -Multiple PushBridges run concurrently — see main.py. They share one -MQTT client; each owns one DTLS session, one cache, one poll thread. -""" -import json -import os -import threading -import time - -import cbor2 - -from .coap_dtls import DtlsCoapSession, fmt_code -from .config import ApplianceConfig, SharedConfig -from .keepalive import KeepaliveTask -from .logger import bridge_logger -from .registry.by_type import for_device -from .registry.discovery import discover -from .state_cache import StateCache - - -DEBUG_BRIDGE = os.environ.get('DEBUG_BRIDGE') == '1' - -# /device/0 is the OCF resource that contains the full link dict used -# to seed the state cache on every connect. -SEED_PATH = ['device', '0'] - - -def _href_to_segs(href: str) -> list[str]: - return [s for s in href.split('/') if s] - - -def _parse_device0_batch(device0: list) -> dict[str, dict]: - """Extract {href: rep} from a /device/0 CBOR list response.""" - out = {} - for entry in device0[1:]: # skip [0] (device-level rep) - if not isinstance(entry, dict): - continue - href = entry.get('href') - rep = entry.get('rep') - if not href: - continue - # rep == {"href": "..."} is a stub (resource present, no current data). - # Include it as {} so capabilities still bind and the entity exists. - if isinstance(rep, dict): - out[href] = {} if set(rep.keys()) == {'href'} else rep - return out - - -SERIAL_PATH = '/information/vs/0' -SERIAL_FIELD = 'x.com.samsung.da.serialNum' - - -def _encode(cfg: dict) -> bytes: - return json.dumps(cfg).encode() - - -def _bridge_diagnostic_discovery(topic_prefix: str, - ha_discovery_prefix: str, - device_name: str, - model: str = 'Bridge') -> list[tuple[str, bytes]]: - """HA-discovery payloads for the per-bridge diagnostic entities - (polling stats, freshness). Returned as a list of - (topic, payload-bytes) tuples to be merged with entity discovery.""" - avail_topic = f"{topic_prefix}/availability" - health_topic = f"{topic_prefix}/bridge/health" - device = { - 'identifiers': [topic_prefix], - 'name': device_name, - 'manufacturer': 'Samsung', - 'model': model, - } - avail = [{'topic': avail_topic, - 'payload_available': 'online', - 'payload_not_available': 'offline'}] - - def sensor(slug: str, name: str, value_template: str, - unit: str | None = None, device_class: str | None = None, - icon: str | None = None) -> tuple[str, bytes]: - cfg = { - 'name': name, - 'unique_id': f"{topic_prefix}_bridge_{slug}", - 'object_id': f"{topic_prefix}_bridge_{slug}", - 'state_topic': health_topic, - 'value_template': value_template, - 'availability': avail, - 'device': device, - 'entity_category': 'diagnostic', - } - if unit is not None: cfg['unit_of_measurement'] = unit - if device_class is not None: cfg['device_class'] = device_class - if icon is not None: cfg['icon'] = icon - topic = (f"{ha_discovery_prefix}/sensor/" - f"{topic_prefix}/bridge_{slug}/config") - return topic, _encode(cfg) - - return [ - sensor('update_source', 'Last Update Source', - "{{ value_json.last_change_source | default('?') }}", - icon='mdi:transit-connection-variant'), - sensor('stalest_age_s', 'Stalest Resource Age', - "{{ value_json.stalest_age_s | default(0) }}", - unit='s', icon='mdi:clock-alert-outline'), - sensor('poll_errors', 'Poll Errors (window)', - "{{ value_json.poll_window_errors | default(0) }}", - icon='mdi:alert-circle-outline'), - ] - - -class PushBridge: - - def __init__(self, - shared: SharedConfig, - app: ApplianceConfig, - mqtt_client): - self.shared = shared - self.app = app - self.mqtt = mqtt_client - - self.log = bridge_logger(app.klass) - self._serial: str | None = None - - self.port = app.ocf_port or 49154 - - self.session: DtlsCoapSession | None = None - self._poll_thread: threading.Thread | None = None - self.keepalive: KeepaliveTask | None = None - - # descriptor is None until _discover_and_publish completes after seed. - self.descriptor = None - self._discovered = False - self._entity_payloads: list = [] - - self.cache = StateCache() - self.cache.set_on_change(self._on_cache_change) - - self.last_state_pub = None - self.last_remote_pub = None - self.last_cycle_pub = None - self.last_avail_pub: str | None = None - self.stop = threading.Event() - self.started_ts = time.time() - self.session_started_ts = None - self.last_change_ts = None - self.last_seed_ts = None - self.notif_count = 0 - self.connect_count = 0 - self.error_count = 0 - self._publish_gate = False - self._last_change_source: str | None = None - - # Poll counters — accumulated across reconnects. - self._poll_count = 0 - self._poll_error_count = 0 - self._win_prev_poll = 0 - self._win_prev_poll_err = 0 - self._win_prev_ping_fail = 0 - - p = app.topic_prefix - self.state_topic = f"{p}/state" - self.avail_topic = f"{p}/availability" - self.remote_topic = f"{p}/remote_available" - self.cycle_topic = f"{p}/cycle_active" - self.health_topic = f"{p}/bridge/health" - self.cmd_handlers = {} - self.cmd_topic_prefix = f"{p}/cmd/" - - # ---- cache plumbing --------------------------------------------- - - def _on_cache_change(self, changed: bool, source: str) -> None: - if changed: - self.notif_count += 1 - self.last_change_ts = time.time() - self._last_change_source = source - self.maybe_publish_state() - - def _debug_log_rep(self, href, rep): - if href == '/mode/vs/0' and isinstance(rep, dict): - self.log.info("mode modes=%r options=%r", - rep.get('x.com.samsung.da.modes'), - rep.get('x.com.samsung.da.options')) - elif href in ('/operational/state/vs/0', '/oven/vs/0', '/power/vs/0'): - self.log.info("REP %s = %r", href, rep) - - def _retag_logger_with_serial(self): - if self._serial is not None: - return - info = self.cache.get(SERIAL_PATH) or {} - serial = info.get(SERIAL_FIELD) - if not serial: - return - self._serial = serial - self.log = bridge_logger(self.app.klass, serial) - self.log.info("identified — serial=%s", serial) - - # ---- session lifecycle ------------------------------------------ - - def session_once(self): - sess = DtlsCoapSession( - self.app.ip, self.port, - cert_path=self.shared.CERT_PATH, - key_path=self.shared.KEY_PATH, - ) - sess.connect() - self.session = sess - self.session_started_ts = time.time() - self.connect_count += 1 - self.cache.descriptor_state.clear() - self._publish_gate = False - - self.log.info("DTLS connected — port %d", self.port) - - sess.start_reader() - - session_ended = threading.Event() - - def _stop_watcher(): - while not session_ended.is_set(): - if self.stop.wait(1.0): - try: - sess.close() - except Exception as e: - self.log.warning("stop close: %s", e) - return - - threading.Thread(target=_stop_watcher, daemon=True, - name=f'{self.app.klass}-stopw').start() - - try: - self._run_session_inner(sess) - finally: - session_ended.set() - - def _run_session_inner(self, sess): - # Seed first — populate cache before discovery can run. - # The publish gate stays closed so seed writes don't trigger - # per-resource publishes before the full snapshot is ready. - self._seed_from_device0(sess) - self._retag_logger_with_serial() - - if DEBUG_BRIDGE: - self._debug_dump_links(sess) - - # Discover entities from the seeded cache, build the runtime - # descriptor, and publish HA discovery payloads. Idempotent: - # on reconnect the descriptor is reused from the first session. - self._discover_and_publish(sess) - - self._publish_gate = True - self.maybe_publish_state(force=True) - self.set_availability(True) - self.log.info("seeded → %d links; sensors live", - len(self.cache.links)) - - keepalive = KeepaliveTask( - sess, - interval_s=float(self.shared.PING_INTERVAL_S), - fail_threshold=3, - on_reachable=self._on_reachable, - on_unreachable=self._on_unreachable, - logger=self.log, - ) - self.keepalive = keepalive - - poll_t = threading.Thread( - target=self._poll_loop, args=(sess,), - daemon=True, name=f'{self.app.klass}-poll') - ka_t = threading.Thread( - target=keepalive.run_forever, args=(self.stop,), - daemon=True, name=f'{self.app.klass}-ping') - poll_t.start() - ka_t.start() - - try: - sess.join() - finally: - self.keepalive = None - - def _seed_from_device0(self, sess): - code, pl = sess.get(SEED_PATH, timeout=15.0) - if code != 0x45: - raise RuntimeError( - f"/{'/'.join(SEED_PATH)} -> {fmt_code(code)}") - try: - body = cbor2.loads(pl) - except Exception as e: - raise RuntimeError( - f"/{'/'.join(SEED_PATH)} cbor decode: {e}" - ) from e - # During the seed we want the cache populated without triggering - # a publish per resource — gate the on_change callback off until - # the publish gate opens just below. - resources = _parse_device0_batch(body) if isinstance(body, list) else {} - for href, rep in resources.items(): - if href not in self.cache.links: - self.cache.apply_rep(href, rep, source='seed') - self.last_seed_ts = time.time() - - def _discover_and_publish(self, sess): - """Run entity discovery against the seeded cache and publish HA - discovery payloads. Idempotent — on reconnect the descriptor - built during the first session is reused; entity payloads are - re-published via reassert_availability instead.""" - if self._discovered: - return - from .registry import CAPABILITIES, build_runtime_descriptor - from .registry.identity import read_identity - resources = self.cache.snapshot() - one_ui_version = (resources.get('/otninformation/vs/0', {}) - .get('swVersionInfo', {}) - .get('oneUiVersion', '')) - type_registry = for_device(one_ui_version) if one_ui_version else None - if type_registry is not None: - self.log.info("device type: %s (oneUiVersion=%r)", - type_registry.name, one_ui_version) - bound = discover(resources, type_registry.capabilities, - type_registry.pattern_capabilities, - log=lambda m: self.log.debug("%s", m)) - else: - if one_ui_version: - self.log.warning( - "unknown device type oneUiVersion=%r; using common caps", - one_ui_version) - bound = discover(resources, CAPABILITIES, - log=lambda m: self.log.debug("%s", m)) - ident = read_identity(sess, self._serial) - model = ident.model or self.app.klass.title() or 'Samsung Appliance' - device_name = self.app.device_name or ident.name or model - self.descriptor = build_runtime_descriptor( - bound, - topic_prefix=self.app.topic_prefix, - ha_prefix=self.shared.HA_DISCOVERY_PREFIX, - device_name=device_name, - model=model, - name=(self.app.klass or 'appliance'), - default_port=self.port) - self.cache.set_on_observation(self.descriptor.on_observation) - self.cmd_handlers = self.descriptor.command_handlers() - diag_payloads = _bridge_diagnostic_discovery( - self.app.topic_prefix, self.shared.HA_DISCOVERY_PREFIX, - device_name, model=model) - self._entity_payloads = self.descriptor.discovery_payloads + diag_payloads - for topic, payload in self._entity_payloads: - self.mqtt.publish(topic, payload, qos=1, retain=True) - self._discovered = True - self.log.info("discovered %d entities", len(self.descriptor.discovery_payloads)) - - def _debug_dump_links(self, sess): - for href, rep in sorted(self.cache.links.items()): - if href == '/mode/vs/0': - short = {k: v for k, v in rep.items() - if k not in ( - 'x.com.samsung.da.modeSpec', - 'x.com.samsung.da.supportedModes', - )} - self.log.info("LINK %s = %r", href, short) - else: - self.log.info("LINK %s = %r", href, rep) - try: - code, pl = sess.get(['oic', 'res'], timeout=10.0) - if code == 0x45 and pl: - self.log.info("OIC_RES = %r", cbor2.loads(pl)) - else: - self.log.info("oic/res → %s", fmt_code(code)) - except Exception as e: - self.log.warning("oic/res get: %s", e) - - def _poll_loop(self, sess: DtlsCoapSession) -> None: - """Batch-poll /device/0 at active or idle interval until stop is set.""" - while not self.stop.is_set(): - rd = self.descriptor - if rd is not None: - snap = self.cache.snapshot() - active = rd.is_active is not None and rd.is_active(snap) - interval = rd.active_interval_s if active else rd.idle_interval_s - else: - interval = 30.0 - if self.stop.wait(interval): - break - try: - code, payload = sess.get(SEED_PATH, timeout=15.0) - except Exception as e: - self.log.warning("poll error: %s", e) - self._poll_error_count += 1 - continue - self._poll_count += 1 - if code == 0x45 and payload: - try: - body = cbor2.loads(payload) - except Exception as e: - self.log.warning("poll cbor decode: %s", e) - self._poll_error_count += 1 - continue - if isinstance(body, list): - new_resources = _parse_device0_batch(body) - for href, rep in new_resources.items(): - self.cache.apply_rep(href, rep, source='poll') - elif code != 0x45: - self.log.warning("poll %s: %s", SEED_PATH, fmt_code(code)) - self._poll_error_count += 1 - - def _on_reachable(self) -> None: - self.set_availability(True) - self.reassert_availability() - - def _on_unreachable(self) -> None: - self.set_availability(False) - - # ---- MQTT publishing -------------------------------------------- - - def maybe_publish_state(self, force=False): - if self.descriptor is None: - return - if not force and not self._publish_gate: - return - snap = self.cache.snapshot() - sensors = self.descriptor.flatten(snap) - project = self.descriptor.project - if project is not None: - sensors = project(self.cache.descriptor_state, sensors) - if not force and sensors == self.last_state_pub: - return - if DEBUG_BRIDGE and self.last_state_pub is not None: - diffs = {k: (self.last_state_pub.get(k), v) - for k, v in sensors.items() - if self.last_state_pub.get(k) != v} - diffs.update({k: (self.last_state_pub.get(k), None) - for k in self.last_state_pub - if k not in sensors}) - if diffs: - self.log.info("sensor diff: %s", - {k: f"{a!r} → {b!r}" for k, (a, b) in diffs.items()}) - self.last_state_pub = sensors - self.mqtt.publish(self.state_topic, - json.dumps(sensors).encode(), - qos=1, retain=True) - field = self.descriptor.remote_available_field - if field is not None: - self.publish_remote_available(sensors.get(field)) - cycle_field = self.descriptor.cycle_active_field - if cycle_field is not None: - self.publish_cycle_active(sensors.get(cycle_field)) - if not force: - log_fn = self.descriptor.log_state_change - extra = log_fn(sensors) if log_fn is not None else '' - self.log.info("state changed [%s] (%s notif#%d)", - self._last_change_source or '?', - extra or 'descriptor-no-log', self.notif_count) - - def publish_remote_available(self, remote_on, force=False): - value = 'online' if remote_on else 'offline' - if not force and value == self.last_remote_pub: - return - self.last_remote_pub = value - try: - self.mqtt.publish(self.remote_topic, value, qos=1, retain=True) - self.log.info("remote_available → %s", value) - except Exception as e: - self.log.warning("remote_available publish: %s", e) - - def publish_cycle_active(self, cycle_on, force=False): - value = 'online' if cycle_on else 'offline' - if not force and value == self.last_cycle_pub: - return - self.last_cycle_pub = value - try: - self.mqtt.publish(self.cycle_topic, value, qos=1, retain=True) - self.log.info("cycle_active → %s", value) - except Exception as e: - self.log.warning("cycle_active publish: %s", e) - - def reassert_availability(self): - if self.session is None or self.last_state_pub is None: - return - self.set_availability(True) - if self._discovered: - for topic, payload in self._entity_payloads: - try: - self.mqtt.publish(topic, payload, qos=1, retain=True) - except Exception as e: - self.log.warning("re-assert discovery: %s", e) - if self.descriptor is not None: - field = self.descriptor.remote_available_field - if field is not None: - self.publish_remote_available( - self.last_state_pub.get(field), force=True) - cycle_field = self.descriptor.cycle_active_field - if cycle_field is not None: - self.publish_cycle_active( - self.last_state_pub.get(cycle_field), force=True) - - def set_availability(self, online): - value = 'online' if online else 'offline' - if value == self.last_avail_pub: - return - self.last_avail_pub = value - try: - self.mqtt.publish(self.avail_topic, value, qos=1, retain=True) - except Exception as e: - self.log.warning("avail publish: %s", e) - if not online and self.descriptor is not None: - if self.descriptor.remote_available_field is not None: - self.last_remote_pub = None - try: - self.mqtt.publish(self.remote_topic, 'offline', - qos=1, retain=True) - except Exception: - pass - if self.descriptor.cycle_active_field is not None: - self.last_cycle_pub = None - try: - self.mqtt.publish(self.cycle_topic, 'offline', - qos=1, retain=True) - except Exception: - pass - - # ---- MQTT command handling -------------------------------------- - - def handle_command(self, topic, payload): - if not topic.startswith(self.cmd_topic_prefix): - return - suffix = topic[len(self.cmd_topic_prefix) - len('cmd/'):] - handler = self.cmd_handlers.get(suffix) - if handler is None: - self.log.warning("unknown command topic: %s", topic) - return - # Handler gets a links snapshot so its read-modify-write sees a - # consistent view across the multi-field operation. - result = handler(payload, self.cache.snapshot()) - if result is None: - self.log.warning("rejected command %s payload=%r", - topic, payload) - return - path_segs, body = result - sess = self.session - if sess is None: - self.log.warning("command %s: no DTLS session", topic) - return - href = '/' + '/'.join(path_segs) - try: - code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0) - except Exception as e: - self.log.warning("command %s POST failed: %s", topic, e) - return - self.log.info("command %s payload=%r → %s", - suffix, payload, fmt_code(code)) - if code >> 5 == 2: - # Optimistic local merge so HA sees the write reflected - # immediately. No Block2 fetchback — that triggers Samsung's - # 3-second revert (project_fetchback_revert_root_cause.md). - # The batch poll will reconcile on its next tick. - self.cache.apply_optimistic(href, body) - - def publish_health(self): - now = time.time() - ka = self.keepalive - - # Per-window deltas for the log summary AND the health topic. - poll = self._poll_count - poll_err = self._poll_error_count - ping_fail = ka.ping_fail_count if ka else 0 - d_poll = poll - self._win_prev_poll - d_err = poll_err - self._win_prev_poll_err - d_ping_fail = ping_fail - self._win_prev_ping_fail - self._win_prev_poll = poll - self._win_prev_poll_err = poll_err - self._win_prev_ping_fail = ping_fail - window_polls_ok = max(0, d_poll - d_err) - - h = { - 'mode': 'poll', - 'device_class': self.app.klass or 'appliance', - 'serial': self._serial, - 'connect_count': self.connect_count, - 'error_count': self.error_count, - 'notif_count': self.notif_count, - 'poll_count': poll, - 'poll_error_count': poll_err, - 'poll_window_ok': window_polls_ok, - 'poll_window_errors': d_err, - 'ping_count': ka.ping_count if ka else 0, - 'ping_fail_count': ping_fail, - 'reachable': ka.reachable if ka else None, - 'last_change_source': self._last_change_source, - 'last_change_age_s': (round(now - self.last_change_ts, 1) - if self.last_change_ts else None), - 'last_seed_age_s': (round(now - self.last_seed_ts, 1) - if self.last_seed_ts else None), - 'session_age_s': (round(now - self.session_started_ts, 1) - if self.session_started_ts else None), - 'uptime_seconds': round(now - self.started_ts, 0), - } - stalest = self.cache.stalest() - if stalest is not None: - h['stalest_href'] = stalest[0] - h['stalest_age_s'] = round(stalest[1], 1) - try: - self.mqtt.publish(self.health_topic, json.dumps(h).encode(), - qos=0, retain=True) - except Exception as e: - self.log.warning("health publish: %s", e) - - if d_poll > 0 or d_err > 0 or d_ping_fail > 0: - self.log.info( - "poll-window: %d ok, %d err, %d ping-fail (%ds)", - window_polls_ok, d_err, d_ping_fail, - self.shared.HEALTH_INTERVAL_S) - - # ---- top-level loop --------------------------------------------- - - def run_forever(self): - backoff = 1.0 - while not self.stop.is_set(): - try: - self.session_once() - backoff = 1.0 - except Exception as e: - self.error_count += 1 - self.log.warning("session error: %s", e) - sess = self.session - self.session = None - if sess is not None: - try: sess.close() - except Exception: pass - self.set_availability(False) - self.session_started_ts = None - if self.stop.is_set(): - break - wait = min(backoff, 30.0) - self.log.info("reconnect in %.0fs", wait) - if self.stop.wait(wait): - break - backoff = min(backoff * 2, 30.0) diff --git a/samsung_appliance/coap_dtls.py b/samsung_appliance/coap_dtls.py index de64044..732cafb 100644 --- a/samsung_appliance/coap_dtls.py +++ b/samsung_appliance/coap_dtls.py @@ -26,7 +26,8 @@ import time from OpenSSL import SSL -from .logger import logger +import logging +logger = logging.getLogger(__name__) # Diagnostic logging — when DEBUG_BRIDGE=1 in env, the bridge dumps diff --git a/samsung_appliance/config.py b/samsung_appliance/config.py deleted file mode 100644 index af9e876..0000000 --- a/samsung_appliance/config.py +++ /dev/null @@ -1,115 +0,0 @@ -"""Configuration — env-var driven. - -Two flavours: - * SharedConfig — MQTT broker, cert paths, HA prefix, timers. One per - process. - * ApplianceConfig — one per appliance the bridge is supervising. Keys - come from `APPLIANCE__*` env vars (1-indexed). The list of - appliances is `APPLIANCE_COUNT` entries long. - -.env in cwd hydrates os.environ at import time; docker-compose env wins -over file contents.""" -import os -from dataclasses import dataclass -from pathlib import Path -from typing import Optional - - -def _load_env_file(filename='.env'): - """If a .env-style file is present in cwd, hydrate os.environ from - it. Existing env wins so docker-compose `environment:` overrides - file contents.""" - env_path = Path.cwd() / filename - if not env_path.exists(): - return - for line in env_path.read_text().splitlines(): - line = line.strip() - if not line or line.startswith('#') or '=' not in line: - continue - k, v = line.split('=', 1) - k = k.strip(); v = v.strip() - if k and k not in os.environ: - os.environ[k] = v - - -_load_env_file('.env') - - -def _resolve_cert(env_key, basename): - """Cert lookup: explicit env > /config/ (Docker mount) > - ./certs/ (bare-metal dev).""" - if os.getenv(env_key): - return Path(os.environ[env_key]) - docker_path = Path('/config') / basename - if docker_path.exists(): - return docker_path - return Path.cwd() / 'certs' / basename - - -@dataclass(frozen=True) -class SharedConfig: - """Process-wide config (MQTT, cert paths, intervals).""" - CERT_PATH: Path - KEY_PATH: Path - MQTT_BROKER: Optional[str] - MQTT_PORT: int - MQTT_USER: Optional[str] - MQTT_PASS: Optional[str] - HA_DISCOVERY_PREFIX: str - HEALTH_INTERVAL_S: int - PING_INTERVAL_S: int - - @classmethod - def from_env(cls) -> 'SharedConfig': - return cls( - CERT_PATH=_resolve_cert('CERT_PATH', 'client_fullchain.pem'), - KEY_PATH=_resolve_cert('KEY_PATH', 'client.key'), - MQTT_BROKER=os.getenv('MQTT_BROKER'), - MQTT_PORT=int(os.getenv('MQTT_PORT', '1883')), - MQTT_USER=os.getenv('MQTT_USER') or None, - MQTT_PASS=os.getenv('MQTT_PASS') or None, - HA_DISCOVERY_PREFIX=os.getenv('HA_DISCOVERY_PREFIX', - 'homeassistant'), - HEALTH_INTERVAL_S=int(os.getenv('HEALTH_INTERVAL_S', '60')), - PING_INTERVAL_S=int(os.getenv('PING_INTERVAL_S', '25')), - ) - - -@dataclass(frozen=True) -class ApplianceConfig: - """Per-appliance runtime config. - - Sourced from `APPLIANCE__*` env vars (1-indexed). `index` is - just a stable identifier for logs; it does not appear in MQTT - topics or HA discovery (those are keyed off `topic_prefix`).""" - index: int - klass: str # 'dryer', 'oven', … — empty string if unknown - ip: str - ocf_port: Optional[int] # None → 49154 (default DTLS-CoAP port) - topic_prefix: str - device_name: str - - @classmethod - def from_env(cls, index: int) -> 'ApplianceConfig': - prefix = f'APPLIANCE_{index}_' - klass = os.getenv(prefix + 'CLASS') or '' - ip = os.getenv(prefix + 'IP') - if not ip: - raise ValueError(f"{prefix}IP not set") - port_env = os.getenv(prefix + 'OCF_PORT') - port = int(port_env) if port_env else None - topic = os.getenv(prefix + 'TOPIC') or (f'samsung_{klass}' if klass else 'samsung_appliance') - name = os.getenv(prefix + 'NAME') or (f'Samsung {klass.title()}' if klass else 'Samsung Appliance') - return cls( - index=index, klass=klass, ip=ip, ocf_port=port, - topic_prefix=topic, device_name=name, - ) - - -def load_appliances() -> list[ApplianceConfig]: - """Read APPLIANCE_COUNT and build the appliance list. At least one - is required.""" - count = int(os.getenv('APPLIANCE_COUNT', '1')) - if count < 1: - raise ValueError("APPLIANCE_COUNT must be >= 1") - return [ApplianceConfig.from_env(i + 1) for i in range(count)] diff --git a/samsung_appliance/keepalive.py b/samsung_appliance/keepalive.py deleted file mode 100644 index 9e47ac3..0000000 --- a/samsung_appliance/keepalive.py +++ /dev/null @@ -1,110 +0,0 @@ -"""DTLS-layer liveness via CoAP empty-CON ping + poll-success watchdog. - -Each interval_s the task: - 1. Sends a CoAP ping. This is fire-and-forget — Samsung's RT-OCF - doesn't reliably reply with an RST, so the send itself is the - keepalive (it tickles Samsung's observer state). The send only - fails if the underlying socket is gone, in which case the failure - counts toward fail_threshold. - 2. Calls liveness_fn() if provided. This is the real half-open - detection: PollScheduler exposes last_success_ts, and the bridge - wraps it as "did we get a 2.05 in the last 60s". If not, count - the tick as a failure even though the ping send succeeded. - -After fail_threshold consecutive failures, fires on_unreachable. First -success after a fail streak fires on_reachable. Bridge wires these to -MQTT availability. -""" -from __future__ import annotations - -import threading -from typing import Callable, Optional - -from .coap_dtls import DtlsCoapSession - - -class KeepaliveTask: - - def __init__(self, - session: DtlsCoapSession, - interval_s: float = 25.0, - fail_threshold: int = 3, - on_reachable: Optional[Callable[[], None]] = None, - on_unreachable: Optional[Callable[[], None]] = None, - logger=None, - liveness_fn: Optional[Callable[[], bool]] = None): - self.session = session - self.interval_s = interval_s - self.fail_threshold = fail_threshold - self.on_reachable = on_reachable - self.on_unreachable = on_unreachable - self.log = logger - self.liveness_fn = liveness_fn - - self._fail_streak = 0 - self._reachable = True - self._ping_count = 0 - self._ping_fail_count = 0 - - @property - def ping_count(self) -> int: - return self._ping_count - - @property - def ping_fail_count(self) -> int: - return self._ping_fail_count - - @property - def reachable(self) -> bool: - return self._reachable - - def run_forever(self, stop: threading.Event) -> None: - while not stop.wait(self.interval_s): - self._tick() - - def _tick(self) -> None: - ok = False - try: - self.session.ping() - ok = True - except Exception as e: - if self.log: self.log.warning("ping: %s", e) - # Real half-open detection: ping sends can succeed against a - # silently-wedged peer, but polls won't. If the scheduler - # hasn't recorded a 2.05 inside the liveness window, treat - # this tick as a failure even though the ping itself went out. - if ok and self.liveness_fn is not None: - try: - alive = bool(self.liveness_fn()) - except Exception as e: - if self.log: self.log.warning("liveness_fn: %s", e) - alive = True - if not alive: - if self.log: - self.log.warning("liveness: no successful poll " - "in the liveness window") - ok = False - self._ping_count += 1 - if ok: - if not self._reachable: - if self.log: - self.log.info("ping recovered after %d fails", - self._fail_streak) - self._reachable = True - if self.on_reachable is not None: - try: self.on_reachable() - except Exception as e: - if self.log: self.log.warning("on_reachable: %s", e) - self._fail_streak = 0 - return - self._ping_fail_count += 1 - self._fail_streak += 1 - if self._reachable and self._fail_streak >= self.fail_threshold: - if self.log: - self.log.warning("device unreachable after %d ping failures", - self._fail_streak) - self._reachable = False - if self.on_unreachable is not None: - try: self.on_unreachable() - except Exception as e: - if self.log: self.log.warning("on_unreachable: %s", e) diff --git a/samsung_appliance/logger.py b/samsung_appliance/logger.py deleted file mode 100644 index 55d6439..0000000 --- a/samsung_appliance/logger.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Stdout logging — works correctly under Docker's PYTHONUNBUFFERED=1. - -Two logger families share the root handler: - * `samsung_appliance` (and its children) for module-level lines — - DTLS warnings, generic startup chatter, MQTT plumbing. - * `.` for per-appliance bridge lines — e.g. - `dryer.`. Each PushBridge gets its own such logger - via bridge_logger() once the seed reveals the appliance's serial. - -Both trees propagate to the root logger, which is the one with the -StreamHandler, so the same format applies everywhere. - -WARNING lines render yellow, ERROR red, when colour is enabled. Set -`NO_COLOR=1` in the environment to fall back to plain text (e.g. when -writing logs to a file).""" -import logging -import os -import sys - - -_LEVEL_COLOURS = { - logging.WARNING: '\033[33m', # yellow - logging.ERROR: '\033[31m', # red - logging.CRITICAL: '\033[1;31m', # bold red -} -_RESET = '\033[0m' - - -def _colour_enabled() -> bool: - return 'NO_COLOR' not in os.environ - - -class _LevelColourFormatter(logging.Formatter): - """Wraps the rendered line in an ANSI colour for WARNING+ levels. - INFO/DEBUG pass through unchanged so the bulk of the log stays - readable and a yellow line draws the eye.""" - - def __init__(self, fmt: str, datefmt: str, use_colour: bool): - super().__init__(fmt=fmt, datefmt=datefmt) - self.use_colour = use_colour - - def format(self, record: logging.LogRecord) -> str: - line = super().format(record) - if not self.use_colour: - return line - colour = _LEVEL_COLOURS.get(record.levelno) - if colour is None: - return line - return f"{colour}{line}{_RESET}" - - -_handler = logging.StreamHandler(sys.stdout) -_handler.setFormatter(_LevelColourFormatter( - fmt='%(asctime)s %(levelname)-5s %(name)-26s %(message)s', - datefmt='%H:%M:%S', - use_colour=_colour_enabled(), -)) - -logging.basicConfig(level=logging.INFO, handlers=[_handler], force=True) - -logger = logging.getLogger("samsung_appliance") - - -def bridge_logger(klass: str, serial: str | None = None) -> logging.Logger: - """Return a top-level logger tagged with the appliance class and, - once known, the serial. Pre-seed callers pass serial=None and get - e.g. `dryer`; post-seed callers pass the serial and get e.g. - `dryer.`.""" - name = f"{klass}.{serial}" if serial else klass - return logging.getLogger(name) diff --git a/samsung_appliance/poll_scheduler.py b/samsung_appliance/poll_scheduler.py deleted file mode 100644 index a18ab15..0000000 --- a/samsung_appliance/poll_scheduler.py +++ /dev/null @@ -1,303 +0,0 @@ -"""Tiered adaptive polling against a DtlsCoapSession. - -Tiers are descriptor-declared (hot/warm/cold + sweep). Per tick: -each tier whose deadline has passed polls all its paths sequentially -on the shared session, writing into the StateCache. The sweep tier -issues one Block2 GET of /device/0 and uses index_links to fan its -result into many href reps. - -Adaptive cadence: when descriptor.is_active(cache.links) returns True -and tier.active_interval_s is set, that tier uses the tighter cadence. -If the previous health window saw `active_throttle_threshold` timeouts, -the throttle drops back to idle cadence even when active=True — the -RT-OCF stack wedges under load and stacking poll attempts only makes -it worse. - -Per-tier timeouts: tier.timeout_s overrides the scheduler default. Hot -tiers want a tight ceiling (e.g. 2s) so one wedged path can't eat -several poll cycles. - -Cooldown on timeout: when a path times out, it's deferred for ~3 of -its tier's intervals (clamped 5–60s) via the same _defer_until mechanism -used for write_in_progress. Breaks the cluster-cascade where the next -tier tick fires immediately after an 8s wedge and reattempts the same -stalled path. - -Post-write defer: bridge calls write_in_progress(href) before POSTing -a write; the scheduler skips that href for settle_s to avoid Samsung's -fetchback-revert bug. -""" -from __future__ import annotations - -import threading -import time -from dataclasses import dataclass -from typing import Callable, Optional, TYPE_CHECKING - -import cbor2 - -from .coap_dtls import DtlsCoapSession, fmt_code - -if TYPE_CHECKING: - from .state_cache import StateCache - - -@dataclass(frozen=True) -class PollTier: - name: str - interval_s: float - paths: tuple[tuple[str, ...], ...] - active_interval_s: Optional[float] = None - is_sweep: bool = False - # Per-tier CoAP request timeout. Falls back to PollScheduler.timeout_s - # when None. Hot tiers want a tight ceiling (e.g. 2s) so one wedged - # path can't eat several poll cycles; sweep tiers tolerate longer. - timeout_s: Optional[float] = None - - -class PollScheduler: - - def __init__(self, - session: DtlsCoapSession, - cache: 'StateCache', - tiers: list[PollTier], - sweep_index_fn: Callable[[object], dict[str, dict]], - is_active_fn: Optional[Callable[[dict[str, dict]], bool]] = None, - logger=None, - timeout_s: float = 8.0, - active_throttle_timeout_threshold: int = 3): - self.session = session - self.cache = cache - self.tiers = tiers - self.sweep_index = sweep_index_fn - self.is_active_fn = is_active_fn - self.log = logger - self.timeout_s = timeout_s - self.active_throttle_threshold = active_throttle_timeout_threshold - - now = time.monotonic() - self._next_due: dict[str, float] = {t.name: now for t in tiers} - self._defer_until: dict[str, float] = {} - self._defer_lock = threading.Lock() - self._poll_count = 0 - self._poll_error_count = 0 - self._last_active: Optional[bool] = None - self._last_throttled: bool = False - # Real-liveness signal for KeepaliveTask: updated on every 2.05 - # we receive from the wire. Initialized to "now" so the first - # keepalive tick after start doesn't fire a false unreachable. - self._last_success_ts: float = now - - # Per-window tail-latency tracking. Bridge consumes-and-resets - # these via take_window_stats() once per HEALTH_INTERVAL_S. - # _window_max_rtt_ms tracks SUCCESSFUL polls only — timeouts go - # into _window_timeout_count so the dashboard sees the real tail - # instead of an 8000ms wall. - self._stats_lock = threading.Lock() - self._window_max_rtt_ms = 0.0 - self._window_slow_count = 0 - self._window_timeout_count = 0 - # Snapshot of the previous window's timeout count, used by the - # active-window throttle to back off when polls are wedging. - self._last_window_timeouts = 0 - self.slow_threshold_ms = 1000.0 - - def write_in_progress(self, href: str, settle_s: float = 4.0) -> None: - with self._defer_lock: - self._defer_until[href] = time.monotonic() + settle_s - - def run_forever(self, stop: threading.Event) -> None: - while not stop.is_set(): - self._run_due_tiers() - sleep_for = max(0.05, min(1.0, self._earliest_deadline() - time.monotonic())) - if stop.wait(sleep_for): - return - - @property - def poll_count(self) -> int: - return self._poll_count - - @property - def poll_error_count(self) -> int: - return self._poll_error_count - - @property - def last_success_ts(self) -> float: - """Monotonic timestamp of the most recent 2.05 response from any - tier. KeepaliveTask uses this as its half-open-detection signal: - if no 2.05 has landed in `liveness_window_s`, the session is - wedged regardless of whether ping sends succeed.""" - return self._last_success_ts - - def take_window_stats(self) -> tuple[float, int, int]: - """Return (max RTT ms over successful polls, slow-poll count, - timeout count) seen since the last call, and reset all three. - Slow threshold is `self.slow_threshold_ms`. The timeout count - is snapshotted into `_last_window_timeouts` for the throttle.""" - with self._stats_lock: - out = (self._window_max_rtt_ms, - self._window_slow_count, - self._window_timeout_count) - self._last_window_timeouts = self._window_timeout_count - self._window_max_rtt_ms = 0.0 - self._window_slow_count = 0 - self._window_timeout_count = 0 - return out - - def _record_rtt(self, rtt_ms: float, *, timed_out: bool = False) -> None: - with self._stats_lock: - if timed_out: - self._window_timeout_count += 1 - return - if rtt_ms > self._window_max_rtt_ms: - self._window_max_rtt_ms = rtt_ms - if rtt_ms >= self.slow_threshold_ms: - self._window_slow_count += 1 - - def _earliest_deadline(self) -> float: - return min(self._next_due.values()) - - def _run_due_tiers(self) -> None: - now = time.monotonic() - active = False - if self.is_active_fn is not None: - try: - active = bool(self.is_active_fn(self.cache.snapshot())) - except Exception as e: - if self.log: self.log.warning("is_active: %s", e) - if active != self._last_active: - if self.log and self._last_active is not None: - self.log.info("active=%s", active) - self._last_active = active - # Active-window throttle: if the previous health window saw a - # cluster of timeouts (RT-OCF wedging under load), drop back to - # idle cadence even when is_active=True. Lets the device breathe - # instead of stacking poll attempts on a stalled responder. - with self._stats_lock: - recent_to = self._last_window_timeouts - throttled = (active - and recent_to >= self.active_throttle_threshold) - if throttled != self._last_throttled: - if self.log: - if throttled: - self.log.warning( - "active-throttle ON (%d timeouts last window) — " - "using idle cadence", recent_to) - else: - self.log.info("active-throttle OFF") - self._last_throttled = throttled - effective_active = active and not throttled - for tier in self.tiers: - if self._next_due[tier.name] > now: - continue - interval = (tier.active_interval_s - if (effective_active and tier.active_interval_s is not None) - else tier.interval_s) - self._next_due[tier.name] = now + interval - try: - if tier.is_sweep: - self._do_sweep(tier) - else: - self._do_tier(tier) - except Exception as e: - self._poll_error_count += 1 - if self.log: self.log.warning("tier %s: %s", tier.name, e) - - def _tier_timeout(self, tier: PollTier) -> float: - return tier.timeout_s if tier.timeout_s is not None else self.timeout_s - - def _cooldown_for(self, tier: PollTier) -> float: - # On timeout, defer the wedged href for ~3 cycles (clamped to a - # 5–60s band) so repeated tier ticks don't stack attempts on a - # stalled responder. Breaks the cluster-cascade we see in the - # Poll Max RTT chart during heavy device use. - return max(5.0, min(60.0, tier.interval_s * 3.0)) - - def _set_cooldown(self, href: str, cooldown_s: float) -> None: - with self._defer_lock: - self._defer_until[href] = time.monotonic() + cooldown_s - - def _do_tier(self, tier: PollTier) -> None: - timeout = self._tier_timeout(tier) - cooldown = self._cooldown_for(tier) - for path in tier.paths: - href = '/' + '/'.join(path) - with self._defer_lock: - if self._defer_until.get(href, 0) > time.monotonic(): - continue - self._poll_count += 1 - t0 = time.monotonic() - try: - code, body = self.session.get(list(path), timeout=timeout) - except TimeoutError: - self._poll_error_count += 1 - self._record_rtt(0.0, timed_out=True) - self._set_cooldown(href, cooldown) - if self.log: - self.log.warning("poll %s timeout (cooldown %.0fs)", - href, cooldown) - return - except Exception as e: - self._poll_error_count += 1 - self._record_rtt((time.monotonic() - t0) * 1000.0) - if self.log: self.log.warning("poll %s: %s", href, e) - return - self._record_rtt((time.monotonic() - t0) * 1000.0) - if code != 0x45 or not body: - self._poll_error_count += 1 - if self.log: self.log.warning("poll %s -> %s", href, fmt_code(code)) - continue - self._last_success_ts = time.monotonic() - try: - rep = cbor2.loads(body) - except Exception as e: - self._poll_error_count += 1 - if self.log: self.log.warning("poll %s cbor: %s", href, e) - continue - if isinstance(rep, dict): - self.cache.apply_rep(href, rep, source='poll') - - def _do_sweep(self, tier: PollTier) -> None: - timeout = self._tier_timeout(tier) - cooldown = self._cooldown_for(tier) - path = list(tier.paths[0]) - href = '/' + '/'.join(path) - t0 = time.monotonic() - self._poll_count += 1 - try: - code, body = self.session.get(path, timeout=timeout) - except TimeoutError: - self._poll_error_count += 1 - self._record_rtt(0.0, timed_out=True) - self._set_cooldown(href, cooldown) - if self.log: - self.log.warning("sweep %s timeout (cooldown %.0fs)", - path, cooldown) - return - except Exception as e: - self._poll_error_count += 1 - self._record_rtt((time.monotonic() - t0) * 1000.0) - if self.log: self.log.warning("sweep %s: %s", path, e) - return - self._record_rtt((time.monotonic() - t0) * 1000.0) - if code != 0x45 or not body: - self._poll_error_count += 1 - if self.log: self.log.warning("sweep -> %s", fmt_code(code)) - return - self._last_success_ts = time.monotonic() - try: - tree = cbor2.loads(body) - except Exception as e: - self._poll_error_count += 1 - if self.log: self.log.warning("sweep cbor: %s", e) - return - indexed = self.sweep_index(tree) - for href, rep in indexed.items(): - with self._defer_lock: - if self._defer_until.get(href, 0) > time.monotonic(): - continue - self.cache.apply_rep(href, rep, source='sweep') - if self.log: - elapsed_ms = (time.monotonic() - t0) * 1000.0 - self.log.info("sweep complete (%d links, %.0fms)", - len(indexed), elapsed_ms) diff --git a/samsung_appliance/registry/__init__.py b/samsung_appliance/registry/__init__.py index f1045d1..3cb5880 100644 --- a/samsung_appliance/registry/__init__.py +++ b/samsung_appliance/registry/__init__.py @@ -1,6 +1,5 @@ -from .adapter import RuntimeDescriptor, build_runtime_descriptor +from .adapter import flatten, is_active, _key from .discovery import BoundEntity, discover from .registry import CAPABILITIES -__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', - 'RuntimeDescriptor', 'build_runtime_descriptor'] +__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', 'is_active', '_key'] diff --git a/samsung_appliance/registry/adapter.py b/samsung_appliance/registry/adapter.py index e966850..ee5e35f 100644 --- a/samsung_appliance/registry/adapter.py +++ b/samsung_appliance/registry/adapter.py @@ -1,222 +1,33 @@ -"""Adapter: a discovered BoundEntity list -> a RuntimeDescriptor the bridge -consumes. Replaces the per-appliance flatten/build_discovery/command_handlers. -""" +"""Adapter: BoundEntity list → flat state dict and command dispatch.""" from __future__ import annotations -import json -from dataclasses import dataclass, field -from typing import Callable, Optional +from typing import Any from .discovery import BoundEntity -from .entities import ( - BinarySensorDesc, ButtonDesc, NumberDesc, PLATFORM_OF, SelectDesc, - SensorDesc, SwitchDesc, -) - -_TIER_CADENCE = {'hot': (1.0, 0.5), 'warm': (15.0, None), 'cold': (300.0, None)} - -# Keys produced by oven capabilities that require cycle-active gating. -# When any of these appear in the produced set, the runtime descriptor -# exposes cycle_active_field='cycle_active' so the bridge can gate writes. -_CYCLE_GATED_KEYS = {'oven_setpoint', 'oven_mode', 'cook_time'} - - -def _segs(href: str) -> list[str]: - return [s for s in href.strip('/').split('/') if s] def _key(b: BoundEntity) -> str: return f"{b.key_override or b.desc.key}{b.instance}" -def _encode(cfg: dict) -> bytes: - return json.dumps(cfg).encode() - - -@dataclass -class RuntimeDescriptor: - name: str - seed_path: list[str] - active_interval_s: float = 5.0 - idle_interval_s: float = 30.0 - is_active: Optional[Callable[[dict], bool]] = None - flatten: Optional[Callable[[dict], dict]] = None - discovery_payloads: list[tuple[str, bytes]] = field(default_factory=list) - command_handlers: Optional[Callable[[], dict]] = None - project: Optional[Callable[[dict, dict], dict]] = None - on_observation: Optional[Callable[[dict, str, dict], None]] = None - remote_available_field: Optional[str] = None - cycle_active_field: Optional[str] = None - log_state_change: Optional[Callable[[dict], str]] = None - - -def _make_flatten(bound: list[BoundEntity]) -> Callable[[dict], dict]: - reads = [b for b in bound if b.desc.field] # buttons have no field - - def flatten(links: dict) -> dict: - out: dict = {} - for b in reads: - rep = links.get(b.href) or {} - if b.desc.exists_fn is not None and not b.desc.exists_fn(rep): - continue - out[_key(b)] = b.desc.value_fn(rep.get(b.desc.field)) - return out - - return flatten - - -def _make_is_active(bound): - caps = {b.href: b.capability for b in bound if b.capability.active_when} - if not caps: - return None - - def is_active(links: dict) -> bool: - return any(cap.active_when(links.get(href) or {}) - for href, cap in caps.items()) - - return is_active - - -def _make_on_observation(bound): - caps = {b.href: b.capability for b in bound if b.capability.on_observation} - if not caps: - return None - - def on_observation(state: dict, href: str, rep: dict) -> None: - cap = caps.get(href) - if cap is not None: - cap.on_observation(state, rep) - - return on_observation - - -def _make_project(bound): - projs = [] - seen = set() +def flatten(bound: list[BoundEntity], resources: dict) -> dict[str, Any]: + """Map bound entities to their current scalar values.""" + out: dict[str, Any] = {} for b in bound: - if b.capability.project and b.capability.href not in seen: - seen.add(b.capability.href) - projs.append(b.capability.project) - if not projs: - return None - - def project(state: dict, sensors: dict) -> dict: - for p in projs: - sensors = p(state, sensors) - return sensors - - return project - - -def _make_command_handlers(bound): - writers = [b for b in bound if getattr(b.desc, 'write_fn', None)] - - def command_handlers() -> dict: - out = {} - for b in writers: - href, fn = b.href, b.desc.write_fn - - def handler(payload, links, _fn=fn, _href=href): - rep = (links or {}).get(_href) or {} - return _fn(payload, rep, _href) - - out[f"cmd/{_key(b)}"] = handler - return out - - return command_handlers - - -# --- MQTT discovery config emitters (per platform) ------------------------ -def _avail(topic_prefix): - return [{'topic': f"{topic_prefix}/availability", - 'payload_available': 'online', 'payload_not_available': 'offline'}] - - -def _device(topic_prefix, device_name, model): - return {'identifiers': [topic_prefix], 'name': device_name, - 'manufacturer': 'Samsung', 'model': model} - - -def _discovery_payloads(bound, topic_prefix, ha_prefix, device_name, model): - state_topic = f"{topic_prefix}/state" - dev = _device(topic_prefix, device_name, model) - avail = _avail(topic_prefix) - out = [] - for b in bound: - d = b.desc - key = _key(b) - platform = PLATFORM_OF[type(d)] - uid = f"{topic_prefix}_{key}" - cfg = {'name': d.name or key, 'unique_id': uid, 'object_id': uid, - 'device': dev, 'availability': avail} - if d.icon: - cfg['icon'] = d.icon - if d.entity_category: - cfg['entity_category'] = d.entity_category - if isinstance(d, SensorDesc): - cfg['state_topic'] = state_topic - cfg['value_template'] = f"{{{{ value_json.{key} }}}}" - for k, attr in (('unit_of_measurement', 'unit'), - ('device_class', 'device_class'), - ('state_class', 'state_class')): - if getattr(d, attr): - cfg[k] = getattr(d, attr) - elif isinstance(d, BinarySensorDesc): - cfg['state_topic'] = state_topic - cfg['value_template'] = ( - f"{{{{ 'ON' if value_json.{key} else 'OFF' }}}}") - cfg['payload_on'] = 'ON' - cfg['payload_off'] = 'OFF' - if d.device_class: - cfg['device_class'] = d.device_class - elif isinstance(d, SelectDesc): - cfg['state_topic'] = state_topic - cfg['value_template'] = f"{{{{ value_json.{key} }}}}" - cfg['command_topic'] = f"{topic_prefix}/cmd/{key}" - cfg['options'] = list(d.options) if d.options else [] - elif isinstance(d, SwitchDesc): - cfg['state_topic'] = state_topic - cfg['value_template'] = f"{{% if value_json.{key} %}}On{{% else %}}Off{{% endif %}}" - cfg['state_on'] = 'On' - cfg['state_off'] = 'Off' - cfg['payload_on'] = 'On' - cfg['payload_off'] = 'Off' - cfg['command_topic'] = f"{topic_prefix}/cmd/{key}" - elif isinstance(d, ButtonDesc): - cfg['command_topic'] = f"{topic_prefix}/cmd/{key}" - cfg['payload_press'] = d.payload - elif isinstance(d, NumberDesc): - cfg['state_topic'] = state_topic - cfg['value_template'] = f"{{{{ value_json.{key} }}}}" - cfg['command_topic'] = f"{topic_prefix}/cmd/{key}" - for k, attr in (('unit_of_measurement', 'unit'), - ('device_class', 'device_class'), - ('min', 'native_min'), ('max', 'native_max'), - ('step', 'step')): - if getattr(d, attr) is not None: - cfg[k] = getattr(d, attr) - out.append((f"{ha_prefix}/{platform}/{topic_prefix}/{key}/config", - _encode(cfg))) + if not b.desc.field: + continue + rep = resources.get(b.href) or {} + if b.desc.exists_fn is not None and not b.desc.exists_fn(rep): + continue + out[_key(b)] = b.desc.value_fn(rep.get(b.desc.field)) return out -def build_runtime_descriptor(bound, *, topic_prefix, ha_prefix, device_name, - model, name, default_port=None, - active_interval_s=5.0, idle_interval_s=30.0) -> RuntimeDescriptor: - produced = {_key(b) for b in bound} - return RuntimeDescriptor( - name=name, - seed_path=['device', '0'], - active_interval_s=active_interval_s, - idle_interval_s=idle_interval_s, - is_active=_make_is_active(bound), - flatten=_make_flatten(bound), - discovery_payloads=_discovery_payloads( - bound, topic_prefix, ha_prefix, device_name, model), - command_handlers=_make_command_handlers(bound), - project=_make_project(bound), - on_observation=_make_on_observation(bound), - remote_available_field='remote_control' if 'remote_control' in produced else None, - cycle_active_field='cycle_active' if produced & _CYCLE_GATED_KEYS else None, - log_state_change=None, - ) +def is_active(bound: list[BoundEntity], resources: dict) -> bool: + """Return True if any capability's active_when fires.""" + for b in bound: + if b.capability.active_when: + rep = resources.get(b.href) or {} + if b.capability.active_when(rep): + return True + return False diff --git a/samsung_appliance/registry/by_type/dryer.py b/samsung_appliance/registry/by_type/dryer.py index 77f02f3..ebf0548 100644 --- a/samsung_appliance/registry/by_type/dryer.py +++ b/samsung_appliance/registry/by_type/dryer.py @@ -1,7 +1,7 @@ """Dryer device registry. -Note: dryer devices use port 49155 (not 49154). Pass default_port=49155 -when calling build_runtime_descriptor() for this registry. +Note: dryer devices use port 49155 (not 49154). The config flow handles +this automatically via port probing. """ from ..capabilities import common, dryer, fridge, laundry, operational from ._base import DeviceRegistry, _build diff --git a/samsung_appliance/registry/capabilities/operational.py b/samsung_appliance/registry/capabilities/operational.py index 828711d..3cbb58e 100644 --- a/samsung_appliance/registry/capabilities/operational.py +++ b/samsung_appliance/registry/capabilities/operational.py @@ -97,17 +97,17 @@ OPERATIONAL_STATE = Capability( name='Delay start time', icon='mdi:timer-pause'), ButtonDesc(key='start', field='', name='Start cycle', payload='Run', icon='mdi:play', - write_fn=lambda p, rep: ( + 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', icon='mdi:pause', - write_fn=lambda p, rep: ( + 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: ( + write_fn=lambda p, rep, href=None: ( ['operational', 'state', 'vs', '0'], {'x.com.samsung.da.state': p})), ), diff --git a/samsung_appliance/sensors.py b/samsung_appliance/sensors.py deleted file mode 100644 index 5344492..0000000 --- a/samsung_appliance/sensors.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Shared sensor helpers. - -Appliance-specific flattening lives in samsung_appliance/appliances/*.py; -this module only carries utilities that every descriptor uses (currently -the /device/0 link-dict indexer). -""" - - -def index_links(device0_body): - """Turn the /device/0 CBOR list-of-{href, rep} into a dict keyed - by href. The first list entry is the device-level rep itself and - isn't useful here, so skip it.""" - out = {} - if not isinstance(device0_body, list): - return out - for entry in device0_body[1:]: - if isinstance(entry, dict) and 'href' in entry: - out[entry['href']] = entry.get('rep') or {} - return out diff --git a/samsung_appliance/state_cache.py b/samsung_appliance/state_cache.py deleted file mode 100644 index f44eef7..0000000 --- a/samsung_appliance/state_cache.py +++ /dev/null @@ -1,78 +0,0 @@ -"""Single source of truth for one appliance's state. - -All writers (OBSERVE notify, poll, seed, optimistic) call apply_rep(). -A registered on_change callback fires after any apply that mutated the -cache, which the bridge wires to its MQTT publish gate. -""" -from __future__ import annotations - -import threading -import time -from typing import Callable, Optional - - -class StateCache: - - def __init__(self): - self.links: dict[str, dict] = {} - self.last_updated: dict[str, float] = {} - self.source: dict[str, str] = {} - self.descriptor_state: dict = {} - self._on_change: Optional[Callable[[bool, str], None]] = None - self._on_observation: Optional[Callable[[dict, str, dict], None]] = None - self._lock = threading.RLock() - - def set_on_change(self, cb: Callable[[bool, str], None]) -> None: - self._on_change = cb - - def set_on_observation(self, hook: Optional[Callable[[dict, str, dict], None]]) -> None: - self._on_observation = hook - - def apply_rep(self, href: str, rep: dict, source: str) -> bool: - if not isinstance(rep, dict): - return False - with self._lock: - prior = self.links.get(href) - changed = prior != rep - self.links[href] = rep - self.last_updated[href] = time.time() - self.source[href] = source - hook = self._on_observation - if hook is not None: - try: - hook(self.descriptor_state, href, rep) - except Exception: - pass - if self._on_change is not None: - try: - self._on_change(changed, source) - except Exception: - pass - return changed - - def apply_optimistic(self, href: str, body: dict) -> bool: - if not isinstance(body, dict): - return False - with self._lock: - merged = dict(self.links.get(href) or {}) - merged.update(body) - return self.apply_rep(href, merged, source='optimistic') - - def get(self, href: str) -> Optional[dict]: - with self._lock: - return self.links.get(href) - - def snapshot(self) -> dict[str, dict]: - with self._lock: - return dict(self.links) - - def freshness_s(self, href: str) -> Optional[float]: - ts = self.last_updated.get(href) - return None if ts is None else (time.time() - ts) - - def stalest(self) -> Optional[tuple[str, float]]: - with self._lock: - if not self.last_updated: - return None - href = min(self.last_updated, key=self.last_updated.get) - return href, time.time() - self.last_updated[href] diff --git a/tests/conftest.py b/tests/conftest.py index a95a808..f14036f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,8 +7,8 @@ FIXTURES = Path(__file__).resolve().parent / 'fixtures' def _resources_from_dump(dump: dict) -> dict[str, dict]: - from samsung_appliance.bridge import _parse_device0_batch - return _parse_device0_batch(dump['device0']) + from samsung_appliance.batch import parse_device0_batch + return parse_device0_batch(dump['device0']) def _load_device(name: str) -> dict[str, dict]: diff --git a/tests/test_adapter.py b/tests/test_adapter.py deleted file mode 100644 index ceff09e..0000000 --- a/tests/test_adapter.py +++ /dev/null @@ -1,88 +0,0 @@ -import json - -from samsung_appliance.registry.adapter import build_runtime_descriptor -from samsung_appliance.registry.capabilities import common -from samsung_appliance.registry.discovery import discover - - -def _bound(resources): - reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER)} - return discover(resources, reg) - - -def _bound_with_power(resources): - reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER, common.POWER)} - return discover(resources, reg) - - -def test_flatten_applies_value_fn(dishwasher_resources): - rd = build_runtime_descriptor( - _bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant', - device_name='Dishwasher', model='M', name='dishwasher', default_port=49154) - flat = rd.flatten(dishwasher_resources) - assert flat['power_watts'] is not None - assert isinstance(flat['child_lock'], bool) - - -def test_discovery_payloads_have_unique_ids(dishwasher_resources): - rd = build_runtime_descriptor( - _bound(dishwasher_resources), topic_prefix='samsung_dishwasher', - ha_prefix='homeassistant', device_name='Dishwasher', model='M', - name='dishwasher', default_port=49154) - topics = [t for t, _ in rd.discovery_payloads] - # one config topic per produced sensor/binary_sensor entity - assert any('/binary_sensor/samsung_dishwasher/child_lock/config' in t for t in topics) - assert any('/sensor/samsung_dishwasher/power_watts/config' in t for t in topics) - # payloads are JSON bytes carrying unique_id - cfg = json.loads(next(p for t, p in rd.discovery_payloads if 'child_lock' in t)) - assert cfg['unique_id'] == 'samsung_dishwasher_child_lock' - - -def test_switch_discovery_config_has_correct_payloads(dishwasher_resources): - rd = build_runtime_descriptor( - _bound_with_power(dishwasher_resources), topic_prefix='samsung_dishwasher', - ha_prefix='homeassistant', device_name='Dishwasher', model='M', - name='dishwasher', default_port=49154) - cfg = json.loads(next(p for t, p in rd.discovery_payloads if 'power_switch' in t)) - assert cfg['payload_on'] == 'On' - assert cfg['payload_off'] == 'Off' - assert cfg['state_on'] == 'On' - assert cfg['state_off'] == 'Off' - assert 'On' in cfg['value_template'] - assert 'Off' in cfg['value_template'] - - -def test_bound_entity_key_override_takes_precedence(): - from samsung_appliance.registry.adapter import _key - from samsung_appliance.registry.discovery import BoundEntity - from samsung_appliance.registry.entities import SensorDesc - - cap = common.ENERGY_METER - desc = SensorDesc( - key='power_watts', - field='x.com.samsung.da.power', - unit='W', - device_class='power', - state_class='measurement', - value_fn=lambda x: x, - ) - - # Without key_override, should use desc.key - b1 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='') - assert _key(b1) == 'power_watts' - - # With key_override, should use key_override - b2 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='', key_override='foo') - assert _key(b2) == 'foo' - - # key_override with instance suffix should preserve instance suffix - b3 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='_1', key_override='bar') - assert _key(b3) == 'bar_1' - - -def test_runtime_descriptor_has_active_and_idle_intervals(dishwasher_resources): - rd = build_runtime_descriptor( - _bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant', - device_name='Dishwasher', model='M', name='dishwasher', default_port=49154) - assert rd.active_interval_s == 5.0 - assert rd.idle_interval_s == 30.0 diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py deleted file mode 100644 index 49fce9b..0000000 --- a/tests/test_end_to_end.py +++ /dev/null @@ -1,24 +0,0 @@ -import pytest - -from tests.conftest import _load_resources -from samsung_appliance.registry import (CAPABILITIES, discover, - build_runtime_descriptor) - - -@pytest.mark.parametrize('name,ip', [ - ('dishwasher', '10.0.0.129'), - ('refrigerator', '10.0.0.254'), -]) -def test_full_pipeline_produces_entities_and_intervals(name, ip): - resources = _load_resources(ip) - bound = discover(resources, CAPABILITIES) - rd = build_runtime_descriptor( - bound, topic_prefix=f'samsung_{name}', ha_prefix='homeassistant', - device_name=name.title(), model='M', name=name, default_port=49154) - assert rd.discovery_payloads, "no entities discovered" - assert rd.active_interval_s > 0, "active_interval_s must be positive" - assert rd.idle_interval_s > rd.active_interval_s, ( - "idle_interval_s must be greater than active_interval_s") - # flatten runs cleanly over the real rep set - flat = rd.flatten(resources) - assert flat diff --git a/tests/test_end_to_end_v2.py b/tests/test_end_to_end_v2.py index f705c2d..15fb373 100644 --- a/tests/test_end_to_end_v2.py +++ b/tests/test_end_to_end_v2.py @@ -2,7 +2,7 @@ import pytest from tests.conftest import _load_device from samsung_appliance.registry.by_type import for_device, _type_key from samsung_appliance.registry.discovery import discover -from samsung_appliance.registry.adapter import build_runtime_descriptor +from samsung_appliance.registry.adapter import flatten, is_active @pytest.mark.parametrize('name,expected_type_key', [ @@ -22,11 +22,7 @@ def test_full_pipeline_v2(name, expected_type_key): bound = discover(resources, reg.capabilities, reg.pattern_capabilities) assert bound - rd = build_runtime_descriptor(bound, topic_prefix=f'samsung_{name}', - ha_prefix='homeassistant', device_name=name.title(), - model='M', name=name, default_port=49154) - assert rd.discovery_payloads - flat = rd.flatten(resources) - assert flat - assert rd.active_interval_s > 0 - assert rd.idle_interval_s > rd.active_interval_s + state = flatten(bound, resources) + assert state + # is_active returns bool; just verify it runs without error + is_active(bound, resources) diff --git a/tests/test_golden_regression.py b/tests/test_golden_regression.py index 6a2d6c7..a5a1412 100644 --- a/tests/test_golden_regression.py +++ b/tests/test_golden_regression.py @@ -6,10 +6,10 @@ import pytest GOLDEN = Path(__file__).parent / 'fixtures' / 'golden' -def _new_state_and_uids(name, resources): +def _new_state_keys(name, resources): from samsung_appliance.registry.by_type import for_device from samsung_appliance.registry.discovery import discover - from samsung_appliance.registry.adapter import build_runtime_descriptor + from samsung_appliance.registry.adapter import flatten otn = resources.get('/otninformation/vs/0', {}) one_ui = otn.get('swVersionInfo', {}).get('oneUiVersion', '') reg = for_device(one_ui) if one_ui else None @@ -19,12 +19,8 @@ def _new_state_and_uids(name, resources): else: caps, pats = reg.capabilities, reg.pattern_capabilities bound = discover(resources, caps, pats) - rd = build_runtime_descriptor( - bound, topic_prefix=f'samsung_{name}', ha_prefix='homeassistant', - device_name=name.title(), model='M', name=name, default_port=49154) - state = rd.flatten(resources) - uids = [json.loads(p)['unique_id'] for _, p in rd.discovery_payloads] - return sorted(state.keys()), sorted(uids) + state = flatten(bound, resources) + return sorted(state.keys()) @pytest.mark.parametrize('name,ip', [ @@ -35,17 +31,12 @@ def test_registry_reproduces_golden_state_keys(name, ip, request): from tests.conftest import _load_resources resources = _load_resources(ip) golden = json.loads((GOLDEN / f'{name}.json').read_text()) - state_keys, uids = _new_state_and_uids(name, resources) + state_keys = _new_state_keys(name, 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))}" ) - assert set(uids) == set(golden['discovery_unique_ids']), ( - f"unique_ids mismatch:\n" - f" extra: {sorted(set(uids) - set(golden['discovery_unique_ids']))}\n" - f" missing: {sorted(set(golden['discovery_unique_ids']) - set(uids))}" - ) def test_resources_from_batch_preferred_over_flat(): diff --git a/tests/test_oven_capabilities.py b/tests/test_oven_capabilities.py index 85628d3..df4b737 100644 --- a/tests/test_oven_capabilities.py +++ b/tests/test_oven_capabilities.py @@ -1,6 +1,5 @@ -"""Unit tests for oven-family capabilities and the cycle-active adapter wiring.""" +"""Unit tests for oven-family capabilities.""" from samsung_appliance.registry.capabilities import oven -from samsung_appliance.registry.adapter import build_runtime_descriptor from samsung_appliance.registry.discovery import discover @@ -151,36 +150,3 @@ def test_cook_time_rejects_out_of_range(): assert desc.write_fn(-1, {}) is None assert desc.write_fn(1440, {}) is None - -# --------------------------------------------------------------------------- -# Adapter cycle_active_field wiring -# --------------------------------------------------------------------------- - -def test_adapter_sets_cycle_active_field_for_oven(): - """build_runtime_descriptor must set cycle_active_field when oven keys present.""" - reg = {c.href: [c] for c in (oven.OVEN_OPERATIONAL_STATE, oven.OVEN_SETPOINT)} - resources = { - '/operational/state/vs/0': {'x.com.samsung.da.state': 'Run'}, - '/temperatures/vs/0': {'x.com.samsung.da.items': [ - {'x.com.samsung.da.current': '180', 'x.com.samsung.da.desired': '200'} - ]}, - } - bound = discover(resources, reg) - rd = build_runtime_descriptor( - bound, topic_prefix='t', ha_prefix='homeassistant', - device_name='Oven', model='M', name='oven', default_port=49154) - assert rd.cycle_active_field == 'cycle_active' - flat = rd.flatten(resources) - assert flat['cycle_active'] is True # 'Run' -> active - - -def test_adapter_no_cycle_active_without_oven_keys(): - """Non-oven appliances must not get cycle_active_field.""" - from samsung_appliance.registry.capabilities import common - reg = {c.href: [c] for c in (common.KIDS_LOCK,)} - resources = {'/kidslock/vs/0': {'x.com.samsung.da.kidsLock': 'Ready'}} - bound = discover(resources, reg) - rd = build_runtime_descriptor( - bound, topic_prefix='t', ha_prefix='homeassistant', - device_name='Dishwasher', model='M', name='dishwasher', default_port=49154) - assert rd.cycle_active_field is None diff --git a/tests/test_state_cache.py b/tests/test_state_cache.py deleted file mode 100644 index b5b9560..0000000 --- a/tests/test_state_cache.py +++ /dev/null @@ -1,15 +0,0 @@ -from samsung_appliance.state_cache import StateCache - - -def test_state_cache_constructs_without_descriptor(): - c = StateCache() - assert c.links == {} - assert c.descriptor_state == {} - - -def test_on_observation_hook_fires(): - c = StateCache() - seen = [] - c.set_on_observation(lambda st, href, rep: seen.append((href, rep))) - c.apply_rep('/x/vs/0', {'a': 1}, source='seed') - assert seen == [('/x/vs/0', {'a': 1})]