refactor: move samsung_appliance/ into custom_components/localthings/ocf/

Makes the integration fully self-contained for HACS distribution.
Users installing via HACS get everything in one directory; no external
samsung_appliance package needed.

All component imports updated to relative (.ocf.*); all test imports
updated to custom_components.localthings.ocf.*. ocf_root_ca.pem moves
with the package and resolves correctly via Path(__file__).parent.

custom_components/localthings/binary_sensor.py

@@ -6,7 +6,7 @@ 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 .ocf.registry.entities import BinarySensorDesc
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator

custom_components/localthings/config_flow.py

@@ -26,8 +26,8 @@ def _probe_and_validate(host: str, cert_pem: str, key_pem: str) -> dict:
     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
+    from .ocf.coap_dtls import DtlsCoapSession
+    from .ocf.batch import parse_device0_batch
 
     # Write PEM strings to temporary files — DtlsCoapSession requires
     # cert_path / key_path rather than in-memory PEM strings.

custom_components/localthings/coordinator.py

@@ -14,13 +14,13 @@ 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 .ocf.batch import parse_device0_batch
+from .ocf.coap_dtls import DtlsCoapSession
+from .ocf.registry.by_type import for_device
+from .ocf.registry.discovery import discover, BoundEntity
+from .ocf.registry import CAPABILITIES
+from .ocf.registry.adapter import flatten, is_active
+from .ocf.registry.identity import read_identity, DeviceIdentity
 
 from .const import (
     DOMAIN, CONF_HOST, CONF_PORT, CONF_CERT_PEM, CONF_KEY_PEM,

custom_components/localthings/entity.py

@@ -4,8 +4,8 @@ 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 .ocf.registry.adapter import _key
+from .ocf.registry.discovery import BoundEntity
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator

custom_components/localthings/number.py

@@ -6,7 +6,7 @@ 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 .ocf.registry.entities import NumberDesc
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator

custom_components/localthings/ocf/__init__.py

@@ -0,0 +1,5 @@
+"""Samsung appliance local-API library (custom_components/localthings/ocf).
+
+CoAP/DTLS session management, OCF batch resource parsing, and a
+capability registry for discovering and reading Samsung appliance state."""
+__version__ = "2.0.0"

custom_components/localthings/ocf/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

custom_components/localthings/ocf/coap_dtls.py

@@ -0,0 +1,636 @@
+"""CoAP-over-DTLS client for Samsung RT-OCF appliances (RFC 7252 + 6347).
+
+Replaces the TLS-over-TCP transport used in the original dryer bridge.
+Both the oven (UDP/49154) and the dryer (UDP/49155) speak CoAP-over-DTLS
+with the ECDHE-ECDSA-AES128-GCM-SHA256 cipher and ab0b0ac4 client cert.
+
+Wire-level details that matter (from local-tools/oven-findings.md §17):
+  * DTLS ciphertext MTU must be 1200; otherwise OpenSSL fragments the
+    client cert across two datagrams and TizenRT drops the second.
+  * Samsung's RT-OCF uses ACK+separate-CON for the larger responses.
+    The reader MUST correlate by (token, mid) — not arrival order —
+    or interleaved one-shot / OBSERVE traffic mis-attributes.
+  * Multi-block GET requires the SAME CoAP token across every block
+    of the response ("token-stable Block2"). Fresh-token-per-block
+    is silently dropped by the server.
+
+Reader thread owns the UDP socket. Callers issue get()/post() and block
+on a per-token Event the reader signals. OBSERVE notifications are
+delivered via the on_notification callback.
+"""
+import os
+import socket
+import struct
+import threading
+import time
+from pathlib import Path
+
+from OpenSSL import SSL
+
+_OCF_ROOT_CA = str(Path(__file__).parent / 'ocf_root_ca.pem')
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+# Diagnostic logging — when DEBUG_BRIDGE=1 in env, the bridge dumps
+# every received CoAP frame, every /operational/state/vs/0 + /oven/vs/0
+# + /power/vs/0 + /mode/vs/0-options rep change, the full link tree at
+# seed time, and the /oic/res directory. Useful for reverse-engineering
+# new resources and field semantics; otherwise quiet.
+DEBUG_BRIDGE = os.environ.get('DEBUG_BRIDGE') == '1'
+
+
+# CoAP option numbers (RFC 7252 + 7641 + 7959)
+URI_PATH       = 11
+URI_QUERY      = 15
+OBSERVE        =  6
+CONTENT_FORMAT = 12
+ACCEPT         = 17
+BLOCK2         = 23
+SIZE2          = 28
+
+# CoAP message types
+TYPE_CON = 0
+TYPE_NON = 1
+TYPE_ACK = 2
+TYPE_RST = 3
+
+# CoAP method codes
+METHOD_GET  = 0x01
+METHOD_POST = 0x02
+
+# CoAP content-format value for application/cbor
+CF_CBOR = b'\x3c'
+
+# OBSERVE option values (RFC 7641 §2)
+OBSERVE_REGISTER   = b''           # register / refresh
+OBSERVE_DEREGISTER = bytes([1])    # deregister
+
+# Block2 SZX=6 → 1024-byte blocks. The largest size Samsung's RT-OCF
+# will honour and the only one the probes have validated end-to-end.
+BLOCK_SZX = 6
+
+
+def _vlen(v):
+    """Variable-length integer encoder used in option deltas + lengths."""
+    if v < 13:    return v, b''
+    if v < 269:   return 13, bytes([v - 13])
+    return 14, struct.pack('>H', v - 269)
+
+
+def encode_options(opts):
+    """Encode a list of (option_number, value_bytes) tuples."""
+    out = b''
+    prev = 0
+    for n, val in sorted(opts, key=lambda x: x[0]):
+        d, dx = _vlen(n - prev)
+        l, lx = _vlen(len(val))
+        out += bytes([(d << 4) | l]) + dx + lx + val
+        prev = n
+    return out
+
+
+def parse_coap(data):
+    """Decode a CoAP datagram. Returns (mtype, code, mid, token,
+    options, payload). options is a list of (num, value_bytes)."""
+    mt = (data[0] >> 4) & 0x03
+    tkl = data[0] & 0x0F
+    code = data[1]
+    mid = int.from_bytes(data[2:4], 'big')
+    tok = data[4:4 + tkl]
+    i = 4 + tkl
+    opts = []
+    prev = 0
+    payload = b''
+    while i < len(data):
+        b = data[i]
+        if b == 0xFF:
+            payload = data[i + 1:]
+            break
+        d_nib, l_nib = b >> 4, b & 0x0F
+        i += 1
+        if d_nib == 13:
+            delta = 13 + data[i]; i += 1
+        elif d_nib == 14:
+            delta = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2
+        elif d_nib == 15:
+            raise ValueError("reserved option delta nibble 15")
+        else:
+            delta = d_nib
+        if l_nib == 13:
+            length = 13 + data[i]; i += 1
+        elif l_nib == 14:
+            length = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2
+        elif l_nib == 15:
+            raise ValueError("reserved option length nibble 15")
+        else:
+            length = l_nib
+        num = prev + delta
+        opts.append((num, data[i:i + length]))
+        i += length
+        prev = num
+    return mt, code, mid, tok, opts, payload
+
+
+def build_coap(mtype, code, mid, token, options, payload=b''):
+    """Build a CoAP datagram. mtype: CON/NON/ACK/RST. token: bytes (may
+    be empty for ACK). options: list of (num, value_bytes)."""
+    tkl = len(token)
+    hdr = bytes([(1 << 6) | (mtype << 4) | tkl, code,
+                 (mid >> 8) & 0xFF, mid & 0xFF])
+    body = hdr + token + encode_options(options)
+    if payload:
+        body += b'\xFF' + payload
+    return body
+
+
+def block_value(num, more, szx):
+    """Encode a CoAP Block-N option value."""
+    v = (num << 4) | ((more & 1) << 3) | (szx & 7)
+    if v <= 0xFF:    return bytes([v])
+    if v <= 0xFFFF:  return struct.pack('>H', v)
+    return struct.pack('>I', v)[1:]
+
+
+def fmt_code(c):
+    """0x45 → '2.05', 0x84 → '4.04'. Used in log lines."""
+    return f"{c >> 5}.{c & 0x1F:02d}"
+
+
+def _split_dtls(buf):
+    """Split a UDP datagram that contains one-or-more DTLS records.
+    OpenSSL sometimes hands the BIO multiple records back-to-back; we
+    must send each as its own UDP datagram or TizenRT drops them."""
+    o, out = 0, []
+    while o + 13 <= len(buf):
+        L = int.from_bytes(buf[o + 11:o + 13], 'big')
+        end = o + 13 + L
+        if end > len(buf):
+            break
+        out.append(buf[o:end])
+        o = end
+    return out
+
+
+class DtlsCoapSession:
+    """Single sustained DTLS-CoAP session.
+
+    Caller drives lifecycle:
+        sess = DtlsCoapSession(host, port, cert, key)
+        sess.connect()
+        sess.start_reader()
+        sess.subscribe([...], on_notification=cb)   # OBSERVE
+        code, body = sess.get(['device', '0'])      # Block2 fetch
+        code, _    = sess.post(['mode','vs','0'], cbor)
+        sess.close()
+    """
+
+    HANDSHAKE_TIMEOUT_S = 12.0
+    READER_RECV_TIMEOUT_S = 1.0  # short so stop_event propagates quickly
+    MAX_BLOCKS = 32              # safety bound for Block2 fetches
+
+    def __init__(self, host, port, cert_path, key_path,
+                 on_notification=None, mtu=1200):
+        self.host = host
+        self.port = port
+        self.cert_path = str(cert_path)
+        self.key_path  = str(key_path)
+        self.on_notification = on_notification  # fn(href, payload_bytes)
+        self.mtu = mtu
+
+        self.sock = None
+        self.conn = None
+        self.dest = None
+
+        self._send_lock = threading.Lock()
+        # Randomize MID and token counter starting points so reconnects
+        # don't reuse identifiers from previous sessions — Samsung's
+        # RT-OCF appears to remember observer state across DTLS
+        # sessions, and re-registering with a token it still thinks is
+        # active is silently no-ops.
+        self._mid = int.from_bytes(os.urandom(2), 'big')
+        self._tok_counter = int.from_bytes(os.urandom(4), 'big')
+        # OBSERVE tokens are 1-byte (Samsung silently drops TKL>1
+        # OBSERVE registrations). Pick a random starting byte in the
+        # 0x40..0xff range so each session uses fresh values.
+        self._observe_tok_counter = 0x40 + (os.urandom(1)[0] & 0xBF)
+        # token (bytes) → (Event, container_dict)
+        self._pending = {}
+        # token (bytes) → href (str)
+        self._observe_tokens = {}
+
+        self._stop = threading.Event()
+        self._reader_thread = None
+
+    # ---- lifecycle ---------------------------------------------------
+
+    def connect(self):
+        """DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises
+        ConnectionError / TimeoutError on failure."""
+        ctx = SSL.Context(SSL.DTLS_METHOD)
+        ctx.load_verify_locations(_OCF_ROOT_CA)
+        ctx.set_verify(SSL.VERIFY_PEER, lambda conn, cert, err, depth, ok: ok)
+        ctx.set_cipher_list(b'ECDHE-ECDSA-AES128-GCM-SHA256')
+        ctx.use_certificate_chain_file(self.cert_path)
+        ctx.use_privatekey_file(self.key_path)
+        ctx.check_privatekey()
+
+        conn = SSL.Connection(ctx, None)
+        conn.set_connect_state()
+        conn.set_ciphertext_mtu(self.mtu)
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.settimeout(2.0)
+        dest = (self.host, self.port)
+
+        t0 = time.time()
+        while time.time() - t0 < self.HANDSHAKE_TIMEOUT_S:
+            try:
+                conn.do_handshake()
+                break
+            except SSL.WantReadError:
+                pass
+            except SSL.Error as e:
+                sock.close()
+                raise ConnectionError(f"DTLS handshake error: {e}") from e
+            try:
+                o = conn.bio_read(65535)
+                if o:
+                    for r in _split_dtls(o):
+                        sock.sendto(r, dest)
+            except SSL.WantReadError:
+                pass
+            try:
+                d, _ = sock.recvfrom(65535)
+                if d:
+                    conn.bio_write(d)
+            except socket.timeout:
+                pass
+            time.sleep(0.05)
+        else:
+            sock.close()
+            raise TimeoutError(
+                f"DTLS handshake timeout to {self.host}:{self.port}")
+
+        self.sock = sock
+        self.conn = conn
+        self.dest = dest
+        self._stop.clear()
+
+    def start_reader(self):
+        """Spawn the reader thread. Must be called after connect()."""
+        if self.sock is None:
+            raise RuntimeError("connect() before start_reader()")
+        t = threading.Thread(target=self._reader_loop,
+                             daemon=True, name='dtls-reader')
+        t.start()
+        self._reader_thread = t
+
+    def join(self):
+        """Block until the reader thread exits (i.e. socket dies)."""
+        if self._reader_thread is not None:
+            self._reader_thread.join()
+
+    def _send_observe_dereg(self, tok, path_segs):
+        """Send a single OBSERVE deregister GET (Observe option = 1)
+        on the existing token. Best-effort — caller swallows errors."""
+        if self.conn is None:
+            return
+        mid = self._next_mid()
+        opts = [(URI_PATH, s.encode()) for s in path_segs]
+        opts.append((OBSERVE, OBSERVE_DEREGISTER))
+        opts.append((ACCEPT, CF_CBOR))
+        self._send_dgram(
+            build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+
+    def close(self):
+        """Tear down session. Sends best-effort OBSERVE deregisters
+        first so Samsung's RT-OCF cleans up its observer table —
+        without this, the per-cert observer state survives DTLS close
+        and a quick reconnect with the same tokens silently no-ops."""
+        # Send dereg for every active observation while the conn is
+        # still healthy. Tiny sleep lets the records reach the wire
+        # before we shut DTLS down.
+        if self.conn is not None and self._observe_tokens:
+            for tok, href in list(self._observe_tokens.items()):
+                segs = [s for s in href.split('/') if s]
+                try:
+                    self._send_observe_dereg(tok, segs)
+                except Exception as e:
+                    logger.warning("dereg %s: %s", href, e)
+            time.sleep(0.1)
+
+        self._stop.set()
+        if self.conn is not None:
+            try:
+                self.conn.shutdown()
+            except Exception:
+                pass
+        if self.sock is not None:
+            try:
+                self.sock.close()
+            except Exception:
+                pass
+        for tok, (ev, container) in list(self._pending.items()):
+            container.setdefault('err', 'socket closed')
+            ev.set()
+        self._pending.clear()
+        self._observe_tokens.clear()
+        self.sock = None
+        self.conn = None
+
+    # ---- send / receive plumbing -------------------------------------
+
+    def _next_mid(self):
+        self._mid = (self._mid + 1) & 0xFFFF
+        return self._mid
+
+    def _next_tok(self):
+        self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF
+        # 4-byte tokens — fits within tkl=8 cap with headroom and
+        # avoids collisions across long-running OBSERVE subscriptions.
+        return self._tok_counter.to_bytes(4, 'big')
+
+    def _next_observe_tok(self):
+        # Single-byte tokens for OBSERVE registrations. Samsung
+        # RT-OCF accepts these but silently drops TKL=4 OBSERVE
+        # registrations. Counter is randomly seeded per session so
+        # reconnects don't collide with stale observer state Samsung
+        # may still be holding from the previous run.
+        self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF
+        # Avoid 0x00 — some CoAP stacks treat an all-zero token as
+        # equivalent to "no token" / empty (TKL=0).
+        if self._observe_tok_counter == 0:
+            self._observe_tok_counter = 1
+        return bytes([self._observe_tok_counter])
+
+    def _send_dgram(self, datagram):
+        """Send a CoAP datagram. Holds the send lock for the
+        BIO-drain so two writers can't interleave records."""
+        with self._send_lock:
+            if self.conn is None:
+                raise ConnectionError("DTLS session closed")
+            self.conn.send(datagram)
+            try:
+                while True:
+                    o = self.conn.bio_read(65535)
+                    if not o:
+                        break
+                    for r in _split_dtls(o):
+                        self.sock.sendto(r, self.dest)
+            except SSL.WantReadError:
+                pass
+
+    def _reader_loop(self):
+        """Pump UDP socket → DTLS BIO → CoAP parser. Demuxes to pending
+        / observe handlers. Exits on socket error or stop event."""
+        sock = self.sock
+        conn = self.conn
+        sock.settimeout(self.READER_RECV_TIMEOUT_S)
+        try:
+            while not self._stop.is_set():
+                try:
+                    d, _ = sock.recvfrom(65535)
+                except socket.timeout:
+                    continue
+                except (OSError, ValueError):
+                    return
+                if not d:
+                    continue
+                # pyOpenSSL's SSL.Connection is not thread-safe — the
+                # same SSL object must not be touched by multiple
+                # threads concurrently. Drain decrypted records into a
+                # local list under _send_lock so the reader never races
+                # a sender's conn.send()/bio_read(). Dispatch happens
+                # AFTER releasing the lock because _dispatch_coap may
+                # call _send_dgram (auto-ACK for CON frames), which
+                # re-acquires the lock — holding it across dispatch
+                # would deadlock.
+                packets = []
+                exit_reader = False
+                with self._send_lock:
+                    try:
+                        conn.bio_write(d)
+                    except SSL.Error as e:
+                        logger.warning("DTLS bio_write: %s", e)
+                        return
+                    while True:
+                        try:
+                            pl = conn.recv(65535)
+                        except SSL.WantReadError:
+                            break
+                        except SSL.ZeroReturnError:
+                            logger.info("DTLS peer closed connection")
+                            exit_reader = True
+                            break
+                        except SSL.Error as e:
+                            logger.warning("DTLS recv: %s", e)
+                            exit_reader = True
+                            break
+                        if not pl:
+                            break
+                        packets.append(pl)
+                for pl in packets:
+                    try:
+                        self._dispatch_coap(pl)
+                    except Exception as e:
+                        logger.warning("dispatch: %s", e)
+                if exit_reader:
+                    return
+        finally:
+            # Make sure pending waiters don't hang if the reader dies.
+            for tok, (ev, container) in list(self._pending.items()):
+                container.setdefault('err', 'reader exited')
+                ev.set()
+
+    def _dispatch_coap(self, datagram):
+        try:
+            mt, code, mid, tok, ropts, payload = parse_coap(datagram)
+        except Exception as e:
+            logger.debug("malformed CoAP: %s", e)
+            return
+
+        if DEBUG_BRIDGE:
+            kind = ['CON', 'NON', 'ACK', 'RST'][mt]
+            logger.info("rx %s code=%s mid=%04x tok=%s opts=%d pl=%d",
+                        kind, fmt_code(code), mid, tok.hex() or '-',
+                        len(ropts), len(payload))
+
+        # ACK back any CON from the device to suppress retransmits.
+        # RFC 7252 §4.2 — ACK is a bare frame (token len 0, code 0).
+        if mt == TYPE_CON:
+            try:
+                self._send_dgram(build_coap(TYPE_ACK, 0, mid, b'', []))
+            except Exception as e:
+                logger.warning("ACK send: %s", e)
+
+        # Empty ACK with no options & no payload = "separate response
+        # coming" — used by Samsung's RT-OCF for the larger reads. Stop
+        # the retransmit timer on the client side and wait for the CON.
+        if mt == TYPE_ACK and code == 0 and not payload and not ropts:
+            return
+
+        # Pending one-shot? Resolve and return.
+        rec = self._pending.get(tok)
+        if rec is not None:
+            ev, container = rec
+            container['code']    = code
+            container['mtype']   = mt
+            container['mid']     = mid
+            container['options'] = ropts
+            container['payload'] = payload
+            ev.set()
+            return
+
+        # OBSERVE notification?
+        href = self._observe_tokens.get(tok)
+        if href is not None:
+            if code != 0x45:
+                logger.warning("observe %s: non-2.05 %s",
+                               href, fmt_code(code))
+                return
+            cb = self.on_notification
+            if cb is not None:
+                try:
+                    cb(href, payload)
+                except Exception as e:
+                    logger.warning("notification callback %s: %s",
+                                   href, e)
+            return
+
+        # Stale token (post-reconnect or unknown) — drop quietly.
+
+    # ---- request primitives ------------------------------------------
+
+    def get(self, path_segs, query=(), timeout=10.0):
+        """Token-stable Block2 GET. Returns (code, payload_bytes).
+
+        Reuses one CoAP token across every block of a multi-block
+        response — Samsung's server keys per-transfer state on the
+        token, and dropping a fresh token on block 1+ silently drops
+        the request."""
+        if self.conn is None:
+            raise ConnectionError("DTLS session closed")
+        tok = self._next_tok()
+        blob = b''
+        num = 0
+        last_code = None
+        last_opts = []
+        deadline = time.time() + timeout
+        while True:
+            ev = threading.Event()
+            container = {}
+            self._pending[tok] = (ev, container)
+            try:
+                mid = self._next_mid()
+                opts = [(URI_PATH, s.encode()) for s in path_segs]
+                for q in query:
+                    opts.append((URI_QUERY, q.encode()))
+                opts.append((ACCEPT, CF_CBOR))
+                if num > 0:
+                    opts.append((BLOCK2, block_value(num, 0, BLOCK_SZX)))
+                self._send_dgram(
+                    build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+                wait = max(0.1, deadline - time.time())
+                if not ev.wait(wait):
+                    raise TimeoutError(
+                        f"GET /{'/'.join(path_segs)} block {num} timeout")
+                if 'err' in container:
+                    raise ConnectionError(container['err'])
+            finally:
+                self._pending.pop(tok, None)
+
+            code = container['code']
+            payload = container['payload']
+            ropts = container['options']
+            last_code = code
+            last_opts = ropts
+            blob += payload
+            # 4.xx / 5.xx responses don't carry Block2 continuation —
+            # bail with whatever we got. Caller decides if 4.xx is fatal.
+            if code >> 5 != 2:
+                return code, blob
+            b2 = [v for n, v in ropts if n == BLOCK2]
+            more = 0
+            if b2:
+                bv = int.from_bytes(b2[0], 'big')
+                more = (bv >> 3) & 1
+            if not more:
+                break
+            num += 1
+            if num > self.MAX_BLOCKS:
+                raise ConnectionError(
+                    f"GET /{'/'.join(path_segs)}: >{self.MAX_BLOCKS} "
+                    f"blocks, aborting")
+        return last_code, blob
+
+    def post(self, path_segs, body_cbor, timeout=8.0):
+        """Single-frame POST with a CBOR-encoded body. Returns
+        (code, payload_bytes). body_cbor must already be encoded."""
+        if self.conn is None:
+            raise ConnectionError("DTLS session closed")
+        tok = self._next_tok()
+        mid = self._next_mid()
+        opts = [(URI_PATH, s.encode()) for s in path_segs]
+        opts.append((CONTENT_FORMAT, CF_CBOR))
+        opts.append((ACCEPT, CF_CBOR))
+        datagram = build_coap(TYPE_CON, METHOD_POST, mid, tok, opts,
+                              body_cbor)
+        ev = threading.Event()
+        container = {}
+        self._pending[tok] = (ev, container)
+        try:
+            self._send_dgram(datagram)
+            if not ev.wait(timeout):
+                raise TimeoutError(
+                    f"POST /{'/'.join(path_segs)} timeout")
+            if 'err' in container:
+                raise ConnectionError(container['err'])
+            return container['code'], container['payload']
+        finally:
+            self._pending.pop(tok, None)
+
+    def ping(self):
+        """RFC 7252 §4.4 CoAP Ping — empty CON, no token, no payload.
+        Peer MUST respond with an RST sharing the same Message ID.
+        We don't block waiting for the RST here; the reader thread
+        sees it, finds nothing in _pending/_observe_tokens for an
+        empty token, and drops it quietly — which is the documented
+        behaviour for unsolicited RST.
+
+        Sole purpose is keepalive: gives Samsung's RT-OCF stack a
+        visible "client still here" signal so it doesn't expire our
+        OBSERVE subscriptions (symptom of expiry: POSTs still get
+        2.04 but OBSERVE notifications stop arriving)."""
+        if self.conn is None:
+            raise ConnectionError("DTLS session closed")
+        mid = self._next_mid()
+        # Empty message: ver=01, type=CON, tkl=0, code=0.00, mid, no
+        # token, no options, no payload. build_coap with empty token
+        # and no options yields exactly that.
+        self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', []))
+        return mid
+
+    def subscribe(self, path_segs):
+        """Register an OBSERVE on the given path. The initial 2.05
+        notification and all subsequent state-change notifications
+        will fire on_notification(href, payload_bytes).
+
+        Returns the token used (in case the caller wants to deregister
+        later)."""
+        if self.conn is None:
+            raise ConnectionError("DTLS session closed")
+        tok = self._next_observe_tok()
+        href = '/' + '/'.join(path_segs)
+        # Register the token BEFORE sending — otherwise the device
+        # could respond between send() and the dict insert, and the
+        # reader thread would drop the initial 2.05 as "stale".
+        self._observe_tokens[tok] = href
+        mid = self._next_mid()
+        opts = [(URI_PATH, s.encode()) for s in path_segs]
+        opts.append((OBSERVE, OBSERVE_REGISTER))
+        opts.append((ACCEPT, CF_CBOR))
+        self._send_dgram(
+            build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+        return tok

custom_components/localthings/ocf/observe_refresh.py

@@ -0,0 +1,50 @@
+"""Periodic OBSERVE re-subscribe.
+
+CoAP OBSERVE (RFC 7641) has no built-in TTL, but real-world peers age
+out observer state on their own schedule — Samsung's RT-OCF is known
+to silently drop notify delivery during cloud auth blips even though
+the DTLS session stays healthy. Without a re-subscribe, recovery from
+such a blip requires a full session reconnect.
+
+This task derregisters the current observer tokens and re-subscribes
+every `interval_s`. Cheap (one register CON per path), idempotent
+(Samsung silently no-ops a register on an already-active token), and
+resilient — individual subscribe failures are logged but don't abort
+the task.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Optional
+
+from .coap_dtls import DtlsCoapSession
+
+
+class ObserveRefreshTask:
+
+    def __init__(self,
+                 session: DtlsCoapSession,
+                 paths,
+                 interval_s: float = 6 * 3600.0,
+                 logger=None):
+        self.session = session
+        self.paths = [list(p) for p in paths]
+        self.interval_s = interval_s
+        self.log = logger
+        self._refresh_count = 0
+
+    @property
+    def refresh_count(self) -> int:
+        return self._refresh_count
+
+    def run_forever(self, stop: threading.Event) -> None:
+        while not stop.wait(self.interval_s):
+            try:
+                self.session.refresh_observes(self.paths)
+                self._refresh_count += 1
+                if self.log:
+                    self.log.info("OBSERVE refresh #%d (%d paths)",
+                                  self._refresh_count, len(self.paths))
+            except Exception as e:
+                if self.log:
+                    self.log.warning("OBSERVE refresh: %s", e)

custom_components/localthings/ocf/ocf_root_ca.pem

@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICXTCCAgGgAwIBAgIBATAMBggqhkjOPQQDAgUAMGsxKDAmBgNVBAMTH1NhbXN1
+bmcgRWxlY3Ryb25pY3MgT0NGIFJvb3QgQ0ExFDASBgNVBAsTC09DRiBSb290IENB
+MRwwGgYDVQQKExNTYW1zdW5nIEVsZWN0cm9uaWNzMQswCQYDVQQGEwJLUjAgFw0x
+NjExMjQwMjU1MTFaGA8yMDY5MTIzMTE0NTk1OVowazEoMCYGA1UEAxMfU2Ftc3Vu
+ZyBFbGVjdHJvbmljcyBPQ0YgUm9vdCBDQTEUMBIGA1UECxMLT0NGIFJvb3QgQ0Ex
+HDAaBgNVBAoTE1NhbXN1bmcgRWxlY3Ryb25pY3MxCzAJBgNVBAYTAktSMFkwEwYH
+KoZIzj0CAQYIKoZIzj0DAQcDQgAEYsf/Kx+sUFBQESbuytTDPwLPIe0X/8/B1L7b
+3abxE9w0gQZAfI8WYUkKfNfP7HXh1M5SCnOkfwWraltGOKTeX6OBkTCBjjAOBgNV
+HQ8BAf8EBAMCAcYwMgYDVR0fBCswKTAnoCWgI4YhaHR0cDovL3Byb2RjYS5zYW1z
+dW5naW90cy5jb20vY3JsMA8GA1UdEwEB/wQFMAMBAf8wNwYIKwYBBQUHAQEEKzAp
+MCcGCCsGAQUFBzABhhtodHRwOi8vb2NzcC5zYW1zdW5naW90cy5jb20wDAYIKoZI
+zj0EAwIFAANIADBFAiARY9aSE30q31q5v8B4sJczBqOp7AsD9o8ZIuNmH7IwSwIh
+ALfR56jcXoFis/nDx0tQ2xTI/f0b7F6tWPqj2vyKQeNR
+-----END CERTIFICATE-----

custom_components/localthings/ocf/registry/__init__.py

@@ -0,0 +1,5 @@
+from .adapter import flatten, is_active, _key
+from .discovery import BoundEntity, discover
+from .registry import CAPABILITIES
+
+__all__ = ['CAPABILITIES', 'discover', 'BoundEntity', 'flatten', 'is_active', '_key']

custom_components/localthings/ocf/registry/adapter.py

@@ -0,0 +1,33 @@
+"""Adapter: BoundEntity list → flat state dict and command dispatch."""
+from __future__ import annotations
+
+from typing import Any
+
+from .discovery import BoundEntity
+
+
+def _key(b: BoundEntity) -> str:
+    return f"{b.key_override or b.desc.key}{b.instance}"
+
+
+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 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 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

custom_components/localthings/ocf/registry/by_type/__init__.py

@@ -0,0 +1,57 @@
+"""Per-device-type registries."""
+from typing import Optional
+
+from ._base import DeviceRegistry
+from . import dishwasher, dryer, oven, refrigerator
+
+__all__ = ['DeviceRegistry', '_type_key', 'for_device']
+
+
+_REGISTRY_BY_KEY: dict[str, DeviceRegistry] = {
+    'dishwasher': dishwasher.REGISTRY,
+    'dryer': dryer.REGISTRY,
+    'oven': oven.REGISTRY,
+    'refrigerator': refrigerator.REGISTRY,
+}
+
+
+def _type_key(one_ui_version: str) -> str:
+    """Convert oneUiVersion string to registry key.
+
+    Args:
+        one_ui_version: String like '7.0 Dishwasher' or 'Oven'.
+
+    Returns:
+        Lowercase key with version prefix stripped and spaces/hyphens converted to underscores.
+
+    Examples:
+        '7.0 Dishwasher' -> 'dishwasher'
+        '7.0 French Door Refrigerator' -> 'french_door_refrigerator'
+        'Oven' -> 'oven'
+    """
+    if ' ' in one_ui_version:
+        # Strip version prefix: everything before and including the first space
+        suffix = one_ui_version.split(' ', 1)[-1]
+    else:
+        suffix = one_ui_version
+
+    return suffix.lower().replace(' ', '_').replace('-', '_')
+
+
+def for_device(one_ui_version: str) -> Optional[DeviceRegistry]:
+    """Return the DeviceRegistry for the given oneUiVersion string, or None if unknown.
+
+    Args:
+        one_ui_version: Device's oneUiVersion string (e.g., '7.0 Dishwasher').
+
+    Returns:
+        DeviceRegistry if a matching registry exists, None otherwise.
+    """
+    key = _type_key(one_ui_version)
+    if key in _REGISTRY_BY_KEY:
+        return _REGISTRY_BY_KEY[key]
+    # Suffix fallback: e.g. "french_door_refrigerator" ends with "_refrigerator"
+    for rkey, reg in _REGISTRY_BY_KEY.items():
+        if key.endswith(f'_{rkey}'):
+            return reg
+    return None

custom_components/localthings/ocf/registry/by_type/_base.py

@@ -0,0 +1,47 @@
+"""Base DeviceRegistry dataclass and builder."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from ..capability import Capability
+
+
+@dataclass(frozen=True)
+class DeviceRegistry:
+    """Registry of capabilities for a specific device type."""
+    name: str
+    capabilities: dict[str, list[Capability]]
+    pattern_capabilities: list[Capability] = field(default_factory=list)
+
+
+def _build(caps: list[Capability]) -> dict[str, list[Capability]]:
+    """Build a capabilities dict from a list of Capability objects.
+
+    Args:
+        caps: List of Capability objects to organize by href.
+
+    Returns:
+        A dict mapping href to list of Capability objects sharing that href.
+
+    Raises:
+        ValueError: If any capability has href=None (use pattern_capabilities instead),
+                   or if multiple caps share an href without all having rt_filter or match_fn.
+    """
+    out: dict[str, list[Capability]] = {}
+
+    for cap in caps:
+        if cap.href is None:
+            raise ValueError(f"Use pattern_capabilities for href=None caps")
+        if cap.href_prefix is not None:
+            raise ValueError(
+                f"href_prefix is only valid for pattern caps (href=None); "
+                f"cap with href={cap.href!r} must not set href_prefix")
+        out.setdefault(cap.href, []).append(cap)
+
+    # Validate that multi-cap hrefs have proper discrimination
+    for href, cs in out.items():
+        if len(cs) > 1 and any(c.rt_filter is None and c.match_fn is None for c in cs):
+            raise ValueError(
+                f"href {href!r} has multiple caps but at least one lacks rt_filter and match_fn")
+
+    return out

custom_components/localthings/ocf/registry/by_type/dishwasher.py

@@ -0,0 +1,20 @@
+"""Dishwasher device registry."""
+from ..capabilities import common, fridge, laundry, operational
+from ._base import DeviceRegistry, _build
+
+REGISTRY = DeviceRegistry(
+    name='dishwasher',
+    capabilities=_build([
+        common.POWER,
+        common.KIDS_LOCK,
+        common.REMOTE_CONTROL,
+        common.ALARMS,
+        common.ENERGY_METER,
+        common.WATER_METER,
+        common.WATER_FILTER,
+        operational.OPERATIONAL_STATE,
+        laundry.DOOR_LED,
+        laundry.SOUND_MODE,
+        fridge.FIRMWARE_UPDATE,
+    ]),
+)

custom_components/localthings/ocf/registry/by_type/dryer.py

@@ -0,0 +1,26 @@
+"""Dryer device 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
+
+REGISTRY = DeviceRegistry(
+    name='dryer',
+    capabilities=_build([
+        common.POWER,
+        common.KIDS_LOCK,
+        common.REMOTE_CONTROL,
+        common.ALARMS,
+        common.ENERGY_METER,
+        operational.OPERATIONAL_STATE,
+        laundry.DOOR_LED,
+        laundry.SOUND_MODE,
+        dryer.DRYER_SETTINGS,
+        dryer.DRYER_COURSE,
+        dryer.JOB_BEGINNING_STATUS,
+        dryer.DRYER_DIAGNOSIS,
+        fridge.FIRMWARE_UPDATE,
+    ]),
+)

custom_components/localthings/ocf/registry/by_type/oven.py

@@ -0,0 +1,20 @@
+"""Oven device registry."""
+from ..capabilities import common, fridge, oven
+from ._base import DeviceRegistry, _build
+
+REGISTRY = DeviceRegistry(
+    name='oven',
+    capabilities=_build([
+        common.POWER,
+        common.KIDS_LOCK,
+        common.REMOTE_CONTROL,
+        common.ALARMS,
+        oven.OVEN_CAVITY,
+        oven.OVEN_SETPOINT,
+        oven.OVEN_MODE,
+        oven.OVEN_OPERATIONAL_STATE,
+        oven.OVEN_DOOR,
+        oven.OVEN_CONNECTED,
+        fridge.FIRMWARE_UPDATE,
+    ]),
+)

custom_components/localthings/ocf/registry/by_type/refrigerator.py

@@ -0,0 +1,29 @@
+"""Refrigerator device registry."""
+from ..capabilities import common, fridge
+from ._base import DeviceRegistry, _build
+
+REGISTRY = DeviceRegistry(
+    name='refrigerator',
+    capabilities=_build([
+        common.POWER,
+        common.KIDS_LOCK,
+        common.REMOTE_CONTROL,
+        common.ALARMS,
+        common.ENERGY_METER,
+        common.WATER_FILTER,
+        fridge.DOORS_STATUS,
+        fridge.ICEMAKER_STATUS,
+        fridge.REFRIGERATION,
+        fridge.AUTOFILL,
+        fridge.CABINET_LIGHT,
+        fridge.SABBATH,
+        fridge.BEVERAGE_ZONE,
+        fridge.FIRMWARE_UPDATE,
+    ]),
+    pattern_capabilities=[
+        fridge.TEMP_CURRENT_GENERIC,
+        fridge.TEMP_SETPOINT_GENERIC,
+        fridge.ICEMAKER_GENERIC,
+        fridge.DOOR_GENERIC,
+    ],
+)

custom_components/localthings/ocf/registry/capabilities/__init__.py

@@ -0,0 +1,19 @@
+from . import common, fridge, laundry, operational, oven
+from ..capability import Capability
+
+
+def _is_capability(v):
+    return isinstance(v, Capability)
+
+
+# Only OVEN_CAVITY (/oven/vs/0) is safe to include globally — that href
+# is oven-unique. OVEN_SETPOINT (/temperatures/vs/0) and OVEN_MODE
+# (/mode/vs/0) collide with fridge hrefs that share the same path but
+# have different schemas. Those capabilities require rt_filter or class-
+# scoped registry support before they can be added back to ALL.
+_OVEN_GLOBAL_CAPS = [
+    oven.OVEN_CAVITY,
+]
+
+ALL = [v for mod in (common, operational, laundry, fridge)
+       for v in vars(mod).values() if _is_capability(v)] + _OVEN_GLOBAL_CAPS

custom_components/localthings/ocf/registry/capabilities/common.py

@@ -0,0 +1,126 @@
+"""Capabilities shared across multiple appliance families.
+
+Each capability is keyed on the stable OCF resource href (not rt), verified
+against live device dumps:
+  /kidslock/vs/0            -> x.com.samsung.da.kidsLock
+  /remotectrl/vs/0          -> x.com.samsung.da.remoteControlEnabled
+  /power/vs/0               -> x.com.samsung.da.power
+  /alarms/vs/0              -> x.com.samsung.da.items
+  /energy/consumption/vs/0  -> x.com.samsung.da.instantaneousPower / cumulativePower
+  /water/consumption/vs/0   -> x.com.samsung.da.cumulativeWater
+  /filter/waterfilter/vs/0  -> x.com.samsung.da.filterUsage / filterStatus
+"""
+from ..capability import Capability
+from ..entities import BinarySensorDesc, SensorDesc, SwitchDesc
+
+
+def _num(v):
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _clamp_power(v):
+    n = _num(v)
+    return 0.0 if (n is not None and n < 0) else n
+
+
+def _wh_to_kwh(v):
+    n = _num(v)
+    return round(n / 1000.0, 2) if n is not None else None
+
+
+def _ml_to_l(v):
+    n = _num(v)
+    return round(n / 1000.0, 1) if n is not None else None
+
+
+def _last_alarm_code(items):
+    if items and isinstance(items, list):
+        return items[-1].get('x.com.samsung.da.code')
+    return None
+
+
+KIDS_LOCK = Capability(
+    href='/kidslock/vs/0',
+    entities=(
+        BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock',
+                         name='Child lock', device_class='lock',
+                         value_fn=lambda v: v != 'Ready'),
+        BinarySensorDesc(key='child_lock_binary', field='x.com.samsung.da.kidsLock',
+                         name='Child lock (binary)', device_class='lock',
+                         value_fn=lambda v: v != 'Ready'),
+    ),
+)
+
+REMOTE_CONTROL = Capability(
+    href='/remotectrl/vs/0',
+    entities=(
+        BinarySensorDesc(key='remote_control',
+                         field='x.com.samsung.da.remoteControlEnabled',
+                         name='Remote control', device_class='connectivity',
+                         value_fn=lambda v: str(v).lower() == 'true'),
+        BinarySensorDesc(key='remote_control_binary',
+                         field='x.com.samsung.da.remoteControlEnabled',
+                         name='Remote control (binary)', device_class='connectivity',
+                         value_fn=lambda v: str(v).lower() == 'true'),
+    ),
+)
+
+POWER = Capability(
+    href='/power/vs/0',
+    entities=(
+        SwitchDesc(key='power_switch', field='x.com.samsung.da.power',
+                   name='Power',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=lambda p, rep, href=None: (['power', 'vs', '0'], {'x.com.samsung.da.power': 'On' if p == 'On' else 'Off'})),
+        SensorDesc(key='power_state', field='x.com.samsung.da.power',
+                   name='Power state'),
+        BinarySensorDesc(key='power_state_binary', field='x.com.samsung.da.power',
+                         name='Power state (binary)',
+                         value_fn=lambda v: v == 'On'),
+    ),
+)
+
+ALARMS = Capability(
+    href='/alarms/vs/0',
+    entities=(
+        SensorDesc(key='alarm_code', field='x.com.samsung.da.items',
+                   name='Alarm code', icon='mdi:alert',
+                   entity_category='diagnostic', value_fn=_last_alarm_code),
+    ),
+)
+
+ENERGY_METER = Capability(
+    href='/energy/consumption/vs/0',
+    entities=(
+        SensorDesc(key='power_watts', field='x.com.samsung.da.instantaneousPower',
+                   name='Power', device_class='power', state_class='measurement',
+                   unit='W', value_fn=_clamp_power),
+        SensorDesc(key='energy_kwh', field='x.com.samsung.da.cumulativePower',
+                   name='Energy', device_class='energy',
+                   state_class='total_increasing', unit='kWh', value_fn=_wh_to_kwh),
+    ),
+)
+
+WATER_METER = Capability(
+    href='/water/consumption/vs/0',
+    entities=(
+        SensorDesc(key='water_liters', field='x.com.samsung.da.cumulativeWater',
+                   name='Water consumption', device_class='water',
+                   state_class='total_increasing', unit='L', icon='mdi:water',
+                   value_fn=_ml_to_l),
+    ),
+)
+
+WATER_FILTER = Capability(
+    href='/filter/waterfilter/vs/0',
+    entities=(
+        SensorDesc(key='filter_usage', field='x.com.samsung.da.filterUsage',
+                   name='Filter usage', unit='%', state_class='measurement',
+                   icon='mdi:filter'),
+        SensorDesc(key='filter_status', field='x.com.samsung.da.filterStatus',
+                   name='Filter status', icon='mdi:filter-check'),
+    ),
+)

custom_components/localthings/ocf/registry/capabilities/dryer.py

@@ -0,0 +1,102 @@
+"""Capabilities specific to the dryer family (Samsung DV5000T-class).
+
+Resources derived from the old dryer.py descriptor OBSERVE_PATHS and
+flatten() implementation:
+  /washer/vs/0              -> DRYER_SETTINGS (dryLevel, dryTime, dryerType, wrinklePrevent)
+  /st/dryercourse/vs/0      -> DRYER_COURSE (dryer_mode SelectDesc)
+  /wm/jobbeginingstatus/vs/0 -> JOB_BEGINNING_STATUS
+  /diagnosis/vs/0           -> DRYER_DIAGNOSIS
+
+Course table captured 2026-05-29 on a DA_WM_TP2_20_COMMON_DV5000T. Other
+dryer models may use a different course table; options=() means HA renders
+whatever the device reports, and the write_fn validates against this table.
+"""
+from ..capability import Capability
+from ..entities import SelectDesc, SensorDesc, SwitchDesc
+
+# Course table: hex codes -> human names (Table_03, DV5000T-class).
+_COURSE_NAMES = {
+    0x16: 'Cotton',
+    0x18: 'Synthetics',
+    0x19: 'Delicates',
+    0x1A: 'Wool',
+    0x1B: 'Bedding',
+    0x1C: 'Shirts',
+    0x1D: 'Towels',
+    0x1E: 'Outdoor',
+    0x1F: 'Mixed Load',
+    0x20: 'Iron Dry',
+    0x23: 'Quick Dry 35',
+    0x24: 'Cool Air',
+    0x25: 'Warm Air',
+    0x27: 'Time Dry',
+}
+_COURSE_CODE_BY_NAME = {name: code for code, name in _COURSE_NAMES.items()}
+
+
+def _wrinkle_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    return ['washer', 'vs', '0'], {'x.com.samsung.da.wrinklePrevent': p}
+
+
+def _course_write(p, rep, href=None):
+    """Encode a human course name to the Samsung hex-encoded course string."""
+    code = _COURSE_CODE_BY_NAME.get(p)
+    if code is None:
+        return None
+    return ['st', 'dryercourse', 'vs', '0'], {
+        'x.com.samsung.da.st.dryerMode': f'Course_{code:02X}',
+    }
+
+
+# ---------------------------------------------------------------------------
+# Capabilities
+# ---------------------------------------------------------------------------
+
+DRYER_SETTINGS = Capability(
+    href='/washer/vs/0',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='dry_level', field='x.com.samsung.da.dryLevel',
+                   name='Dry level', icon='mdi:water-percent'),
+        SensorDesc(key='dry_time', field='x.com.samsung.da.dryTime',
+                   name='Dry time', icon='mdi:timer'),
+        SensorDesc(key='dryer_type', field='x.com.samsung.da.dryerType',
+                   name='Dryer type', icon='mdi:tumble-dryer'),
+        SwitchDesc(key='wrinkle_prevent', field='x.com.samsung.da.wrinklePrevent',
+                   name='Wrinkle prevent', icon='mdi:iron',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_wrinkle_write),
+    ),
+)
+
+DRYER_COURSE = Capability(
+    href='/st/dryercourse/vs/0',
+    poll_tier='warm',
+    entities=(
+        SelectDesc(key='dryer_mode', field='x.com.samsung.da.st.dryerMode',
+                   name='Dryer mode', icon='mdi:tumble-dryer',
+                   options=(), write_fn=_course_write),
+    ),
+)
+
+JOB_BEGINNING_STATUS = Capability(
+    href='/wm/jobbeginingstatus/vs/0',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='job_beginning_status',
+                   field='x.com.samsung.da.jobBeginingStatus',
+                   name='Job beginning status',
+                   entity_category='diagnostic'),
+    ),
+)
+
+DRYER_DIAGNOSIS = Capability(
+    href='/diagnosis/vs/0',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='diagnosis', field='x.com.samsung.da.diagnosisStart',
+                   name='Diagnosis', entity_category='diagnostic'),
+    ),
+)

custom_components/localthings/ocf/registry/capabilities/fridge.py

@@ -0,0 +1,278 @@
+"""Capabilities for the refrigerator family (Samsung RF9000B-class).
+
+Resources verified against the dump at local-tools/dumps/10.0.0.254.json.
+
+Temperature fields in all /temperature/* resources are in Fahrenheit on this
+model. Setpoints are NumberDesc with direct-write write_fn — generic caps
+derive the CoAP PUT path from href at write time.
+
+Multi-instance note: the two door resources (/door/cooler/0 and
+/door/freezer/0) and the two ice-maker resources (/icemaker/one/vs/0 and
+/icemaker/two/vs/0) use named path segments, so they are modelled via
+pattern capabilities that auto-derive distinct entity keys from href segments.
+"""
+from ..capability import Capability
+from ..entities import (
+    BinarySensorDesc, NumberDesc, SelectDesc, SensorDesc, SwitchDesc,
+)
+
+# Beverage zone flex modes.
+BZONE_MODES = ('SP_TTYPE_BEER_DRINKS', 'SP_TTYPE_WINE_DESSERT')
+
+
+def _int(v):
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Temperature (generic — covers /temperature/current/* and /temperature/desired/*)
+# ---------------------------------------------------------------------------
+
+TEMP_CURRENT_GENERIC = Capability(
+    href=None,
+    href_prefix='/temperature/current/',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='temp_f', field='temperature',
+                   name=None, icon='mdi:thermometer'),
+    ),
+)
+
+TEMP_SETPOINT_GENERIC = Capability(
+    href=None,
+    href_prefix='/temperature/desired/',
+    poll_tier='warm',
+    entities=(
+        NumberDesc(key='temp_f', field='temperature',
+                   name=None, native_min=-20.0, native_max=50.0,
+                   write_fn=lambda p, rep, href=None: (
+                       [s for s in href.strip('/').split('/') if s],
+                       {'temperature': int(round(float(p)))}
+                   ) if href else None),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Icemaker (generic — covers /icemaker/one/vs/0, /icemaker/two/vs/0)
+# /icemaker/status/vs/0 is kept as exact-href cap and binds first.
+# /icemaker/nighttime/vs/0 is excluded by match_fn (lacks iceMaker.state).
+# ---------------------------------------------------------------------------
+
+def _icemaker_write(field):
+    return lambda p, rep, href=None: (
+        [s for s in href.strip('/').split('/') if s],
+        {field: p}
+    ) if href else None
+
+ICEMAKER_GENERIC = Capability(
+    href=None,
+    href_prefix='/icemaker/',
+    match_fn=lambda rep, resources: 'x.com.samsung.da.iceMaker.state' in rep,
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='state', field='x.com.samsung.da.iceMaker.state',
+                   name=None, icon='mdi:cube-outline'),
+        BinarySensorDesc(key='on', field='x.com.samsung.da.iceMaker.state',
+                         name=None, device_class='running',
+                         value_fn=lambda v: v == 'On'),
+        SensorDesc(key='making_status',
+                   field='x.com.samsung.da.iceMaker.iceMakingStatus',
+                   name=None, icon='mdi:cube-outline'),
+        SwitchDesc(key='enabled', field='x.com.samsung.da.iceMaker.state',
+                   name=None, icon='mdi:cube-outline',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_icemaker_write('x.com.samsung.da.iceMaker.state')),
+        SelectDesc(key='type', field='x.com.samsung.da.iceType.desired',
+                   name=None, icon='mdi:cube-outline',
+                   options=(),
+                   write_fn=_icemaker_write('x.com.samsung.da.iceType.desired')),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Doors
+# ---------------------------------------------------------------------------
+
+DOORS_STATUS = Capability(
+    href='/doors/vs/0',
+    poll_tier='hot',
+    entities=(
+        BinarySensorDesc(
+            key='any_door_open',
+            field='x.com.samsung.da.items',
+            name='Any door open',
+            device_class='door',
+            value_fn=lambda items: any(
+                it.get('x.com.samsung.da.openState') == 'Open'
+                for it in (items or [])
+            ),
+        ),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Ice maker status (exact-href — binds before ICEMAKER_GENERIC pattern)
+# ---------------------------------------------------------------------------
+
+ICEMAKER_STATUS = Capability(
+    href='/icemaker/status/vs/0',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='ice_maker_enabled', field='x.com.samsung.da.iceMaker',
+                   name='Ice maker enabled', icon='mdi:snowflake'),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Refrigeration modes (rapid cooling)
+# ---------------------------------------------------------------------------
+
+def _refrigeration_write(field_name):
+    def _write(p, rep, href=None):
+        if p not in ('On', 'Off'):
+            return None
+        return ['refrigeration', 'vs', '0'], {field_name: p}
+    return _write
+
+
+REFRIGERATION = Capability(
+    href='/refrigeration/vs/0',
+    poll_tier='warm',
+    entities=(
+        SwitchDesc(key='rapid_fridge', field='x.com.samsung.da.rapidFridge',
+                   name='Rapid fridge', icon='mdi:fridge-industrial',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_refrigeration_write('x.com.samsung.da.rapidFridge')),
+        SwitchDesc(key='rapid_freezing', field='x.com.samsung.da.rapidFreezing',
+                   name='Rapid freezing', icon='mdi:snowflake',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_refrigeration_write('x.com.samsung.da.rapidFreezing')),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Autofill
+# ---------------------------------------------------------------------------
+
+def _autofill_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    return ['autofill', 'vs', '0'], {'x.com.samsung.da.autofill': p}
+
+
+AUTOFILL = Capability(
+    href='/autofill/vs/0',
+    poll_tier='warm',
+    entities=(
+        SwitchDesc(key='autofill', field='x.com.samsung.da.autofill',
+                   name='Autofill', icon='mdi:cup-water',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_autofill_write),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Cabinet light
+# ---------------------------------------------------------------------------
+
+def _cabinet_light_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    return ['cabinet', 'light', 'total', 'vs', '0'], {
+        'x.com.samsung.da.lightControl': p,
+    }
+
+
+CABINET_LIGHT = Capability(
+    href='/cabinet/light/total/vs/0',
+    poll_tier='warm',
+    entities=(
+        SensorDesc(key='cabinet_light', field='x.com.samsung.da.lightControl',
+                   name='Cabinet light level', icon='mdi:fridge-outline'),
+        BinarySensorDesc(key='cabinet_light_on', field='x.com.samsung.da.lightControl',
+                         name='Cabinet light on', icon='mdi:fridge-outline',
+                         value_fn=lambda v: v == 'On'),
+        SwitchDesc(key='cabinet_light_switch', field='x.com.samsung.da.lightControl',
+                   name='Cabinet light', icon='mdi:fridge-outline',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_cabinet_light_write),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Sabbath mode
+# ---------------------------------------------------------------------------
+
+def _sabbath_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    return ['sabbath', 'vs', '0'], {'x.com.samsung.da.sabbathMode': p}
+
+
+SABBATH = Capability(
+    href='/sabbath/vs/0',
+    poll_tier='warm',
+    entities=(
+        SwitchDesc(key='sabbath_mode', field='x.com.samsung.da.sabbathMode',
+                   name='Sabbath mode', icon='mdi:star-david',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_sabbath_write),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Beverage zone
+# ---------------------------------------------------------------------------
+
+def _bzone_write(p, rep, href=None):
+    if p not in BZONE_MODES:
+        return None
+    return ['specialzone', 'one', 'vs', '0'], {'roomDesiredMode': p}
+
+
+BEVERAGE_ZONE = Capability(
+    href='/specialzone/one/vs/0',
+    poll_tier='warm',
+    entities=(
+        SelectDesc(key='beverage_zone_mode', field='roomDesiredMode',
+                   name='Beverage zone mode', icon='mdi:glass-wine',
+                   options=BZONE_MODES, write_fn=_bzone_write),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Firmware update
+# ---------------------------------------------------------------------------
+
+FIRMWARE_UPDATE = Capability(
+    href='/otninformation/vs/0',
+    poll_tier='cold',
+    entities=(
+        BinarySensorDesc(
+            key='firmware_update',
+            field='x.com.samsung.da.newVersionAvailable',
+            name='Firmware update available',
+            device_class='update',
+            entity_category='diagnostic',
+            value_fn=lambda v: str(v).lower() == 'true' if v is not None else None,
+        ),
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# Generic door pattern capability (href=None — use as pattern_cap only)
+# ---------------------------------------------------------------------------
+
+DOOR_GENERIC = Capability(
+    href=None,
+    href_prefix='/door/',
+    poll_tier='hot',
+    entities=(
+        BinarySensorDesc(key='open', field='openState',
+                         name=None, device_class='door',
+                         value_fn=lambda v: v == 'Open'),
+    ),
+)

custom_components/localthings/ocf/registry/capabilities/laundry.py

@@ -0,0 +1,52 @@
+"""Capabilities specific to the laundry-family appliances (dishwasher, dryer).
+
+Resources verified against the DW9000F-class dump at 10.0.0.129.
+Note: door-LED keys use NO `x.com.samsung.da.` prefix — `setBrightness` and
+`setNightLight` — preserved exactly as they appear in the OCF resource rep.
+"""
+from ..capability import Capability
+from ..entities import SelectDesc, SwitchDesc
+
+_LED_LEVELS = ('Low', 'High')
+_SOUND_MODES = ('voice', 'tone', 'mute')
+
+
+def _led_brightness_write(p, rep, href=None):
+    if p not in _LED_LEVELS:
+        return None
+    return ['doorled', 'light', 'vs', '0'], {'setBrightness': p}
+
+
+def _led_night_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    return ['doorled', 'light', 'vs', '0'], {'setNightLight': p}
+
+
+def _sound_mode_write(p, rep, href=None):
+    if p not in _SOUND_MODES:
+        return None
+    return ['settings', 'sound', 'mode', 'vs', '0'], {'mode': p}
+
+
+DOOR_LED = Capability(
+    href='/doorled/light/vs/0',
+    entities=(
+        SelectDesc(key='led_brightness', field='setBrightness',
+                   name='Door LED brightness', icon='mdi:brightness-6',
+                   options=_LED_LEVELS, write_fn=_led_brightness_write),
+        SwitchDesc(key='led_night_light', field='setNightLight',
+                   name='Door LED night light', icon='mdi:weather-night',
+                   value_fn=lambda v: v == 'On',
+                   write_fn=_led_night_write),
+    ),
+)
+
+SOUND_MODE = Capability(
+    href='/settings/sound/mode/vs/0',
+    entities=(
+        SelectDesc(key='sound_mode', field='mode',
+                   name='Sound mode', icon='mdi:volume-high',
+                   options=_SOUND_MODES, write_fn=_sound_mode_write),
+    ),
+)

custom_components/localthings/ocf/registry/capabilities/operational.py

@@ -0,0 +1,117 @@
+"""Operational-state capability: machine state, progress, remaining-time.
+
+Carries the rare cross-field hooks (active_when, on_observation, project) for
+remaining-time extrapolation. Shared by dryer/dishwasher/oven/washer families.
+"""
+import time
+
+from ..capability import Capability
+from ..entities import BinarySensorDesc, ButtonDesc, SensorDesc
+
+_SAMSUNG_STATE_TO_OCF = {
+    'Ready': 'idle', 'Run': 'active', 'Running': 'active',
+    'Pause': 'pause', 'Paused': 'pause', 'End': 'idle', 'Stop': 'idle',
+}
+
+
+def _to_ocf(v):
+    return _SAMSUNG_STATE_TO_OCF.get(v, v) if v is not None else None
+
+
+def _progress(v):
+    return 'Idle' if v in (None, 'None') else v
+
+
+def _int(v):
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _active_when(rep):
+    return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
+
+
+def _on_observation(state, rep):
+    rem = rep.get('x.com.samsung.da.remainingTime')
+    if not isinstance(rem, str):
+        return
+    try:
+        h, m, s = rem.split(':')
+        state['remaining_anchor'] = (time.time(),
+                                     int(h) * 3600 + int(m) * 60 + int(s))
+    except (ValueError, AttributeError):
+        pass
+
+
+def _project(state, sensors):
+    anchor = state.get('remaining_anchor')
+    if sensors.get('machine_state') != 'active' or anchor is None:
+        return sensors
+    ts, total = anchor
+    remaining = max(0, int(total - (time.time() - ts)))
+    h, rest = divmod(remaining, 3600)
+    m, s = divmod(rest, 60)
+    sensors = dict(sensors)
+    sensors['completion_time'] = f"{h}:{m:02d}:{s:02d}"
+    sensors['completion_minutes'] = h * 60 + m + (1 if s > 0 else 0)
+    return sensors
+
+
+def _rem_minutes(remaining):
+    if not remaining:
+        return None
+    try:
+        h, m, s = remaining.split(':')
+        return int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
+    except Exception:
+        return None
+
+
+OPERATIONAL_STATE = Capability(
+    href='/operational/state/vs/0',
+    poll_tier='hot',
+    entities=(
+        SensorDesc(key='machine_state', field='x.com.samsung.da.state',
+                   name='Machine state', value_fn=_to_ocf),
+        # cycle_active is a bool derived from machine_state; used by the
+        # adapter to gate oven writes (cycle_active_field='cycle_active').
+        # Harmless for non-oven appliances — just an extra bool in state.
+        BinarySensorDesc(key='cycle_active', field='x.com.samsung.da.state',
+                         name='Cycle active', device_class='running',
+                         value_fn=lambda v: _SAMSUNG_STATE_TO_OCF.get(v) == 'active'),
+        SensorDesc(key='progress', field='x.com.samsung.da.progress',
+                   name='Progress', icon='mdi:progress-wrench', value_fn=_progress),
+        SensorDesc(key='progress_percentage',
+                   field='x.com.samsung.da.progressPercentage',
+                   name='Progress percent', unit='%', state_class='measurement',
+                   value_fn=_int),
+        SensorDesc(key='completion_time', field='x.com.samsung.da.remainingTime',
+                   name='Completion time', icon='mdi:timer-sand'),
+        SensorDesc(key='completion_minutes', field='x.com.samsung.da.remainingTime',
+                   name='Remaining minutes', unit='min', device_class='duration',
+                   state_class='measurement',
+                   value_fn=lambda r: _rem_minutes(r)),
+        SensorDesc(key='delay_start_time', field='x.com.samsung.da.delayStartTime',
+                   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, 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, href=None: (
+                       ['operational', 'state', 'vs', '0'],
+                       {'x.com.samsung.da.state': p})),
+        ButtonDesc(key='stop', field='', name='Stop cycle', payload='Ready',
+                   icon='mdi:stop',
+                   write_fn=lambda p, rep, href=None: (
+                       ['operational', 'state', 'vs', '0'],
+                       {'x.com.samsung.da.state': p})),
+    ),
+    active_when=_active_when,
+    on_observation=_on_observation,
+    project=_project,
+)

custom_components/localthings/ocf/registry/capabilities/oven.py

@@ -0,0 +1,371 @@
+"""Capabilities for the oven family (Samsung NV7000BS-class).
+
+Resources verified against the live device via DTLS-CoAP.
+See `local-tools/comparisons/oven-tree.md` for the full field reference.
+
+Write surfaces this module exposes:
+
+  proven:
+    * Lamp via /mode/vs/0 options RMW (probe_oven_lamp_toggle.py)
+      — works even with Remote Control off.
+
+  unproven (first HA use is also the test):
+    * Sound, FastPreheat, NaturalSteam — same RMW pattern as lamp.
+    * Setpoint via /temperatures/vs/0 items RMW.
+    * Cook time via /operational/state/vs/0 operationTime/remainingTime.
+    * Mode select via /mode/vs/0 .modes — mid-cook acceptance unknown.
+    * Stop via /operational/state/vs/0 state='Ready'.
+
+Note: Cycle start is not implemented. Reverse-engineering shows local-OCF
+cycle start is not reproducible on this firmware (see project_oven_remote
+_start_open.md). Mode writes are also unreliable — the oven rolls them back
+once a cycle is active. OVEN_MODE is provided as a SelectDesc for fidelity
+but is effectively read-only in practice.
+"""
+import time
+
+from ..capability import Capability
+from ..entities import (
+    BinarySensorDesc, ButtonDesc, NumberDesc, SelectDesc, SensorDesc,
+    SwitchDesc,
+)
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+SETPOINT_MIN_C = 30
+SETPOINT_MAX_C = 270
+SETPOINT_STEP_C = 5
+
+# Mode options seen on NV7000BS-class. No dump exists so this list is inferred
+# from Samsung documentation and firmware observations. The firmware will
+# reject unknown modes; missing entries here are a coverage gap, not a bug.
+_OVEN_MODES = (
+    'NoOperation',
+    'Bake',
+    'Broil',
+    'Convection',
+    'ConvectionBake',
+    'ConvectionBroil',
+    'FrozenPizzaPlus',
+    'SlowCook',
+    'PlateWarm',
+    'AirFry',
+)
+
+_SAMSUNG_STATE_TO_OCF = {
+    'Ready':   'idle',
+    'Run':     'active',
+    'Running': 'active',
+    'Pause':   'pause',
+    'Paused':  'pause',
+    'End':     'idle',
+    'Stop':    'idle',
+}
+
+
+def _to_ocf(v):
+    return _SAMSUNG_STATE_TO_OCF.get(v, v) if v is not None else None
+
+
+def _int(v):
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _rem_minutes(remaining):
+    if not remaining:
+        return None
+    try:
+        h, m, s = remaining.split(':')
+        return int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
+    except Exception:
+        return None
+
+
+def _op_minutes(op_time):
+    """Parse 'H:MM:SS' operationTime into integer minutes."""
+    if not op_time:
+        return None
+    try:
+        h, m, s = op_time.split(':')
+        return int(h) * 60 + int(m) + (1 if int(s) > 0 else 0)
+    except Exception:
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Options-array helpers (shared by lamp, sound, fastpreheat, naturalsteam)
+# ---------------------------------------------------------------------------
+
+def _option_value(options, prefix):
+    """Find `<prefix>_<value>` in an options array and return <value>."""
+    for o in (options or []):
+        if isinstance(o, str) and o.startswith(prefix + '_'):
+            return o.split('_', 1)[1]
+    return None
+
+
+def _replace_in_options(options, prefix, new_value):
+    """Return a new options list with the `<prefix>_*` slot replaced."""
+    return [f"{prefix}_{new_value}" if o.startswith(prefix + '_') else o
+            for o in options]
+
+
+# ---------------------------------------------------------------------------
+# Write functions
+# ---------------------------------------------------------------------------
+
+def _oven_setpoint_write(p, rep, href=None):
+    """RMW write to /temperatures/vs/0 items array."""
+    try:
+        temp = float(p)
+    except (TypeError, ValueError):
+        return None
+    temp_i = int(round(temp / SETPOINT_STEP_C) * SETPOINT_STEP_C)
+    if not (SETPOINT_MIN_C <= temp_i <= SETPOINT_MAX_C):
+        return None
+    items = rep.get('x.com.samsung.da.items')
+    if not items:
+        return None
+    items = [dict(it) for it in items]
+    items[0]['x.com.samsung.da.desired'] = str(temp_i)
+    return ['temperatures', 'vs', '0'], {'x.com.samsung.da.items': items}
+
+
+def _cook_time_write(p, rep, href=None):
+    """Write operationTime + remainingTime (H:MM:SS) from minutes."""
+    try:
+        minutes = int(round(float(p)))
+    except (TypeError, ValueError):
+        return None
+    if not (0 <= minutes <= 1439):
+        return None
+    h, m = divmod(minutes, 60)
+    hms = f"{h:02d}:{m:02d}:00"
+    return ['operational', 'state', 'vs', '0'], {
+        'x.com.samsung.da.operationTime': hms,
+        'x.com.samsung.da.remainingTime': hms,
+    }
+
+
+def _stop_write(p, rep, href=None):
+    return ['operational', 'state', 'vs', '0'], {
+        'x.com.samsung.da.state': 'Ready',
+    }
+
+
+def _oven_mode_write(p, rep, href=None):
+    if p not in _OVEN_MODES:
+        return None
+    return ['mode', 'vs', '0'], {'x.com.samsung.da.modes': [p]}
+
+
+def _lamp_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    opts = list(rep.get('x.com.samsung.da.options') or [])
+    if not opts:
+        return None
+    return ['mode', 'vs', '0'], {
+        'x.com.samsung.da.options': _replace_in_options(opts, 'UpperLamp', p),
+    }
+
+
+def _sound_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    opts = list(rep.get('x.com.samsung.da.options') or [])
+    if not opts:
+        return None
+    return ['mode', 'vs', '0'], {
+        'x.com.samsung.da.options': _replace_in_options(opts, 'Sound', p),
+    }
+
+
+def _fastpreheat_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    opts = list(rep.get('x.com.samsung.da.options') or [])
+    if not opts:
+        return None
+    return ['mode', 'vs', '0'], {
+        'x.com.samsung.da.options': _replace_in_options(opts, 'fastpreheat', p),
+    }
+
+
+def _naturalsteam_write(p, rep, href=None):
+    if p not in ('On', 'Off'):
+        return None
+    opts = list(rep.get('x.com.samsung.da.options') or [])
+    if not opts:
+        return None
+    if not any(o.startswith('NaturalSteam_') for o in opts):
+        opts = opts + [f'NaturalSteam_{p}']
+    else:
+        opts = _replace_in_options(opts, 'NaturalSteam', p)
+    return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts}
+
+
+# ---------------------------------------------------------------------------
+# on_observation / project for remaining-time extrapolation
+# ---------------------------------------------------------------------------
+
+def _active_when(rep):
+    return _SAMSUNG_STATE_TO_OCF.get(rep.get('x.com.samsung.da.state')) == 'active'
+
+
+def _on_observation(state, rep):
+    rem = rep.get('x.com.samsung.da.remainingTime')
+    if not isinstance(rem, str):
+        return
+    try:
+        h, m, s = rem.split(':')
+        state['remaining_anchor'] = (time.time(),
+                                     int(h) * 3600 + int(m) * 60 + int(s))
+    except (ValueError, AttributeError):
+        pass
+
+
+def _project(state, sensors):
+    anchor = state.get('remaining_anchor')
+    if sensors.get('machine_state') != 'active' or anchor is None:
+        return sensors
+    ts, total = anchor
+    remaining = max(0, int(total - (time.time() - ts)))
+    h, rest = divmod(remaining, 3600)
+    m, s = divmod(rest, 60)
+    sensors = dict(sensors)
+    sensors['completion_time'] = f"{h}:{m:02d}:{s:02d}"
+    sensors['completion_minutes'] = h * 60 + m + (1 if s > 0 else 0)
+    return sensors
+
+
+# ---------------------------------------------------------------------------
+# Capabilities
+# ---------------------------------------------------------------------------
+
+OVEN_OPERATIONAL_STATE = Capability(
+    href='/operational/state/vs/0',
+    poll_tier='hot',
+    active_when=_active_when,
+    on_observation=_on_observation,
+    project=_project,
+    entities=(
+        SensorDesc(key='machine_state', field='x.com.samsung.da.state',
+                   name='Machine state', icon='mdi:stove', value_fn=_to_ocf),
+        BinarySensorDesc(key='cycle_active', field='x.com.samsung.da.state',
+                         name='Cycle active', device_class='running',
+                         value_fn=lambda v: _SAMSUNG_STATE_TO_OCF.get(v) == 'active'),
+        SensorDesc(key='progress_percentage',
+                   field='x.com.samsung.da.progressPercentage',
+                   name='Progress percent', unit='%', state_class='measurement',
+                   value_fn=_int),
+        SensorDesc(key='operation_time', field='x.com.samsung.da.operationTime',
+                   name='Elapsed time', icon='mdi:timer'),
+        SensorDesc(key='operation_time_minutes',
+                   field='x.com.samsung.da.operationTime',
+                   name='Operation time (minutes)', unit='min',
+                   state_class='measurement', value_fn=_op_minutes),
+        SensorDesc(key='completion_time', field='x.com.samsung.da.remainingTime',
+                   name='Completion time', icon='mdi:timer-sand'),
+        SensorDesc(key='completion_minutes', field='x.com.samsung.da.remainingTime',
+                   name='Remaining minutes', unit='min', device_class='duration',
+                   state_class='measurement', value_fn=_rem_minutes),
+        NumberDesc(key='cook_time', field='x.com.samsung.da.operationTime',
+                   name='Cook time', unit='min', native_min=0, native_max=1439,
+                   step=1.0, icon='mdi:timer', value_fn=_op_minutes,
+                   write_fn=_cook_time_write),
+        ButtonDesc(key='stop', field='', name='Stop cycle', icon='mdi:stop',
+                   payload='Stop', write_fn=_stop_write),
+    ),
+)
+
+OVEN_CAVITY = Capability(
+    href='/oven/vs/0',
+    poll_tier='hot',
+    entities=(
+        SensorDesc(key='oven_state', field='x.com.samsung.da.state',
+                   name='Cavity state'),
+    ),
+)
+
+OVEN_SETPOINT = Capability(
+    href='/temperatures/vs/0',
+    poll_tier='hot',
+    entities=(
+        # NumberDesc first — test_oven_setpoint_write_is_read_modify_write uses entities[0]
+        NumberDesc(key='oven_setpoint', field='x.com.samsung.da.items',
+                   name='Setpoint', device_class='temperature', unit='°C',
+                   native_min=float(SETPOINT_MIN_C), native_max=float(SETPOINT_MAX_C),
+                   step=float(SETPOINT_STEP_C), icon='mdi:thermometer-chevron-up',
+                   value_fn=lambda items: _int(
+                       (items[0].get('x.com.samsung.da.desired') if items else None)),
+                   write_fn=_oven_setpoint_write),
+        SensorDesc(key='current_temp_c', field='x.com.samsung.da.items',
+                   name='Temperature', device_class='temperature',
+                   state_class='measurement', unit='°C',
+                   value_fn=lambda items: _int(
+                       (items[0].get('x.com.samsung.da.current') if items else None))),
+    ),
+)
+
+OVEN_DOOR = Capability(
+    href='/doors/vs/0',
+    poll_tier='hot',
+    entities=(
+        SensorDesc(key='door', field='x.com.samsung.da.items',
+                   name='Door state',
+                   value_fn=lambda items: (
+                       items[0].get('x.com.samsung.da.openState')
+                       if items else None)),
+        BinarySensorDesc(key='door_open', field='x.com.samsung.da.items',
+                         name='Door', device_class='door',
+                         value_fn=lambda items: (
+                             items[0].get('x.com.samsung.da.openState') == 'Open'
+                             if items else None)),
+    ),
+)
+
+OVEN_CONNECTED = Capability(
+    href='/connected/vs/0',
+    poll_tier='warm',
+    entities=(
+        BinarySensorDesc(key='cloud_connected', field='x.com.samsung.da.connected',
+                         name='Cloud connected', device_class='connectivity',
+                         entity_category='diagnostic',
+                         value_fn=lambda v: v == 'On'),
+    ),
+)
+
+OVEN_MODE = Capability(
+    href='/mode/vs/0',
+    poll_tier='warm',
+    entities=(
+        # SelectDesc first — test_oven_mode_options_nonempty uses entities[0]
+        SelectDesc(key='oven_mode', field='x.com.samsung.da.modes',
+                   name='Cooking mode', icon='mdi:tune',
+                   options=_OVEN_MODES,
+                   value_fn=lambda v: v[0] if v else None,
+                   write_fn=_oven_mode_write),
+        SwitchDesc(key='lamp', field='x.com.samsung.da.options',
+                   name='Lamp', icon='mdi:track-light',
+                   value_fn=lambda opts: _option_value(opts, 'UpperLamp') == 'On',
+                   write_fn=_lamp_write),
+        SwitchDesc(key='sound', field='x.com.samsung.da.options',
+                   name='Sound', icon='mdi:volume-high',
+                   value_fn=lambda opts: _option_value(opts, 'Sound') == 'On',
+                   write_fn=_sound_write),
+        SwitchDesc(key='fast_preheat', field='x.com.samsung.da.options',
+                   name='Fast preheat', icon='mdi:fire',
+                   value_fn=lambda opts: _option_value(opts, 'fastpreheat') == 'On',
+                   write_fn=_fastpreheat_write),
+        SwitchDesc(key='natural_steam', field='x.com.samsung.da.options',
+                   name='Natural steam', icon='mdi:kettle-steam',
+                   value_fn=lambda opts: _option_value(opts, 'NaturalSteam') == 'On',
+                   write_fn=_naturalsteam_write),
+    ),
+)

custom_components/localthings/ocf/registry/capability.py

@@ -0,0 +1,21 @@
+"""A Capability binds one OCF resource href to the entities it produces."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Optional
+
+from .entities import SamsungEntityDescription
+
+
+@dataclass(frozen=True, kw_only=True)
+class Capability:
+    href: Optional[str] = None
+    entities: tuple[SamsungEntityDescription, ...] = ()
+    poll_tier: str = 'warm'                  # 'hot' | 'warm' | 'cold'
+    rt_filter: Optional[str] = None          # bind only if rt_filter in rep.get('rt', ())
+    href_prefix: Optional[str] = None        # pattern caps only: bind only if href starts with this
+    match_fn: Optional[Callable[[dict, dict], bool]] = None  # match_fn(rep, resources) -> bool
+    # Rare optional hooks — only operational-state-style resources use these.
+    active_when: Optional[Callable[[dict], bool]] = None
+    on_observation: Optional[Callable[[dict, dict], None]] = None
+    project: Optional[Callable[[dict, dict], dict]] = None

custom_components/localthings/ocf/registry/discovery.py

@@ -0,0 +1,83 @@
+"""Runtime discovery: device resources -> bound entities.
+
+A device advertises a set of OCF resources keyed by href. For every href
+present in the registry, emit the capability's entities bound to that resource.
+Unknown hrefs are a coverage gap, logged at debug and skipped.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Iterable, Optional
+
+from .capability import Capability
+from .entities import SamsungEntityDescription
+
+
+@dataclass
+class BoundEntity:
+    href: str
+    capability: Capability
+    desc: SamsungEntityDescription
+    instance: str = ''
+    key_override: Optional[str] = None
+
+
+def instance_suffix(href: str) -> str:
+    """'' for the index-0 instance, else '_<n>' from the trailing segment."""
+    tail = href.rstrip('/').rsplit('/', 1)[-1]
+    if tail.isdigit() and tail != '0':
+        return f'_{tail}'
+    return ''
+
+
+def discover(
+    resources: dict[str, dict],
+    registry: dict[str, list[Capability]],
+    pattern_caps: Iterable[Capability] = (),
+    log: Optional[Callable[[str], None]] = None,
+) -> list[BoundEntity]:
+    out: list[BoundEntity] = []
+
+    for href, rep in resources.items():
+        if not isinstance(rep, dict):
+            continue
+        rts = rep.get('rt') or ()
+        caps = registry.get(href) or []
+        matched = False
+
+        for cap in caps:
+            if cap.rt_filter is not None and cap.rt_filter not in rts:
+                continue
+            if cap.match_fn is not None and not cap.match_fn(rep, resources):
+                continue
+            inst = instance_suffix(href)
+            for desc in cap.entities:
+                out.append(BoundEntity(href=href, capability=cap,
+                                       desc=desc, instance=inst))
+            matched = True
+
+        if matched:
+            continue
+
+        # Pattern cap fallback — first matching pattern wins
+        for cap in pattern_caps:
+            if cap.href_prefix and not href.startswith(cap.href_prefix):
+                continue
+            if cap.rt_filter is not None and cap.rt_filter not in rts:
+                continue
+            if cap.match_fn is not None and not cap.match_fn(rep, resources):
+                continue
+            inst = instance_suffix(href)
+            # Auto-derive key prefix from href: strip digits, join non-empty segs with '_'
+            segs = [s for s in href.strip('/').split('/') if s and not s.isdigit() and s != 'vs']
+            for desc in cap.entities:
+                key_override = '_'.join(segs) + '_' + desc.key
+                out.append(BoundEntity(href=href, capability=cap, desc=desc,
+                                       instance=inst, key_override=key_override))
+            matched = True
+            break
+
+        if not matched and log is not None:
+            log(f"unknown resource {href}")
+
+    return out

custom_components/localthings/ocf/registry/entities.py

@@ -0,0 +1,79 @@
+"""HA-shaped entity descriptions. The subclass *type* selects the HA platform.
+
+Frozen dataclasses so the future native HA component can consume them as
+EntityDescription subclasses unchanged. Read transforms live in value_fn;
+presence gating in exists_fn; write logic in write_fn on command platforms.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, Optional
+
+WriteFn = Optional[Callable[[Any, dict], "tuple[list[str], dict] | None"]]
+
+
+def _identity(v: Any) -> Any:
+    return v
+
+
+@dataclass(frozen=True, kw_only=True)
+class SamsungEntityDescription:
+    key: str
+    field: str = ''
+    name: Optional[str] = None
+    translation_key: Optional[str] = None
+    icon: Optional[str] = None
+    entity_category: Optional[str] = None      # 'diagnostic' | 'config' | None
+    enabled_default: bool = True
+    value_fn: Callable[[Any], Any] = _identity
+    exists_fn: Optional[Callable[[dict], bool]] = None
+
+
+@dataclass(frozen=True, kw_only=True)
+class SensorDesc(SamsungEntityDescription):
+    device_class: Optional[str] = None
+    state_class: Optional[str] = None
+    unit: Optional[str] = None
+
+
+@dataclass(frozen=True, kw_only=True)
+class BinarySensorDesc(SamsungEntityDescription):
+    device_class: Optional[str] = None         # value_fn must return bool
+
+
+@dataclass(frozen=True, kw_only=True)
+class SelectDesc(SamsungEntityDescription):
+    options: Any = ()        # tuple[str,...] | Callable[[dict], list[str]]
+    write_fn: WriteFn = None
+
+
+@dataclass(frozen=True, kw_only=True)
+class SwitchDesc(SamsungEntityDescription):
+    device_class: Optional[str] = None
+    write_fn: WriteFn = None
+
+
+@dataclass(frozen=True, kw_only=True)
+class ButtonDesc(SamsungEntityDescription):
+    payload: str = ''
+    write_fn: WriteFn = None
+
+
+@dataclass(frozen=True, kw_only=True)
+class NumberDesc(SamsungEntityDescription):
+    device_class: Optional[str] = None
+    unit: Optional[str] = None
+    native_min: Optional[float] = None
+    native_max: Optional[float] = None
+    step: Optional[float] = None
+    write_fn: WriteFn = None
+
+
+PLATFORM_OF: dict[type, str] = {
+    SensorDesc:       'sensor',
+    BinarySensorDesc: 'binary_sensor',
+    SelectDesc:       'select',
+    SwitchDesc:       'switch',
+    ButtonDesc:       'button',
+    NumberDesc:       'number',
+}

custom_components/localthings/ocf/registry/identity.py

@@ -0,0 +1,37 @@
+"""Read device identity from standard OCF resources (/oic/p, /oic/d)."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+import cbor2
+
+
+@dataclass(frozen=True)
+class DeviceIdentity:
+    manufacturer: str
+    model: str
+    name: str
+    serial: Optional[str]
+
+
+def _get(sess, path) -> dict:
+    try:
+        code, pl = sess.get(path, timeout=10.0)
+        if code == 0x45 and pl:
+            body = cbor2.loads(pl)
+            return body if isinstance(body, dict) else {}
+    except Exception:
+        pass
+    return {}
+
+
+def read_identity(sess, serial: Optional[str]) -> DeviceIdentity:
+    p = _get(sess, ['oic', 'p'])
+    d = _get(sess, ['oic', 'd'])
+    return DeviceIdentity(
+        manufacturer=p.get('mnmn') or 'Samsung',
+        model=p.get('mnmo') or '',
+        name=d.get('n') or '',
+        serial=serial,
+    )

custom_components/localthings/ocf/registry/registry.py

@@ -0,0 +1,46 @@
+"""Global CAPABILITIES registry: href -> list[Capability].
+
+Built from the full list of Capability objects in the capabilities package.
+Consumed by discover() at connection time to bind device resources to entities.
+
+Raises ValueError at import if any href group contains an unfiltered cap
+alongside other caps (i.e., a cap with neither rt_filter nor match_fn set
+in a group with multiple caps).
+"""
+from .capabilities import ALL
+from .capability import Capability
+
+
+def _build() -> dict[str, list[Capability]]:
+    """Build the registry, raising ValueError for invalid duplicate hrefs.
+
+    Duplicates are allowed only when every cap in the group has at least one
+    of rt_filter or match_fn set. An unfiltered cap sharing an href with
+    another cap would be ambiguous.
+
+    Skips capabilities with href=None (pattern capabilities, handled elsewhere).
+    """
+    out: dict[str, list[Capability]] = {}
+    for cap in ALL:
+        if cap.href is None:
+            continue
+        if cap.href not in out:
+            out[cap.href] = [cap]
+        else:
+            out[cap.href].append(cap)
+
+    # Validate: every group with >1 cap must have all caps filtered
+    for href, caps in out.items():
+        if len(caps) > 1:
+            unfiltered = [c for c in caps
+                          if c.rt_filter is None and c.match_fn is None]
+            if unfiltered:
+                raise ValueError(
+                    f"duplicate capability href {href!r} has unfiltered cap(s); "
+                    f"each cap sharing an href must set rt_filter or match_fn"
+                )
+
+    return out
+
+
+CAPABILITIES: dict[str, list[Capability]] = _build()

custom_components/localthings/select.py

@@ -6,7 +6,7 @@ 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 .ocf.registry.entities import SelectDesc
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator

custom_components/localthings/sensor.py

@@ -6,7 +6,7 @@ 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 .ocf.registry.entities import SensorDesc
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator

custom_components/localthings/switch.py

@@ -6,7 +6,7 @@ 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 .ocf.registry.entities import SwitchDesc
 
 from .const import DOMAIN
 from .coordinator import LocalThingsCoordinator