Add Hot Spring Connected Spa integration

Unofficial cloud integration for Watkins/Hot Spring "Connected Spa" hot tubs.
Reverse-engineered from the official Android app.

- climate entity (setpoint + current water temp) via heater/control
- sensors: water temperature, salt output level, salt cartridge
- REST (JWT) login + spa discovery; mTLS fetch of rotating HiveMQ broker creds;
  MQTT subscribe (telemetry) / publish (control)
- config flow prompts for account email + password (stored encrypted, never in repo)
- no secrets committed: vendor mTLS client cert is loaded from host PEM files
  (see README); .gitignore blocks pem/p12/der/key/env

Fixes bug-41mqxddz7zeh

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01F6B8b3iYNv6QUftK2FDfYb

custom_components/hotspring/__init__.py

@@ -0,0 +1,36 @@
+"""The Hot Spring Connected Spa integration."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .const import DOMAIN
+from .coordinator import HotSpringCoordinator
+
+PLATFORMS: list[Platform] = [Platform.CLIMATE, Platform.SENSOR]
+
+type HotSpringConfigEntry = ConfigEntry[HotSpringCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: HotSpringConfigEntry) -> bool:
+    coordinator = HotSpringCoordinator(hass, entry)
+    try:
+        await coordinator.async_setup()
+    except Exception as err:  # noqa: BLE001 - retried by HA as not-ready
+        raise ConfigEntryNotReady(f"Hot Spring setup failed: {err}") from err
+
+    entry.runtime_data = coordinator
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: HotSpringConfigEntry) -> bool:
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        coordinator: HotSpringCoordinator = hass.data[DOMAIN].pop(entry.entry_id)
+        await coordinator.async_shutdown()
+    return unload_ok

custom_components/hotspring/api/__init__.py

@@ -0,0 +1,19 @@
+"""Reverse-engineered Hot Spring / Watkins Connected Spa client (cloud + MQTT)."""
+
+from .cloud import (
+    HotSpringApiError,
+    HotSpringAuthError,
+    HotSpringCloud,
+    MqttCredentials,
+    Spa,
+)
+from .spa import HotSpringSpa
+
+__all__ = [
+    "HotSpringCloud",
+    "HotSpringSpa",
+    "MqttCredentials",
+    "Spa",
+    "HotSpringAuthError",
+    "HotSpringApiError",
+]

custom_components/hotspring/api/cloud.py

@@ -0,0 +1,126 @@
+"""Cloud (REST) client: authenticate, discover spas, fetch live MQTT creds.
+
+Synchronous + dependency-free (``urllib`` + ``ssl``). Home Assistant calls these
+from the executor (they perform blocking network I/O), keeping the event loop
+free. See ``coordinator.py``.
+"""
+
+from __future__ import annotations
+
+import json
+import ssl
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from typing import Optional
+
+from . import protocol
+
+
+class HotSpringAuthError(Exception):
+    """Login / token failure (surfaced as an auth error in the config flow)."""
+
+
+class HotSpringApiError(Exception):
+    """Any other API / transport failure."""
+
+
+@dataclass
+class Spa:
+    name: str
+    root_topic: str
+    raw: dict
+
+
+@dataclass
+class MqttCredentials:
+    url: str
+    port: int
+    user: str
+    password: str
+    raw: dict
+
+
+class HotSpringCloud:
+    def __init__(self, email, password, client_cert, client_key, ca_cert, timeout=25):
+        self._email = email
+        self._password = password
+        self._client_cert = client_cert
+        self._client_key = client_key
+        self._ca_cert = ca_cert
+        self._timeout = timeout
+        self.access: Optional[str] = None
+        self.refresh: Optional[str] = None
+
+    def _request(self, url, *, method="GET", body=None, bearer=None, ssl_ctx=None):
+        headers = {"Accept": "application/json"}
+        data = None
+        if body is not None:
+            data = json.dumps(body).encode()
+            headers["Content-Type"] = "application/json"
+        if bearer:
+            headers["Authorization"] = "Bearer " + bearer
+        req = urllib.request.Request(url, data=data, headers=headers, method=method)
+        try:
+            with urllib.request.urlopen(req, timeout=self._timeout, context=ssl_ctx) as r:
+                return r.status, r.read().decode()
+        except urllib.error.HTTPError as e:
+            return e.code, e.read().decode()
+        except (urllib.error.URLError, OSError) as e:
+            raise HotSpringApiError(str(e)) from e
+
+    def login(self) -> None:
+        status, raw = self._request(
+            protocol.API_BASE + protocol.LOGIN_PATH,
+            method="POST",
+            body={"email": self._email, "password": self._password},
+        )
+        if status in (400, 401, 403):
+            raise HotSpringAuthError(f"login rejected (HTTP {status})")
+        if status != 200:
+            raise HotSpringApiError(f"login failed: HTTP {status}: {raw[:200]}")
+        data = json.loads(raw)
+        self.access = data.get("access")
+        self.refresh = data.get("refresh")
+        if not self.access:
+            raise HotSpringAuthError("login returned no access token")
+
+    def refresh_access(self) -> None:
+        if not self.refresh:
+            raise HotSpringAuthError("no refresh token; login first")
+        status, raw = self._request(
+            protocol.API_BASE + protocol.REFRESH_PATH,
+            method="POST",
+            body={"refresh": self.refresh},
+        )
+        if status != 200:
+            raise HotSpringAuthError(f"refresh failed: HTTP {status}")
+        self.access = json.loads(raw).get("access", self.access)
+
+    def get_spas(self) -> list[Spa]:
+        if not self.access:
+            raise HotSpringAuthError("not authenticated")
+        status, raw = self._request(
+            protocol.API_BASE + protocol.SPA_DETAILS_PATH, bearer=self.access
+        )
+        if status != 200:
+            raise HotSpringApiError(f"spa details failed: HTTP {status}")
+        items = json.loads(raw)
+        if isinstance(items, dict):
+            items = items.get("data") or items.get("spas") or [items]
+        return [Spa(s.get("spaName", "Spa"), s["rootTopic"], s) for s in items]
+
+    def fetch_mqtt_credentials(self) -> MqttCredentials:
+        """mTLS GET returning the rotating HiveMQ broker host/user/pass."""
+        ctx = ssl.create_default_context(cafile=self._ca_cert)
+        # The firmware CA is a self-signed cert lacking the modern CA
+        # extensions Python 3.12+ requires under VERIFY_X509_STRICT. The app
+        # pins this exact CA + hostname; we keep CA + hostname verification and
+        # only drop the strict-extension check.
+        ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT
+        ctx.load_cert_chain(self._client_cert, self._client_key)
+        status, raw = self._request(protocol.MQTT_CREDS_URL, ssl_ctx=ctx)
+        if status != 200:
+            raise HotSpringApiError(f"mqtt creds failed: HTTP {status}")
+        d = json.loads(raw)
+        return MqttCredentials(d["url"], int(d.get("port", 8883)), d["user"], d["pass"], d)

custom_components/hotspring/api/protocol.py

@@ -0,0 +1,54 @@
+"""Vendor protocol facts + command builders for the Watkins / Hot Spring
+"Connected Spa" cloud.
+
+Reverse-engineered from the official Android app (``org.watkins.hotspring``).
+This module holds every hard-coded vendor value in one auditable place and
+contains no secrets: the account credentials come from the config entry, and the
+mutual-TLS client certificate is loaded from files on the host (see the README).
+"""
+
+from __future__ import annotations
+
+# --- REST (Django SimpleJWT) -------------------------------------------------
+API_BASE = "https://iotsupportportalapi.watkinsmfg.com"
+LOGIN_PATH = "/api/v1/login/"
+REFRESH_PATH = "/api/v1/refresh/"
+SPA_DETAILS_PATH = "/user_spa_details/"
+
+# --- MQTT credential endpoint (mutual TLS) -----------------------------------
+# Server cert is signed by the firmware CA bundled in the app; verified against
+# the CA PEM provided on the host rather than the public trust store.
+MQTT_CREDS_URL = "https://iotfirmwareota.watkinsmfg.com:8577/"
+
+# --- Topics ------------------------------------------------------------------
+STATUS_WILDCARD = "{root}/#"
+CONTROL_TOPIC = "{root}/{subsystem}/control"
+
+
+def build_control_payload(subsystem: str, control: dict) -> dict:
+    """Wrap a control dict the way every command in the app does.
+
+    ``build_control_payload("heater", {"temperatureABS": "104"})``
+    -> ``{"heater": {"control": {"temperatureABS": "104"}}}``
+    """
+    return {subsystem: {"control": control}}
+
+
+# High-level command factories -> (subsystem_path, control_dict).
+# Temperature is an integer string in the spa's configured unit (the app sends
+# ``String.valueOf((int) temp)``).
+
+def cmd_set_temperature(value: int):
+    return "heater", {"temperatureABS": str(int(value))}
+
+
+def cmd_nudge_temperature(up: bool):
+    return "heater", {"temperatureControl": "temp_up" if up else "temp_down"}
+
+
+def cmd_set_water_care_level(level: int):
+    return "waterCare", {"level": str(int(level))}
+
+
+def cmd_toggle_water_care_boost():
+    return "waterCare", {"boost": "toggle"}

custom_components/hotspring/api/spa.py

@@ -0,0 +1,171 @@
+"""MQTT client for one spa: subscribe to telemetry, publish control commands.
+
+Uses paho-mqtt (bundled with Home Assistant). paho runs its network loop in its
+own thread; telemetry is pushed out via the ``on_status`` callback, which the
+coordinator bridges back onto the HA event loop.
+"""
+
+from __future__ import annotations
+
+import json
+import ssl
+import threading
+import time
+from typing import Callable, Optional
+
+import paho.mqtt.client as mqtt
+
+from . import protocol
+from .cloud import MqttCredentials
+
+
+def _new_client(client_id: str) -> mqtt.Client:
+    """Create an MQTTv5 client compatible with paho-mqtt 1.x and 2.x.
+
+    paho 2.0 requires the callback API version explicitly; the VERSION1
+    callback signatures used here match what this module implements.
+    """
+    try:
+        from paho.mqtt.enums import CallbackAPIVersion
+
+        return mqtt.Client(
+            CallbackAPIVersion.VERSION1,
+            client_id=client_id,
+            protocol=mqtt.MQTTv5,
+        )
+    except ImportError:  # paho-mqtt < 2.0
+        return mqtt.Client(client_id=client_id, protocol=mqtt.MQTTv5)
+
+
+class HotSpringSpa:
+    def __init__(
+        self,
+        root_topic: str,
+        creds: MqttCredentials,
+        on_status: Optional[Callable[[str, dict], None]] = None,
+    ):
+        self.root_topic = root_topic
+        self._creds = creds
+        self._on_status = on_status
+        self.state: dict[str, dict] = {}   # topic-suffix -> payload
+        self.merged: dict = {}             # deep-merged view for convenient lookups
+        self._client: Optional[mqtt.Client] = None
+        self._connected = threading.Event()
+
+    # -- connection -----------------------------------------------------------
+    def connect(self, keepalive: int = 30, wait: float = 15.0) -> None:
+        cl = _new_client(f"ha-hotspring-{int(time.time())}")
+        cl.username_pw_set(self._creds.user, self._creds.password)
+        cl.tls_set_context(ssl.create_default_context())  # broker uses a public CA
+        cl.on_connect = self._on_connect
+        cl.on_message = self._on_message
+        cl.reconnect_delay_set(min_delay=1, max_delay=60)
+        self._client = cl
+        cl.connect(self._creds.url, self._creds.port, keepalive=keepalive)
+        cl.loop_start()
+        if not self._connected.wait(wait):
+            raise TimeoutError("MQTT connect timed out")
+
+    def disconnect(self) -> None:
+        if self._client:
+            self._client.loop_stop()
+            self._client.disconnect()
+
+    @property
+    def connected(self) -> bool:
+        return bool(self._client) and self._connected.is_set()
+
+    def wait_for_state(self, timeout: float = 8.0) -> None:
+        """Block until the heater setpoint arrives (retained telemetry)."""
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if self.setpoint() is not None:
+                return
+            time.sleep(0.2)
+
+    # -- paho callbacks -------------------------------------------------------
+    def _on_connect(self, client, userdata, flags, rc, props=None):
+        if rc == 0:
+            client.subscribe(
+                protocol.STATUS_WILDCARD.format(root=self.root_topic), qos=1
+            )
+            self._connected.set()
+
+    def _on_message(self, client, userdata, msg):
+        suffix = msg.topic[len(self.root_topic) + 1:]
+        try:
+            payload = json.loads(msg.payload.decode("utf-8", "replace"))
+        except ValueError:
+            return
+        self.state[suffix] = payload
+        _deep_merge(self.merged, payload)
+        if self._on_status:
+            self._on_status(suffix, payload)
+
+    # -- parsed reads ---------------------------------------------------------
+    def _heater(self) -> dict:
+        return (self.merged.get("heater") or {}).get("status") or {}
+
+    def _fwss(self) -> dict:
+        return (((self.merged.get("waterCare") or {}).get("status") or {})
+                .get("FWSSstatus") or {})
+
+    def setpoint(self):
+        return _to_number(self._heater().get("setWaterTemperature"))
+
+    def current_temperature(self):
+        return _to_number(self._heater().get("currentWaterTemperature"))
+
+    def heater_on(self):
+        return self._heater().get("heater") == "on"
+
+    def temperature_unit(self):
+        return self._heater().get("temperatureUnit")  # "DegF" / "DegC"
+
+    def salt_output_level(self):
+        return _to_number(self._fwss().get("Outputlevel"))
+
+    def cartridge_installed(self):
+        v = self._fwss().get("cartridgeInstalled")
+        return None if v is None else (v == "installed")
+
+    # -- writes ---------------------------------------------------------------
+    def publish_control(self, subsystem: str, control: dict, qos: int = 1) -> None:
+        if not self._client:
+            raise RuntimeError("not connected")
+        top = subsystem.split("/")[0]
+        topic = protocol.CONTROL_TOPIC.format(root=self.root_topic, subsystem=subsystem)
+        body = json.dumps(protocol.build_control_payload(top, control))
+        self._client.publish(topic, body.encode(), qos=qos)
+
+    def set_temperature(self, value: int):
+        self.publish_control(*protocol.cmd_set_temperature(value))
+
+    def nudge_temperature(self, up: bool):
+        self.publish_control(*protocol.cmd_nudge_temperature(up))
+
+    def set_water_care_level(self, level: int):
+        self.publish_control(*protocol.cmd_set_water_care_level(level))
+
+    def toggle_water_care_boost(self):
+        self.publish_control(*protocol.cmd_toggle_water_care_boost())
+
+
+def _to_number(v):
+    if v is None:
+        return None
+    if isinstance(v, (int, float)):
+        return float(v)
+    s = "".join(ch for ch in str(v) if ch.isdigit() or ch in ".-")
+    try:
+        return float(s) if s else None
+    except ValueError:
+        return None
+
+
+def _deep_merge(dst: dict, src: dict) -> None:
+    for k, v in src.items():
+        if isinstance(v, dict) and isinstance(dst.get(k), dict):
+            _deep_merge(dst[k], v)
+        else:
+            dst[k] = v

custom_components/hotspring/climate.py

@@ -0,0 +1,88 @@
+"""Climate entity: the spa's water heater setpoint + current temperature."""
+
+from __future__ import annotations
+
+from homeassistant.components.climate import (
+    ClimateEntity,
+    ClimateEntityFeature,
+    HVACMode,
+)
+from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import HotSpringConfigEntry
+from .const import DOMAIN, MANUFACTURER
+from .coordinator import HotSpringCoordinator
+
+# Hot Spot spas run 80-104 F.
+MIN_TEMP_F = 80
+MAX_TEMP_F = 104
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: HotSpringConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    async_add_entities([HotSpringClimate(entry.runtime_data)])
+
+
+class HotSpringClimate(CoordinatorEntity[HotSpringCoordinator], ClimateEntity):
+    _attr_has_entity_name = True
+    _attr_name = None  # use the device name
+    _attr_hvac_modes = [HVACMode.HEAT]
+    _attr_hvac_mode = HVACMode.HEAT
+    _attr_supported_features = ClimateEntityFeature.TARGET_TEMPERATURE
+    _attr_target_temperature_step = 1
+    _enable_turn_on_off_backwards_compatibility = False
+
+    def __init__(self, coordinator: HotSpringCoordinator) -> None:
+        super().__init__(coordinator)
+        root = coordinator.spa_info.root_topic if coordinator.spa_info else "spa"
+        self._attr_unique_id = f"{root}_climate"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, root)},
+            manufacturer=MANUFACTURER,
+            name=coordinator.spa_info.name if coordinator.spa_info else "Hot Spring Spa",
+            model="Connected Spa",
+        )
+
+    @property
+    def _celsius(self) -> bool:
+        return (self.coordinator.spa and self.coordinator.spa.temperature_unit()) == "DegC"
+
+    @property
+    def temperature_unit(self) -> str:
+        return UnitOfTemperature.CELSIUS if self._celsius else UnitOfTemperature.FAHRENHEIT
+
+    @property
+    def min_temp(self) -> float:
+        return MIN_TEMP_F if not self._celsius else round((MIN_TEMP_F - 32) / 1.8)
+
+    @property
+    def max_temp(self) -> float:
+        return MAX_TEMP_F if not self._celsius else round((MAX_TEMP_F - 32) / 1.8)
+
+    @property
+    def current_temperature(self) -> float | None:
+        return self.coordinator.spa.current_temperature() if self.coordinator.spa else None
+
+    @property
+    def target_temperature(self) -> float | None:
+        return self.coordinator.spa.setpoint() if self.coordinator.spa else None
+
+    async def async_set_temperature(self, **kwargs) -> None:
+        temp = kwargs.get(ATTR_TEMPERATURE)
+        if temp is None:
+            return
+        # The spa expects an integer in its configured unit.
+        await self.coordinator.async_publish_control(
+            "heater", {"temperatureABS": str(int(round(temp)))}
+        )
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        self.async_write_ha_state()

custom_components/hotspring/config_flow.py

@@ -0,0 +1,108 @@
+"""Config flow: prompt for the Hot Spring account e-mail + password.
+
+Credentials are validated by logging in, then stored in the config entry (Home
+Assistant encrypts entry storage at rest). They are never written to this repo.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers.selector import (
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .api import HotSpringApiError, HotSpringAuthError, HotSpringCloud
+from .const import (
+    CA_FILE,
+    CERT_FILE,
+    CONF_CERT_DIR,
+    CONF_EMAIL,
+    CONF_PASSWORD,
+    DEFAULT_CERT_DIR,
+    DOMAIN,
+    KEY_FILE,
+)
+
+
+def _schema(defaults: dict[str, Any] | None = None) -> vol.Schema:
+    defaults = defaults or {}
+    return vol.Schema(
+        {
+            vol.Required(CONF_EMAIL, default=defaults.get(CONF_EMAIL, "")): str,
+            vol.Required(CONF_PASSWORD): TextSelector(
+                TextSelectorConfig(type=TextSelectorType.PASSWORD)
+            ),
+            vol.Optional(
+                CONF_CERT_DIR, default=defaults.get(CONF_CERT_DIR, DEFAULT_CERT_DIR)
+            ): str,
+        }
+    )
+
+
+class HotSpringConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle the UI setup."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            cert_dir = user_input[CONF_CERT_DIR]
+            missing = [
+                f
+                for f in (CERT_FILE, KEY_FILE, CA_FILE)
+                if not os.path.isfile(os.path.join(cert_dir, f))
+            ]
+            if missing:
+                errors["base"] = "missing_certs"
+            else:
+                result = await self._validate(user_input)
+                if result is None:
+                    spa_name = self._spa_name or "Hot Spring Spa"
+                    await self.async_set_unique_id(self._unique_id)
+                    self._abort_if_unique_id_configured()
+                    return self.async_create_entry(title=spa_name, data=user_input)
+                errors["base"] = result
+
+        return self.async_show_form(
+            step_id="user", data_schema=_schema(user_input), errors=errors
+        )
+
+    async def _validate(self, data: dict[str, Any]) -> str | None:
+        """Return None on success, or an error key."""
+        cert_dir = data[CONF_CERT_DIR]
+        cloud = HotSpringCloud(
+            data[CONF_EMAIL],
+            data[CONF_PASSWORD],
+            os.path.join(cert_dir, CERT_FILE),
+            os.path.join(cert_dir, KEY_FILE),
+            os.path.join(cert_dir, CA_FILE),
+        )
+
+        def _check() -> tuple[str | None, str | None]:
+            cloud.login()
+            spas = cloud.get_spas()
+            if not spas:
+                return None, None
+            return spas[0].name, spas[0].root_topic
+
+        try:
+            self._spa_name, self._unique_id = await self.hass.async_add_executor_job(
+                _check
+            )
+        except HotSpringAuthError:
+            return "invalid_auth"
+        except HotSpringApiError:
+            return "cannot_connect"
+        except Exception:  # noqa: BLE001 - surface as generic to the user
+            return "unknown"
+        return None

custom_components/hotspring/const.py

@@ -0,0 +1,23 @@
+"""Constants for the Hot Spring Connected Spa integration."""
+
+from __future__ import annotations
+
+DOMAIN = "hotspring"
+
+CONF_EMAIL = "email"
+CONF_PASSWORD = "password"  # noqa: S105 - config-flow field name, not a secret value
+CONF_CERT_DIR = "cert_dir"
+
+# The vendor mutual-TLS client certificate is NOT shipped in this repo (it is
+# Watkins material extractable from the app APK). Place the three PEM files here
+# on the Home Assistant host; see the README.
+DEFAULT_CERT_DIR = "/config/hotspring/certs"
+CERT_FILE = "client1_cert.pem"
+KEY_FILE = "client1_key.pem"
+CA_FILE = "ca_cert.pem"
+
+# Broker credentials rotate; re-fetch and reconnect on this cadence as a
+# safety net in addition to paho's own auto-reconnect.
+CREDS_REFRESH_INTERVAL = 6 * 60 * 60  # seconds
+
+MANUFACTURER = "Watkins Wellness"

custom_components/hotspring/coordinator.py

@@ -0,0 +1,100 @@
+"""Coordinator: owns the cloud session + MQTT connection for one spa and exposes
+the latest telemetry to entities."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .api import HotSpringCloud, HotSpringSpa, Spa
+from .const import (
+    CA_FILE,
+    CERT_FILE,
+    CONF_CERT_DIR,
+    CONF_EMAIL,
+    CONF_PASSWORD,
+    DEFAULT_CERT_DIR,
+    DOMAIN,
+    KEY_FILE,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class HotSpringCoordinator(DataUpdateCoordinator[dict]):
+    """Keeps a live MQTT connection and republishes state to entities.
+
+    Data flow is push-based: telemetry arrives on paho's thread and is bridged
+    onto the event loop via ``async_set_updated_data``. There is no polling
+    ``_async_update_data`` (updates are event-driven), but the coordinator still
+    provides ``data`` and listener plumbing for entities.
+    """
+
+    def __init__(self, hass: HomeAssistant, entry: ConfigEntry):
+        super().__init__(hass, _LOGGER, name=DOMAIN)
+        self.entry = entry
+        self.cloud: HotSpringCloud | None = None
+        self.spa: HotSpringSpa | None = None
+        self.spa_info: Spa | None = None
+
+    def _cert_paths(self) -> tuple[str, str, str]:
+        d = self.entry.data.get(CONF_CERT_DIR, DEFAULT_CERT_DIR)
+        return (
+            os.path.join(d, CERT_FILE),
+            os.path.join(d, KEY_FILE),
+            os.path.join(d, CA_FILE),
+        )
+
+    async def async_setup(self) -> None:
+        """Authenticate, resolve the spa, fetch broker creds, connect MQTT."""
+        cert, key, ca = self._cert_paths()
+        cloud = HotSpringCloud(
+            self.entry.data[CONF_EMAIL],
+            self.entry.data[CONF_PASSWORD],
+            cert,
+            key,
+            ca,
+        )
+
+        def _blocking_connect() -> HotSpringSpa:
+            cloud.login()
+            spas = cloud.get_spas()
+            if not spas:
+                raise RuntimeError("account has no spas")
+            self.spa_info = spas[0]
+            creds = cloud.fetch_mqtt_credentials()
+            spa = HotSpringSpa(
+                self.spa_info.root_topic, creds, on_status=self._on_status
+            )
+            spa.connect()
+            spa.wait_for_state()
+            return spa
+
+        self.cloud = cloud
+        self.spa = await self.hass.async_add_executor_job(_blocking_connect)
+        # Seed data so entities have an initial value.
+        self.async_set_updated_data(self.spa.merged)
+
+    @callback
+    def _on_status(self, suffix: str, payload: dict) -> None:
+        """Runs on paho's thread; hop to the event loop to notify listeners."""
+        if self.spa is None:
+            return
+        merged = self.spa.merged
+        self.hass.loop.call_soon_threadsafe(self.async_set_updated_data, merged)
+
+    async def async_publish_control(self, subsystem: str, control: dict) -> None:
+        if self.spa is None:
+            raise RuntimeError("spa not connected")
+        await self.hass.async_add_executor_job(
+            self.spa.publish_control, subsystem, control
+        )
+
+    async def async_shutdown(self) -> None:
+        if self.spa is not None:
+            await self.hass.async_add_executor_job(self.spa.disconnect)
+        await super().async_shutdown()

custom_components/hotspring/manifest.json

@@ -0,0 +1,12 @@
+{
+  "domain": "hotspring",
+  "name": "Hot Spring Connected Spa",
+  "codeowners": ["@fritzlab"],
+  "config_flow": true,
+  "documentation": "https://code.fritzlab.net/homeassistant/hotspring",
+  "issue_tracker": "https://code.fritzlab.net/homeassistant/hotspring/issues",
+  "iot_class": "cloud_push",
+  "integration_type": "hub",
+  "requirements": [],
+  "version": "0.1.0"
+}

custom_components/hotspring/sensor.py

@@ -0,0 +1,94 @@
+"""Sensors: salt output level, cartridge state, current water temperature."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.const import UnitOfTemperature
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import HotSpringConfigEntry
+from .const import DOMAIN, MANUFACTURER
+from .coordinator import HotSpringCoordinator
+
+
+@dataclass(frozen=True, kw_only=True)
+class HotSpringSensorDescription(SensorEntityDescription):
+    value_fn: Callable[[HotSpringCoordinator], float | str | None]
+
+
+SENSORS: tuple[HotSpringSensorDescription, ...] = (
+    HotSpringSensorDescription(
+        key="water_temperature",
+        translation_key="water_temperature",
+        device_class=SensorDeviceClass.TEMPERATURE,
+        native_unit_of_measurement=UnitOfTemperature.FAHRENHEIT,
+        state_class=SensorStateClass.MEASUREMENT,
+        value_fn=lambda c: c.spa.current_temperature() if c.spa else None,
+    ),
+    HotSpringSensorDescription(
+        key="salt_output_level",
+        translation_key="salt_output_level",
+        state_class=SensorStateClass.MEASUREMENT,
+        icon="mdi:shaker-outline",
+        value_fn=lambda c: c.spa.salt_output_level() if c.spa else None,
+    ),
+    HotSpringSensorDescription(
+        key="cartridge",
+        translation_key="cartridge",
+        icon="mdi:filter-outline",
+        value_fn=lambda c: (
+            None
+            if not c.spa or c.spa.cartridge_installed() is None
+            else ("installed" if c.spa.cartridge_installed() else "missing")
+        ),
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: HotSpringConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    coordinator = entry.runtime_data
+    async_add_entities(HotSpringSensor(coordinator, d) for d in SENSORS)
+
+
+class HotSpringSensor(CoordinatorEntity[HotSpringCoordinator], SensorEntity):
+    _attr_has_entity_name = True
+    entity_description: HotSpringSensorDescription
+
+    def __init__(
+        self,
+        coordinator: HotSpringCoordinator,
+        description: HotSpringSensorDescription,
+    ) -> None:
+        super().__init__(coordinator)
+        self.entity_description = description
+        root = coordinator.spa_info.root_topic if coordinator.spa_info else "spa"
+        self._attr_unique_id = f"{root}_{description.key}"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, root)},
+            manufacturer=MANUFACTURER,
+            name=coordinator.spa_info.name if coordinator.spa_info else "Hot Spring Spa",
+            model="Connected Spa",
+        )
+
+    @property
+    def native_value(self) -> float | str | None:
+        return self.entity_description.value_fn(self.coordinator)
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        self.async_write_ha_state()

custom_components/hotspring/strings.json

@@ -0,0 +1,31 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "title": "Hot Spring Connected Spa",
+        "description": "Sign in with your Hot Spring Spas app account. The mutual-TLS client certificate must already be present on the host (see the integration README).",
+        "data": {
+          "email": "E-mail",
+          "password": "Password",
+          "cert_dir": "Certificate directory"
+        }
+      }
+    },
+    "error": {
+      "invalid_auth": "Invalid e-mail or password.",
+      "cannot_connect": "Could not reach the Hot Spring cloud.",
+      "missing_certs": "The client certificate files were not found in the certificate directory.",
+      "unknown": "Unexpected error."
+    },
+    "abort": {
+      "already_configured": "This spa account is already configured."
+    }
+  },
+  "entity": {
+    "sensor": {
+      "water_temperature": { "name": "Water temperature" },
+      "salt_output_level": { "name": "Salt output level" },
+      "cartridge": { "name": "Salt cartridge" }
+    }
+  }
+}

custom_components/hotspring/translations/en.json

@@ -0,0 +1,31 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "title": "Hot Spring Connected Spa",
+        "description": "Sign in with your Hot Spring Spas app account. The mutual-TLS client certificate must already be present on the host (see the integration README).",
+        "data": {
+          "email": "E-mail",
+          "password": "Password",
+          "cert_dir": "Certificate directory"
+        }
+      }
+    },
+    "error": {
+      "invalid_auth": "Invalid e-mail or password.",
+      "cannot_connect": "Could not reach the Hot Spring cloud.",
+      "missing_certs": "The client certificate files were not found in the certificate directory.",
+      "unknown": "Unexpected error."
+    },
+    "abort": {
+      "already_configured": "This spa account is already configured."
+    }
+  },
+  "entity": {
+    "sensor": {
+      "water_temperature": { "name": "Water temperature" },
+      "salt_output_level": { "name": "Salt output level" },
+      "cartridge": { "name": "Salt cartridge" }
+    }
+  }
+}