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

feat: mint leaf certs from CA keypair, fix DTLS SSL level, polish entity model

Browse Source 
Config flow now accepts the AC14K_M CA cert+key, fetches the Samsung cloud
UUID from connect-v2.samsungiotcloud.com, mints a fresh RSA-2048 leaf cert
signed by the CA, and stores both the CA keypair and generated leaf for
reconnect. The DTLS context uses @SECLEVEL=0 in the cipher string — the only
channel that reaches cryptography's bundled OpenSSL — to permit the SHA-1
signed AC14K_M intermediate in Samsung's cert chain.

Entity model fixes:
- EntityCategory string values coerced to EntityCategory enum in entity.py
- SelectDesc gains options_field for live options from resource data; select
  registration skips entities where exists_fn rejects the initial resource
- NumberDesc gains range_field for live min/max from resource data
- Temperature sensors/numbers now carry device_class and unit (°F)
- Icemaker state sensor and on binary_sensor removed (duplicates of switch)
- Ice type select gated on x.com.samsung.da.iceType.supported presence
- Sabbath mode icon updated to mdi:hands-pray
This commit is contained in:
Marc Billow
2026-06-30 16:34:39 -05:00
parent f132d75f72
commit f25e616950
11 changed files with 278 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/config_flow.py
+144 -48
View File
@@ -1,8 +1,12 @@
"""Config flow for Local Things integration."""
from __future__ import annotations
import datetime
import logging
import os
import re
import socket
import ssl
import tempfile
from typing import Any
@@ -15,41 +19,135 @@ from homeassistant.helpers.selector import (
TextSelectorType,
)
from .const import DOMAIN, CONF_HOST, CONF_PORT, CONF_CERT_PEM, CONF_KEY_PEM, PROBE_PORTS
from .const import (
DOMAIN,
CONF_HOST, CONF_PORT,
CONF_CA_CERT_PEM, CONF_CA_KEY_PEM,
CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM,
PROBE_PORTS,
)
_TEXT = TextSelector(TextSelectorConfig(type=TextSelectorType.TEXT))
_MULTILINE = TextSelector(TextSelectorConfig(type=TextSelectorType.TEXT, multiline=True))
_LOGGER = logging.getLogger(__name__)
_SAMSUNG_CLOUD_HOST = 'connect-v2.samsungiotcloud.com'
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.
class InvalidCA(Exception):
pass
DtlsCoapSession only accepts file paths, so PEM strings are written
to a temp-file pair and cleaned up in the finally block.
def _fetch_samsung_uuid() -> str:
"""Connect to Samsung's cloud gateway and extract the UUID from its TLS cert.
Verification is disabled because Samsung's chain contains a self-signed cert.
We only need to read the UUID from the cert subject, not verify its trust.
"""
from cryptography import x509 as _x509
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with socket.create_connection((_SAMSUNG_CLOUD_HOST, 443), timeout=15) as raw:
with ctx.wrap_socket(raw, server_hostname=_SAMSUNG_CLOUD_HOST) as tls:
der = tls.getpeercert(binary_form=True)
cert = _x509.load_der_x509_certificate(der)
for attr in cert.subject:
if attr.oid == _x509.oid.NameOID.ORGANIZATIONAL_UNIT_NAME:
m = re.search(r'uuid:([0-9a-f-]+)', attr.value, re.IGNORECASE)
if m:
return m.group(1)
raise RuntimeError(f"UUID not found in {_SAMSUNG_CLOUD_HOST} certificate subject")
def _mint_leaf_cert(ca_cert_pem: str, ca_key_pem: str, uuid: str) -> tuple[str, str]:
"""Mint a fresh RSA-2048 leaf cert signed by the CA.
Returns (fullchain_pem, leaf_key_pem) where fullchain_pem is the leaf cert
followed by the full CA PEM, suitable for use_certificate_chain_file.
"""
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa as _rsa
m = re.search(
r'(-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----)',
ca_cert_pem, re.DOTALL,
)
if not m:
raise InvalidCA("No certificate found in CA cert PEM")
try:
ca_cert = x509.load_pem_x509_certificate(m.group(1).encode())
ca_key = serialization.load_pem_private_key(ca_key_pem.encode(), password=None)
except Exception as exc:
raise InvalidCA(f"Failed to load CA credentials: {exc}") from exc
leaf_key = _rsa.generate_private_key(public_exponent=65537, key_size=2048)
now = datetime.datetime.now(datetime.timezone.utc)
leaf_cert = (
x509.CertificateBuilder()
.subject_name(x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, 'KR'),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, 'Samsung Electronics'),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, f'uuid:{uuid}'),
x509.NameAttribute(NameOID.COMMON_NAME, f'urn:uuid:{uuid}'),
]))
.issuer_name(ca_cert.subject)
.public_key(leaf_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now)
.not_valid_after(now + datetime.timedelta(days=10 * 365))
.sign(ca_key, hashes.SHA256())
)
leaf_cert_pem = leaf_cert.public_bytes(serialization.Encoding.PEM).decode()
leaf_key_pem = leaf_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
).decode()
fullchain_pem = leaf_cert_pem + ca_cert_pem
return fullchain_pem, leaf_key_pem
def _probe_and_validate(host: str, ca_cert_pem: str, ca_key_pem: str) -> dict:
"""Fetch UUID, mint leaf cert, probe each port. Returns config entry data dict."""
import cbor2
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.
cert_file = tempfile.NamedTemporaryFile(
mode='w', suffix='.pem', delete=False)
key_file = tempfile.NamedTemporaryFile(
mode='w', suffix='.pem', delete=False)
_LOGGER.debug("Fetching Samsung cloud UUID from %s", _SAMSUNG_CLOUD_HOST)
try:
cert_file.write(cert_pem)
cert_file.flush()
cert_file.close()
key_file.write(key_pem)
key_file.flush()
key_file.close()
uuid = _fetch_samsung_uuid()
except Exception as exc:
_LOGGER.debug("UUID fetch failed: %s", exc, exc_info=True)
raise CannotConnect(f"Failed to fetch Samsung UUID: {exc}") from exc
_LOGGER.debug("Got UUID: %s", uuid)
_LOGGER.debug("Minting leaf cert for UUID %s", uuid)
try:
fullchain_pem, leaf_key_pem = _mint_leaf_cert(ca_cert_pem, ca_key_pem, uuid)
except InvalidCA:
_LOGGER.debug("CA credentials invalid", exc_info=True)
raise
except Exception as exc:
_LOGGER.debug("Leaf cert minting failed: %s", exc, exc_info=True)
raise CannotConnect(f"Failed to mint leaf cert: {exc}") from exc
_LOGGER.debug("Leaf cert minted successfully")
cert_file = tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False)
key_file = tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False)
try:
cert_file.write(fullchain_pem); cert_file.flush(); cert_file.close()
key_file.write(leaf_key_pem); key_file.flush(); key_file.close()
last_exc = None
for port in PROBE_PORTS:
@@ -64,28 +162,24 @@ def _probe_and_validate(host: str, cert_pem: str, key_pem: str) -> dict:
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}")
raise CannotConnect(f"port {port}: unexpected code {code:#04x}")
body = cbor2.loads(payload)
resources = (
parse_device0_batch(body) if isinstance(body, list)
else {}
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}
return {
"port": port,
"serial": serial,
"leaf_cert_pem": fullchain_pem,
"leaf_key_pem": leaf_key_pem,
}
except CannotConnect:
raise
except Exception as exc:
@@ -111,8 +205,8 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
self._host: str = ""
self._cert_pem: str = ""
self._key_pem: str = ""
self._ca_cert_pem: str = ""
self._ca_key_pem: str = ""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -125,47 +219,49 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
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]
self._ca_cert_pem = existing[0].data[CONF_CA_CERT_PEM]
self._ca_key_pem = existing[0].data[CONF_CA_KEY_PEM]
else:
self._cert_pem = user_input[CONF_CERT_PEM].strip()
self._key_pem = user_input[CONF_KEY_PEM].strip()
self._ca_cert_pem = user_input[CONF_CA_CERT_PEM].strip()
self._ca_key_pem = user_input[CONF_CA_KEY_PEM].strip()
try:
info = await self.hass.async_add_executor_job(
_probe_and_validate,
self._host,
self._cert_pem,
self._key_pem,
self._ca_cert_pem,
self._ca_key_pem,
)
except InvalidCA:
errors["base"] = "invalid_ca"
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']}"
)
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,
CONF_HOST: self._host,
CONF_PORT: info["port"],
CONF_CA_CERT_PEM: self._ca_cert_pem,
CONF_CA_KEY_PEM: self._ca_key_pem,
CONF_LEAF_CERT_PEM: info["leaf_cert_pem"],
CONF_LEAF_KEY_PEM: info["leaf_key_pem"],
},
)
if has_creds:
schema = vol.Schema({vol.Required(CONF_HOST): _TEXT})
reuse_note = "Certificate and key will be reused from the existing device."
reuse_note = "CA credentials will be reused from the existing device."
else:
schema = vol.Schema({
vol.Required(CONF_HOST): _TEXT,
vol.Required(CONF_CERT_PEM): _MULTILINE,
vol.Required(CONF_KEY_PEM): _MULTILINE,
vol.Required(CONF_HOST): _TEXT,
vol.Required(CONF_CA_CERT_PEM): _MULTILINE,
vol.Required(CONF_CA_KEY_PEM): _MULTILINE,
})
reuse_note = ""
custom_components/localthings/const.py
+6 -4
View File
@@ -2,10 +2,12 @@ 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"
CONF_HOST = "host"
CONF_PORT = "port"
CONF_CA_CERT_PEM = "ca_cert_pem"
CONF_CA_KEY_PEM = "ca_key_pem"
CONF_LEAF_CERT_PEM = "leaf_cert_pem"
CONF_LEAF_KEY_PEM = "leaf_key_pem"
PROBE_PORTS = [49154, 49155]
custom_components/localthings/coordinator.py
+7 -3
View File
@@ -23,7 +23,7 @@ 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,
DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM,
ACTIVE_INTERVAL_S, IDLE_INTERVAL_S,
)
@@ -63,11 +63,15 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Session management (all blocking — must run in executor)
# ------------------------------------------------------------------
@property
def last_resources(self) -> dict:
return self._last_resources
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_pem: str = self._entry.data[CONF_LEAF_CERT_PEM]
key_pem: str = self._entry.data[CONF_LEAF_KEY_PEM]
cert_f = tempfile.NamedTemporaryFile(suffix='.pem', delete=False, mode='w')
key_f = tempfile.NamedTemporaryFile(suffix='.pem', delete=False, mode='w')
custom_components/localthings/entity.py
+3 -1
View File
@@ -5,6 +5,7 @@ import re
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.const import EntityCategory
from .ocf.registry.adapter import _key
from .ocf.registry.discovery import BoundEntity
@@ -35,7 +36,8 @@ class LocalThingsEntity(CoordinatorEntity[LocalThingsCoordinator]):
self._attr_unique_id = f"{DOMAIN}_{coordinator.device_serial}_{self._state_key}"
self._attr_name = bound.desc.name if bound.desc.name is not None else _derive_name(self._state_key)
self._attr_icon = bound.desc.icon
self._attr_entity_category = bound.desc.entity_category
raw_cat = bound.desc.entity_category
self._attr_entity_category = EntityCategory(raw_cat) if raw_cat else None
self._attr_entity_registry_enabled_default = bound.desc.enabled_default
@property
custom_components/localthings/number.py
+22
View File
@@ -42,6 +42,28 @@ class LocalThingsNumber(LocalThingsEntity, NumberEntity):
if desc.step is not None:
self._attr_native_step = desc.step
def _range_from_resource(self) -> list | None:
desc: NumberDesc = self._bound.desc
if not desc.range_field:
return None
rep = self.coordinator.last_resources.get(self._bound.href) or {}
r = rep.get(desc.range_field)
return r if (isinstance(r, (list, tuple)) and len(r) == 2) else None
@property
def native_min_value(self) -> float:
r = self._range_from_resource()
if r is not None:
return float(r[0])
return self._attr_native_min_value
@property
def native_max_value(self) -> float:
r = self._range_from_resource()
if r is not None:
return float(r[1])
return self._attr_native_max_value
@property
def native_value(self):
return (self.coordinator.data or {}).get(self._state_key)
custom_components/localthings/ocf/coap_dtls.py
+48 -68
View File
@@ -31,57 +31,6 @@ _OCF_ROOT_CA = str(Path(__file__).parent / 'ocf_root_ca.pem')
import logging
logger = logging.getLogger(__name__)
def _set_security_level(ctx: SSL.Context, level: int) -> None:
"""Call SSL_CTX_set_security_level on an OpenSSL context.
pyOpenSSL >= 24.0.0 exposes this directly; older builds require a cffi
call into the underlying cryptography bindings.
"""
# pyOpenSSL >= 24.0.0
if hasattr(ctx, 'set_security_level'):
logger.debug("_set_security_level: ctx.set_security_level(%d)", level)
ctx.set_security_level(level)
return
# cryptography/cffi bindings
try:
from OpenSSL._util import lib as _lib
if hasattr(_lib, 'SSL_CTX_set_security_level'):
logger.debug("_set_security_level: cffi SSL_CTX_set_security_level(%d)", level)
_lib.SSL_CTX_set_security_level(ctx._context, level)
return
logger.debug("_set_security_level: cffi binding missing SSL_CTX_set_security_level")
except Exception as exc:
logger.debug("_set_security_level: cffi attempt failed: %s", exc)
# ctypes — None means RTLD_DEFAULT (already-loaded libs in this process),
# which finds the same OpenSSL instance pyOpenSSL is using. Named paths
# open a fresh handle that may be a different instance (wrong ctx space).
import ctypes, ctypes.util
from OpenSSL._util import ffi as _ffi
raw_ptr = int(_ffi.cast('uintptr_t', ctx._context))
logger.debug("_set_security_level: ctx._context=0x%x", raw_ptr)
for _handle in (None, 'libssl.so.3', 'libssl.so.1.1',
ctypes.util.find_library('ssl')):
if _handle is None:
_desc = 'RTLD_DEFAULT'
else:
_desc = _handle
try:
_lib2 = ctypes.CDLL(_handle)
_lib2.SSL_CTX_set_security_level.restype = None
_lib2.SSL_CTX_set_security_level.argtypes = [ctypes.c_void_p, ctypes.c_int]
_lib2.SSL_CTX_set_security_level(raw_ptr, level)
logger.debug("_set_security_level: ctypes via %s succeeded", _desc)
return
except Exception as exc:
logger.debug("_set_security_level: ctypes via %s failed: %s", _desc, exc)
logger.warning(
"_set_security_level: all approaches failed — "
"SHA-1 cert chain will likely be rejected by OpenSSL"
)
# 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
@@ -279,24 +228,14 @@ class DtlsCoapSession:
ConnectionError / TimeoutError on failure."""
ctx = SSL.Context(SSL.DTLS_METHOD)
# The AC14K_M intermediate CA in our client cert chain is SHA-1
# signed; Samsung's OCF Root CA is SHA-1 self-signed. OpenSSL 3.x
# SECLEVEL=2 rejects SHA-1 in any cert it processes — including
# the LOCAL chain we're sending — raising SSL_R_CA_MD_TOO_WEAK
# before the verify callback is ever reached.
# Must be called before loading any certs.
_set_security_level(ctx, 1)
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')
# use_certificate_file loads only the leaf cert (first PEM block).
# Sending the full chain via use_certificate_chain_file causes
# OpenSSL 3.x to reject the AC14K_M intermediate (SHA-1 signed)
# with SSL_R_CA_MD_TOO_WEAK before the handshake even starts.
# Samsung's device already has AC14K_M in its own trust store,
# so it can verify the leaf cert without us sending the chain.
ctx.use_certificate_file(self.cert_path, SSL.FILETYPE_PEM)
# @SECLEVEL=0 permits SHA-1 in Samsung's server cert chain (AC14K_M
# intermediate is SHA-1 signed). This is the only channel that reaches
# the OpenSSL instance cryptography bundles — ctypes and cffi bindings
# do not expose SSL_CTX_set_security_level on this build.
ctx.set_cipher_list(b'ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0')
ctx.use_certificate_chain_file(self.cert_path)
ctx.use_privatekey_file(self.key_path)
ctx.check_privatekey()
@@ -307,15 +246,25 @@ class DtlsCoapSession:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(2.0)
dest = (self.host, self.port)
logger.debug(
"DTLS handshake starting to %s:%d (timeout=%.1fs)",
self.host, self.port, self.HANDSHAKE_TIMEOUT_S,
)
t0 = time.time()
iterations = 0
while time.time() - t0 < self.HANDSHAKE_TIMEOUT_S:
iterations += 1
try:
conn.do_handshake()
logger.debug(
"DTLS handshake succeeded after %d iterations", iterations,
)
break
except SSL.WantReadError:
pass
except SSL.Error as e:
logger.debug("DTLS SSL.Error on iter %d: %s", iterations, e)
sock.close()
raise ConnectionError(f"DTLS handshake error: {e}") from e
try:
@@ -333,6 +282,10 @@ class DtlsCoapSession:
pass
time.sleep(0.05)
else:
logger.debug(
"DTLS handshake timeout after %d iterations, elapsed=%.2fs",
iterations, time.time() - t0,
)
sock.close()
raise TimeoutError(
f"DTLS handshake timeout to {self.host}:{self.port}")
@@ -435,8 +388,8 @@ class DtlsCoapSession:
with self._send_lock:
if self.conn is None:
raise ConnectionError("DTLS session closed")
self.conn.send(datagram)
try:
self.conn.send(datagram)
while True:
o = self.conn.bio_read(65535)
if not o:
@@ -564,6 +517,10 @@ class DtlsCoapSession:
return
# Stale token (post-reconnect or unknown) — drop quietly.
logger.debug(
"stale CoAP response: code=%s tok=%s mid=%04x",
fmt_code(code), tok.hex() or '-', mid,
)
# ---- request primitives ------------------------------------------
@@ -577,6 +534,10 @@ class DtlsCoapSession:
if self.conn is None:
raise ConnectionError("DTLS session closed")
tok = self._next_tok()
logger.debug(
"GET /%s token=%s timeout=%.1fs",
'/'.join(path_segs), tok.hex(), timeout,
)
blob = b''
num = 0
last_code = None
@@ -598,8 +559,27 @@ class DtlsCoapSession:
build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
wait = max(0.1, deadline - time.time())
if not ev.wait(wait):
logger.debug(
"GET /%s block %d: timed out after %.2fs",
'/'.join(path_segs), num, wait,
)
raise TimeoutError(
f"GET /{'/'.join(path_segs)} block {num} timeout")
code_seen = container.get('code', 0)
# 4.01 = OCF cert/ACL rejection — surface at WARNING
# so it's visible without debug logging.
if code_seen == 0x81:
logger.warning(
"GET /%s → 4.01 Unauthorized",
'/'.join(path_segs),
)
else:
logger.debug(
"GET /%s block %d: %s (%d bytes)",
'/'.join(path_segs), num,
fmt_code(code_seen),
len(container.get('payload', b'')),
)
if 'err' in container:
raise ConnectionError(container['err'])
finally:
custom_components/localthings/ocf/registry/capabilities/fridge.py
+9 -9
View File
@@ -37,7 +37,9 @@ TEMP_CURRENT_GENERIC = Capability(
poll_tier='warm',
entities=(
SensorDesc(key='temp_f', field='temperature',
name=None, icon='mdi:thermometer'),
name=None, icon='mdi:thermometer',
device_class='temperature', unit='°F',
state_class='measurement'),
),
)
@@ -47,7 +49,9 @@ TEMP_SETPOINT_GENERIC = Capability(
poll_tier='warm',
entities=(
NumberDesc(key='temp_f', field='temperature',
name=None, native_min=-20.0, native_max=50.0,
name=None, device_class='temperature', unit='°F',
native_min=-20.0, native_max=50.0,
range_field='range',
write_fn=lambda p, rep, href=None: (
[s for s in href.strip('/').split('/') if s],
{'temperature': int(round(float(p)))}
@@ -73,11 +77,6 @@ ICEMAKER_GENERIC = Capability(
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'),
@@ -87,7 +86,8 @@ ICEMAKER_GENERIC = Capability(
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=(),
options_field='x.com.samsung.da.iceType.supported',
exists_fn=lambda rep: bool(rep.get('x.com.samsung.da.iceType.supported')),
write_fn=_icemaker_write('x.com.samsung.da.iceType.desired')),
),
)
@@ -217,7 +217,7 @@ SABBATH = Capability(
poll_tier='warm',
entities=(
SwitchDesc(key='sabbath_mode', field='x.com.samsung.da.sabbathMode',
name='Sabbath mode', icon='mdi:star-david',
name='Sabbath mode', icon='mdi:hands-pray',
value_fn=lambda v: v == 'On',
write_fn=_sabbath_write),
),
custom_components/localthings/ocf/registry/entities.py
+2
View File
@@ -44,6 +44,7 @@ class BinarySensorDesc(SamsungEntityDescription):
@dataclass(frozen=True, kw_only=True)
class SelectDesc(SamsungEntityDescription):
options: Any = () # tuple[str,...] | Callable[[dict], list[str]]
options_field: Optional[str] = None # resource field that contains the live options list
write_fn: WriteFn = None
@@ -66,6 +67,7 @@ class NumberDesc(SamsungEntityDescription):
native_min: Optional[float] = None
native_max: Optional[float] = None
step: Optional[float] = None
range_field: Optional[str] = None # resource field containing [min, max] list
write_fn: WriteFn = None
custom_components/localthings/select.py
+23 -3
View File
@@ -19,10 +19,21 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][entry.entry_id]
def _include(b) -> bool:
if not isinstance(b.desc, SelectDesc):
return False
# Skip at registration time if exists_fn says the entity isn't present
if b.desc.exists_fn is not None:
rep = coordinator.last_resources.get(b.href) or {}
if not b.desc.exists_fn(rep):
return False
return True
async_add_entities(
LocalThingsSelect(coordinator, b)
for b in coordinator.bound
if isinstance(b.desc, SelectDesc)
if _include(b)
)
@@ -31,8 +42,17 @@ 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)
if not desc.options_field:
opts = desc.options
self._attr_options = list(opts() if callable(opts) else opts)
@property
def options(self) -> list[str]:
desc: SelectDesc = self._bound.desc
if desc.options_field:
rep = self.coordinator.last_resources.get(self._bound.href) or {}
return list(rep.get(desc.options_field) or [])
return self._attr_options
@property
def current_option(self):
custom_components/localthings/strings.json
+7 -6
View File
@@ -3,21 +3,22 @@
"step": {
"user": {
"title": "Add Samsung Appliance",
"description": "Enter the IP address of your Samsung appliance on the local network. {reuse_note}",
"description": "Enter the IP address of your Samsung appliance and the AC14K_M CA credentials used to mint device certificates. {reuse_note}",
"data": {
"host": "IP Address",
"cert_pem": "Certificate (PEM)",
"key_pem": "Private Key (PEM)"
"ca_cert_pem": "CA Certificate (PEM)",
"ca_key_pem": "CA Private Key (PEM)"
},
"data_description": {
"host": "Local IP address of your Samsung appliance (e.g. 192.168.1.50).",
"cert_pem": "The PEM-encoded client certificate used to authenticate with the appliance. Paste the full contents including the BEGIN/END CERTIFICATE header and footer.",
"key_pem": "The PEM-encoded private key matching the certificate above. Paste the full contents including the BEGIN/END PRIVATE KEY header and footer."
"ca_cert_pem": "The PEM-encoded AC14K_M CA certificate chain. Paste the full contents of your fullchain PEM including all BEGIN/END CERTIFICATE blocks.",
"ca_key_pem": "The PEM-encoded AC14K_M private key. Paste the full contents including the BEGIN/END PRIVATE KEY header and footer."
}
}
},
"error": {
"cannot_connect": "Cannot connect to the device. Verify the IP address is reachable and the certificate and key are correct.",
"cannot_connect": "Cannot connect to the device. Verify the IP address is reachable and the CA credentials are correct.",
"invalid_ca": "The CA certificate or private key could not be loaded. Verify the PEM contents are correct and the key matches the certificate.",
"unknown": "Unexpected error. Check the Home Assistant logs for details."
},
"abort": {
custom_components/localthings/translations/en.json
+7 -6
View File
@@ -3,21 +3,22 @@
"step": {
"user": {
"title": "Add Samsung Appliance",
"description": "Enter the IP address of your Samsung appliance on the local network. {reuse_note}",
"description": "Enter the IP address of your Samsung appliance and the AC14K_M CA credentials used to mint device certificates. {reuse_note}",
"data": {
"host": "IP Address",
"cert_pem": "Certificate (PEM)",
"key_pem": "Private Key (PEM)"
"ca_cert_pem": "CA Certificate (PEM)",
"ca_key_pem": "CA Private Key (PEM)"
},
"data_description": {
"host": "Local IP address of your Samsung appliance (e.g. 192.168.1.50).",
"cert_pem": "The PEM-encoded client certificate used to authenticate with the appliance. Paste the full contents including the BEGIN/END CERTIFICATE header and footer.",
"key_pem": "The PEM-encoded private key matching the certificate above. Paste the full contents including the BEGIN/END PRIVATE KEY header and footer."
"ca_cert_pem": "The PEM-encoded AC14K_M CA certificate chain. Paste the full contents of your fullchain PEM including all BEGIN/END CERTIFICATE blocks.",
"ca_key_pem": "The PEM-encoded AC14K_M private key. Paste the full contents including the BEGIN/END PRIVATE KEY header and footer."
}
}
},
"error": {
"cannot_connect": "Cannot connect to the device. Verify the IP address is reachable and the certificate and key are correct.",
"cannot_connect": "Cannot connect to the device. Verify the IP address is reachable and the CA credentials are correct.",
"invalid_ca": "The CA certificate or private key could not be loaded. Verify the PEM contents are correct and the key matches the certificate.",
"unknown": "Unexpected error. Check the Home Assistant logs for details."
},
"abort": {