DelonghiCoffeeLinkAPI.py

#!/usr/bin/env python3

import base64
from dataclasses import asdict, dataclass
import json
import datetime
import logging
import os
import re
import struct
import time
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import requests

# ─────────────────────────── Logging ───────────────────────────

def make_default_logger(name: str = "DeLonghiAPI") -> logging.Logger:
	"""Create a default silent logger."""
	logger = logging.getLogger(name)
	logger.addHandler(logging.NullHandler())
	return logger

# ─────────────────────────── Global logging bridge ───────────────────────────

_global_logger: logging.Logger = make_default_logger("DeLonghiAPI")


def attach_logger(logger: logging.Logger) -> None:
	"""
	Attach a global logger for free functions and static helpers.

	This allows top-level functions (like json_or_debug or print_jwt_info)
	to log via the same logger used by DeLonghiAPI / AylaClient.
	"""
	global _global_logger
	_global_logger = logger


def get_logger(name: Optional[str] = None) -> logging.Logger:
	"""
	Retrieve the currently active global logger, optionally namespaced.
	If no logger is attached, returns a silent default logger.
	"""
	if name:
		return _global_logger.getChild(name)
	return _global_logger

# ─────────────────────────── Helpers ───────────────────────────

def _parse_json_safe(val: Optional[str]) -> Optional[Dict[str, Any]]:
	if not val:
		return None
	try:
		return json.loads(val)
	except Exception:
		return None

def json_or_debug(resp: requests.Response, label: str) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
	"""
	Safely parse a JSON response, with debug fallback on error.

	This helper tries to decode the response body as JSON. If parsing fails
	(e.g. invalid content type or malformed body), it logs detailed
	diagnostic output (HTTP status, first 800 bytes of the body) and re-raises
	the exception.

	Args:
		resp (requests.Response): The HTTP response object returned by `requests`.
		label (str): A short identifier (e.g. endpoint name) used in error logs.

	Returns:
		Union[Dict[str, Any], List[Dict[str, Any]]]: Parsed JSON payload.

	Raises:
		ValueError: If the response body is not valid JSON.
		requests.RequestException: If any request-related errors occur.
	"""
	log = get_logger("helpers")
	try:
		return resp.json()
	except Exception as exc:
		text_preview = resp.text[:800].replace("\n", "\\n")
		log.error(f"[!] {label}: Failed to parse JSON (HTTP {resp.status_code})")
		log.debug(f"↳ Response snippet:\n{text_preview}")
		raise ValueError(f"Invalid JSON in {label}: {exc}") from exc


def print_jwt_info(jwt: str, verbose: bool = False) -> Optional[Dict[str, Any]]:
	"""
	Decode and print basic information from a JSON Web Token (JWT).

	This function extracts and base64-decodes the JWT payload, then logs the
	main fields (`sub`, `iss`, `iat`, `exp`). If an expiration (`exp`) field is
	present, the remaining lifetime is also shown. Optionally, the entire
	decoded payload can be logged at debug level for inspection.

	Args:
		jwt (str): The raw JWT string in standard `header.payload.signature` format.
		verbose (bool, optional): If True, logs the full decoded payload.

	Returns:
		Optional[Dict[str, Any]]: The decoded JWT payload dictionary, or None on error.

	Raises:
		ValueError: If the JWT is malformed or cannot be base64-decoded.
	"""
	log = get_logger("helpers")
	try:
		parts = jwt.split(".")
		if len(parts) < 2:
			raise ValueError("Invalid JWT structure: missing payload section")

		_, payload, *_ = parts
		# JWTs use base64url encoding without padding, so restore correct padding
		padded = payload + "=" * ((4 - len(payload) % 4) % 4)
		decoded_bytes = base64.urlsafe_b64decode(padded)
		data: Dict[str, Any] = json.loads(decoded_bytes)

		exp = data.get("exp")
		core_info = {k: data.get(k) for k in ("sub", "iss", "iat", "exp") if k in data}
		log.debug(f"[i] JWT core info: {core_info}")

		if exp is not None:
			remaining = exp - int(time.time())
			log.debug(f"    → expires in: {remaining} seconds")

		if verbose:
			log.debug(f"[i] Full JWT payload:\n{json.dumps(data, indent=2)}")

		return data

	except Exception as exc:
		log.error(f"[!] Failed to decode JWT: {exc}")
		return None


# ─────────────────────────── Cache Layer ───────────────────────────

class _JsonCache:
	"""
	Lightweight on-disk JSON key-value cache with optional time-based expiry.

	This class provides a simple way to persist small key-value dictionaries
	between program runs. Each value is stored with a timestamp to allow
	time-to-live (TTL) validation on retrieval.

	Attributes:
		path (str): Path to the JSON cache file on disk.
		data (Dict[str, Any]): The in-memory dictionary representing cached data.

	Example:
		>>> cache = _JsonCache("cache.json")
		>>> cache.set("session", {"token": "abc"})
		>>> cache.get("session", ttl=3600)
		{'token': 'abc'}
	"""

	def __init__(self, path: str):
		"""
		Initialize the JSON cache and load data from disk if available.

		Args:
			path (str): Filesystem path to the cache file.
		"""
		self.path = path
		self.data: Dict[str, Any] = {}
		try:
			if os.path.exists(path):
				# Use safe blocking call
				with open(path, "r", encoding="utf-8") as f:
					self.data = json.load(f)
		except Exception as exc:
			print(f"[Cache] Failed to load cache {path}: {exc}")
		self.log = make_default_logger("DeLonghiAPI._JsonCache")
		if os.path.exists(path):
			try:
				with open(path, "r", encoding="utf-8") as f:
					self.data = json.load(f)
			except Exception as exc:
				self.log.warning(f"[!] Failed to read cache from {path}: {exc}")
				self.data = {}

	def get(self, key: str, *, ttl: Optional[int] = None) -> Optional[Any]:
		"""
		Retrieve a cached value by key, optionally enforcing a time-to-live (TTL).

		Args:
			key (str): The cache key to look up.
			ttl (Optional[int]): Optional time-to-live in seconds. If provided,
				entries older than this value are considered expired and ignored.

		Returns:
			Optional[Any]: The cached value, or None if not found or expired.
		"""
		item = self.data.get(key)
		if not item:
			return None

		if ttl is not None:
			ts = float(item.get("ts", 0))
			if time.time() - ts > ttl:
				self.log.debug(f"[→] Cache expired for key '{key}' (TTL={ttl}s)")
				return None

		return item.get("value")

	def set(self, key: str, value: Any) -> None:
		"""
		Store a value in the cache with a timestamp and persist to disk.

		Args:
			key (str): The cache key.
			value (Any): The value to store (must be JSON serializable).
		"""
		self.data[key] = {"value": value, "ts": time.time()}
		try:
			with open(self.path, "w", encoding="utf-8") as f:
				json.dump(self.data, f, indent=2)
			self.log.debug(f"[✓] Cache updated: {key}")
		except Exception as exc:
			self.log.error(f"[!] Failed to write cache {self.path}: {exc}")


# ─────────────────────────── AylaClient ───────────────────────────

class AylaClient:
	"""
	HTTP client for authenticating and communicating with the Ayla Cloud API.

	Handles login via token-based authentication, manages token caching,
	and provides a thin wrapper around the Ayla endpoints with built-in
	re-authentication support.

	Attributes:
		app_id (str): The Ayla app ID for authentication.
		app_secret (str): The Ayla app secret for authentication.
		cache (_JsonCache): Persistent cache for storing authentication tokens.
		main_session (requests.Session): Configured session with headers.
		default_timeout (int): Default timeout (in seconds) for all requests.
		auth_token (Optional[str]): The current active Ayla authentication token.
	"""

	BASE_USER = "https://user-field-eu.aylanetworks.com"
	BASE_DEVICES = "https://ads-eu.aylanetworks.com/apiv1"

	def __init__(
		self,
		app_id: str,
		app_secret: str,
		*,
		ua_token: str = "DeLonghiComfort/3 CFNetwork/1568.300.101 Darwin/24.2.0",
		cache_path: str = "auth_cache.json",
		session: Optional[requests.Session] = None,
		default_timeout: int = 20,
	):
		"""
		Initialize the Ayla API client.

		Args:
			app_id (str): The Ayla application ID.
			app_secret (str): The Ayla application secret.
			ua_token (str, optional): User-Agent string for API requests.
			cache_path (str, optional): Path to the authentication cache file.
			session (Optional[requests.Session], optional): Custom session object.
			default_timeout (int, optional): Default timeout (seconds) for requests.
		"""
		self.app_id = app_id
		self.app_secret = app_secret
		self.cache = _JsonCache(cache_path)
		self.default_timeout = default_timeout

		self.main_session = session or requests.Session()
		self.main_session.headers.update({
			"User-Agent": ua_token,
			"Accept": "application/json",
		})

		self.auth_token: Optional[str] = None
		self._email: Optional[str] = None
		self._password: Optional[str] = None
		self.log = make_default_logger("DeLonghiAPI.AylaClient")

	# ─────────────────────── Helpers ───────────────────────

	def _auth_headers(self) -> Dict[str, str]:
		"""
		Build headers for authenticated requests.

		Returns:
			Dict[str, str]: Headers containing the current auth token.

		Raises:
			RuntimeError: If no valid authentication token is available.
		"""
		if not self.auth_token:
			raise RuntimeError("Not authenticated — call login() first")
		return {"Authorization": f"auth_token {self.auth_token}"}

	def _has_valid_cached_token(self) -> bool:
		"""
		Check for a valid, non-expired cached token.

		Returns:
			bool: True if a valid cached token exists and is loaded.
		"""
		cached = self.cache.get("ayla_auth", ttl=None)
		if not cached:
			return False

		token = cached.get("auth_token")
		exp = cached.get("expires_at", 0)
		if token and exp > time.time():
			self.auth_token = token
			self.log.debug("[✓] Using cached Ayla token.")
			return True

		self.log.debug("[→] Cached token missing or expired.")
		return False

	def _store_token(self, token: str, *, lifetime_seconds: int = 23 * 3600) -> None:
		"""
		Store a freshly obtained authentication token in the cache.

		Args:
			token (str): The authentication token string.
			lifetime_seconds (int, optional): Token lifetime in seconds.
		"""
		self.auth_token = token
		expires_at = int(time.time()) + lifetime_seconds
		self.cache.set("ayla_auth", {
			"auth_token": token,
			"expires_at": expires_at,
		})
		self.log.debug(f"[✓] Stored new Ayla token (expires in {lifetime_seconds}s).")

	# ─────────────────────── Core Methods ───────────────────────

	def ayla_token_sign_in(self, id_token: str) -> str:
		"""
		Exchange a Gigya or external ID token for an Ayla auth token.

		Args:
			id_token (str): The external identity token (e.g. from Gigya).

		Returns:
			str: The Ayla access/auth token.

		Raises:
			RuntimeError: If the server response does not include a token.
			requests.RequestException: On network or HTTP error.
		"""
		self.log.debug("Performing Ayla token_sign_in...")
		url = f"{self.BASE_USER}/api/v1/token_sign_in"
		data = {
			"app_id": self.app_id,
			"app_secret": self.app_secret,
			"token": id_token,
		}
		headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}

		self.log.debug(f"[→] POST {url} with data: {data}")
		r = self.main_session.post(url, data=data, headers=headers, timeout=self.default_timeout)
		r.raise_for_status()

		js = json_or_debug(r, "ayla/token_sign_in")

		if not isinstance(js, dict):
			raise TypeError(f"Expected dict from ayla/token_sign_in, got {type(js).__name__}: {js}")

		token_val = js.get("access_token") or js.get("auth_token") or js.get("token")
		if not isinstance(token_val, str) or not token_val.strip():
			raise RuntimeError(f"Invalid or missing auth token in response: {js}")

		token: str = token_val.strip()
		self.log.debug("[✓] Ayla token_sign_in successful.")
		return token

	def _request(
		self,
		method: str,
		url: str,
		*,
		headers: Optional[Dict[str, str]] = None,
		params: Optional[Dict[str, Any]] = None,
		data: Optional[Any] = None,
		json: Optional[Any] = None,
		allow_refresh: bool = True,
		timeout: Optional[int] = None,
	) -> requests.Response:
		"""
		Perform an HTTP request with automatic token refresh on 401 errors.

		Args:
			method (str): HTTP method (GET, POST, PUT, DELETE, etc.).
			url (str): Full endpoint URL.
			headers (Optional[Dict[str, str]]): Additional request headers.
			params (Optional[Dict[str, Any]]): Query string parameters.
			data (Optional[Any]): Form data for POST/PUT requests.
			json (Optional[Any]): JSON body for POST/PUT requests.
			allow_refresh (bool, optional): Whether to retry once on 401 Unauthorized.
			timeout (Optional[int], optional): Request timeout in seconds.

		Returns:
			requests.Response: The HTTP response object.

		Raises:
			requests.HTTPError: If the response code indicates a fatal error.
			RuntimeError: If re-authentication fails.
		"""
		if timeout is None:
			timeout = self.default_timeout

		self.log.debug(f"[→] {method} {url} (timeout={timeout})")
		r = self.main_session.request(
			method,
			url,
			headers=headers,
			params=params,
			data=data,
			json=json,
			timeout=timeout,
		)

		if r.status_code == 401 and allow_refresh:
			self.log.warning("[!] Auth expired — refreshing and retrying once...")
			self._refresh_auth()
			r = self.main_session.request(
				method,
				url,
				headers=headers,
				params=params,
				data=data,
				json=json,
				timeout=timeout,
			)

		r.raise_for_status()
		self.log.debug(f"[✓] {method} {url} → {r.status_code}")
		return r

	def ayla_list_devices(self) -> List[Dict[str, Any]]:
		"""
		Retrieve the list of all devices linked to the authenticated Ayla account.

		Performs a GET request to the `/devices.json` endpoint using the current
		authentication token and writes the raw response to a local dump file
		(`DUMP_devices.json`) for debugging.

		Returns:
			List[Dict[str, Any]]: A list of device objects as returned by Ayla Cloud.

		Raises:
			RuntimeError: If authentication is missing or invalid.
			requests.HTTPError: On HTTP request failure.
			TypeError: If the response JSON is not a list.
		"""
		self.log.debug("Fetching device list...")
		url = f"{self.BASE_DEVICES}/devices.json"
		headers = self._auth_headers()

		r = self._request("GET", url, headers=headers)

		# Dump for debugging (optional but useful)
		try:
			with open("DUMP_devices.json", "w", encoding="utf-8") as f:
				f.write(r.text)
			self.log.debug("[✓] Device list saved to DUMP_devices.json")
		except Exception as exc:
			self.log.warning(f"[!] Could not write device dump: {exc}")

		js = json_or_debug(r, "ayla/devices")
		if not isinstance(js, list):
			raise TypeError(
				f"Expected list response for ayla/devices, got {type(js).__name__}: {js}"
			)

		self.log.debug(f"[✓] Retrieved {len(js)} device entries.")
		return js

	def ayla_list_device_properties(self, device_key: str) -> List[Dict[str, Any]]:
		"""
		Fetch the full list of properties for a specific Ayla device.

		Args:
			device_key (str): The unique device identifier (DSN or device key).

		Returns:
			List[Dict[str, Any]]: A list of property objects for the specified device.

		Raises:
			RuntimeError: If authentication is missing.
			requests.HTTPError: On network or API failure.
			TypeError: If the response JSON is not a list.
		"""
		self.log.debug(f"Fetching properties for device {device_key}...")
		url = f"{self.BASE_DEVICES}/devices/{device_key}/properties.json"
		headers = self._auth_headers()

		r = self._request("GET", url, headers=headers, timeout=25)
		js = json_or_debug(r, "device/properties")

		if not isinstance(js, list):
			raise TypeError(
				f"Expected list response for device/properties, got {type(js).__name__}: {js}"
			)

		self.log.debug(f"[✓] Retrieved {len(js)} properties for device {device_key}.")
		return js
	# ─────────────────────── Auth Utilities ───────────────────────

	def login_with_cached_token_or(self, fresh_login_callable: Callable[[], str]) -> None:
		"""
		Attempt to use a cached Ayla token; if unavailable or expired, perform a fresh login.

		This helper first checks the token cache for a valid authentication token.
		If one exists and has not expired, it is loaded and reused. Otherwise, it
		calls the provided function (`fresh_login_callable`) to obtain a new token,
		stores it, and logs the event.

		Args:
			fresh_login_callable (Callable[[], str]): A callback returning a new auth token.

		Raises:
			RuntimeError: If the provided callable does not return a valid token.
		"""
		self.log.debug("Checking for cached Ayla token...")
		if self._has_valid_cached_token():
			cached = self.cache.get("ayla_auth")
			if cached:
				remaining = int(cached.get("expires_at", 0) - time.time())
				self.log.debug(f"[+] Using cached Ayla token (valid {max(0, remaining)}s)")
			return

		self.log.debug("[*] Performing fresh login…")
		token = fresh_login_callable()
		if not token:
			raise RuntimeError("Login callback returned an empty or invalid token")

		token = str(token).strip()
		if not token:
			raise RuntimeError("Login callback returned a blank token")

		self._store_token(token)
		self.log.debug("[✓] Cached new Ayla token.")

	def login(self, email: str, password: str, force_refresh: bool = False) -> None:
		"""
		Placeholder login method for subclasses.
		Overridden by DeLonghiAPI to perform real authentication.
		"""
		self.log.warning("[!] AylaClient.login() called directly — no implementation available.")
		raise RuntimeError("Base AylaClient does not implement login() — use a subclass.")

	def _refresh_auth(self) -> None:
		"""
		Refresh the current authentication token using stored credentials.

		This is automatically triggered when a 401 Unauthorized response is
		received. The method reuses the previously stored email and password
		to reauthenticate the client.

		Raises:
			RuntimeError: If stored credentials are missing.
		"""
		if not (self._email and self._password):
			raise RuntimeError("Cannot refresh auth: missing credentials.")
		self.log.debug("[→] Refreshing Ayla authentication token...")
		self.login(self._email, self._password, force_refresh=True)
		self.log.debug("[✓] Authentication successfully refreshed.")

	# ─────────────────────── Device Wrappers ───────────────────────

	def get_devices(self) -> List["DeLonghiDevice"]:
		"""
		Retrieve all connected DeLonghi devices from Ayla Cloud.

		Wraps the raw device list in `DeLonghiDevice` objects for higher-level access.

		Returns:
			List[DeLonghiDevice]: List of initialized device wrappers.

		Raises:
			RuntimeError: If the response structure is invalid or missing device keys.
		"""
		self.log.debug("Loading DeLonghi device list...")
		raw_devices = self.ayla_list_devices()
		devices: List["DeLonghiDevice"] = []

		for entry in raw_devices:
			device = entry.get("device")
			if device:
				devices.append(DeLonghiDevice(self, device))
			else:
				self.log.warning(f"[!] Skipped invalid device entry: {entry}")

		self.log.debug(f"[✓] Loaded {len(devices)} device(s).")
		return devices

	def get_first_device(self) -> "DeLonghiDevice":
		"""
		Convenience method to retrieve the first available device.

		Returns:
			DeLonghiDevice: The first device in the list.

		Raises:
			RuntimeError: If no devices are available in the account.
		"""
		devices = self.get_devices()
		if not devices:
			raise RuntimeError("No devices found for the current account.")
		self.log.debug(f"[✓] Selected first device: {devices[0]}")
		return devices[0]


# ─────────────────────────── DeLonghiAPI ───────────────────────────

class DeLonghiAPI(AylaClient):
	"""
	High-level API client for DeLonghi Coffee Link Cloud.

	This class extends :class:`AylaClient` to handle Gigya authentication,
	Ayla token exchange, and integration with DeLonghi's Reply.it endpoints.

	It encapsulates the entire authentication chain:
	Gigya → Ayla Cloud → DeLonghi devices.

	Attributes:
		gigya_api_key (str): Gigya API key used for authentication.
		datacenter (str): Gigya data center region (e.g., "eu1").
		SOCIALIZE (str): Base URL for Gigya's Socialize API.
		ACCOUNTS (str): Base URL for Gigya's Accounts API.
		REPLY_BASE (str): Base URL for DeLonghi's Reply.it API.
		_transcode_table (Optional[Dict[str, Any]]): Cached Reply.it translation table.
	"""

	def __init__(
		self,
		app_id: str = "DLonghiCoffeeIdKit-sQ-id",
		app_secret: str = "DLonghiCoffeeIdKit-HT6b0VNd4y6CSha9ivM5k8navLw",
		gigya_api_key: str = "4_mXSplGaqrFT0H88TAjqJuA",
		datacenter: str = "eu1",
		ua_token: str = "DeLonghiComfort/3 CFNetwork/1568.300.101 Darwin/24.2.0",
		cache_path: str = "auth_cache.json",
	):
		"""
		Initialize a new DeLonghi Coffee Link API client.

		Args:
			app_id (str, optional): Ayla app ID. Defaults to the official DeLonghi one.
			app_secret (str, optional): Ayla app secret. Defaults to DeLonghi value.
			gigya_api_key (str, optional): Gigya API key for authentication.
			datacenter (str, optional): Gigya datacenter (e.g., "eu1").
			ua_token (str, optional): User-Agent string for HTTP requests.
			cache_path (str, optional): Path to persistent authentication cache.
		"""
		super().__init__(
			app_id=app_id,
			app_secret=app_secret,
			ua_token=ua_token,
			cache_path=cache_path,
		)

		self.gigya_api_key = gigya_api_key
		self.datacenter = datacenter
		self.SOCIALIZE = f"https://socialize.{datacenter}.gigya.com"
		self.ACCOUNTS = f"https://accounts.{datacenter}.gigya.com"
		self.REPLY_BASE = "https://delonghibe.reply.it/api"
		self._transcode_table: Optional[Dict[str, Any]] = None

		self.log = make_default_logger("DeLonghiAPI")

	def set_logger(self, logger: logging.Logger) -> None:
		"""
		Attach a custom logger to this API (and propagate to child components).
		"""
		self.log = logger
		# Also attach to AylaClient base
		if hasattr(super(), "log"):
			super().log = logger

	# ─────────────────────── Gigya & Ayla Authentication ───────────────────────

	def _gigya_get_session_ids(self) -> Tuple[str, str, str]:
		"""
		Retrieve session identifiers (GMID, UCID, GMID ticket) from Gigya.

		Returns:
			Tuple[str, str, str]: A tuple containing:
				- GMID (Global Machine ID)
				- UCID (User Connection ID)
				- GMID ticket (temporary session ticket)

		Raises:
			requests.RequestException: On network or HTTP error.
			RuntimeError: If Gigya response structure is invalid.
			TypeError: If the API returns an unexpected response type.
		"""
		url = f"{self.SOCIALIZE}/socialize.getIDs"
		params = {
			"APIKey": self.gigya_api_key,
			"includeTicket": "true",
			"pageURL": "https://aylaopenid.delonghigroup.com/",
			"sdk": "Android_6.0.0",
			"format": "json",
		}

		self.log.debug(f"[→] GET {url} with params: {params}")
		r = self.main_session.get(url, params=params, timeout=15)
		r.raise_for_status()

		js = json_or_debug(r, "socialize.getIDs")

		# Type narrowing: ensure JSON is a dict
		if not isinstance(js, dict):
			raise TypeError(
				f"Expected dict response for socialize.getIDs, got {type(js).__name__}: {js}"
			)

		raw_gmid = js.get("gmid")
		raw_ucid = js.get("ucid")
		raw_ticket = js.get("gmidTicket")

		# Validate each field individually so Pylance knows they’re strings now
		if not isinstance(raw_gmid, str) or not raw_gmid.strip():
			raise RuntimeError(f"Missing or invalid 'gmid' in response: {js}")
		if not isinstance(raw_ucid, str) or not raw_ucid.strip():
			raise RuntimeError(f"Missing or invalid 'ucid' in response: {js}")
		if not isinstance(raw_ticket, str) or not raw_ticket.strip():
			raise RuntimeError(f"Missing or invalid 'gmidTicket' in response: {js}")

		gmid: str = raw_gmid.strip()
		ucid: str = raw_ucid.strip()
		ticket: str = raw_ticket.strip()

		self.log.debug(f"[✓] Gigya session IDs acquired (gmid={gmid[:8]}..., ucid={ucid[:8]}...)")
		return gmid, ucid, ticket

	def _gigya_login(self, email: str, password: str) -> str:
		"""
		Log into the user's Gigya account and obtain an ID token.

		Args:
			email (str): The user's DeLonghi/Gigya account email.
			password (str): The user's password.

		Returns:
			str: The returned Gigya ID token.

		Raises:
			RuntimeError: If Gigya does not return an `id_token`.
			requests.RequestException: On network or HTTP failure.
			TypeError: If the API response is not a JSON object.
		"""
		url = f"{self.ACCOUNTS}/accounts.login"
		data = {
			"APIKey": self.gigya_api_key,
			"loginID": email,
			"password": password,
			"targetEnv": "mobile",
			"include": "profile,data,emails,subscriptions,preferences,sessionInfo,id_token",
			"includeUserInfo": "true",
			"format": "json",
			"sdk": "Android_6.0.0",
			"sdkBuild": "33001",
			"pageURL": "https://aylaopenid.delonghigroup.com/",
		}

		headers = {
			"Origin": "https://aylaopenid.delonghigroup.com",
			"Content-Type": "application/x-www-form-urlencoded",
			"X-Gigya-Client-Device-Type": "Android",
			"X-Gigya-Client-Device-ID": "android-" + str(int(time.time())),
		}

		self.log.debug(f"[→] POST {url} with headers={headers} and data={data}")
		r = self.main_session.post(url, data=data, headers=headers, timeout=20)
		r.raise_for_status()

		js = json_or_debug(r, "accounts.login")

		# Ensure we have a dict
		if not isinstance(js, dict):
			raise TypeError(
				f"Expected dict response for accounts.login, got {type(js).__name__}: {js}"
			)

		raw_token = js.get("id_token") or js.get("idToken")
		if not isinstance(raw_token, str) or not raw_token.strip():
			raise RuntimeError(f"accounts.login did not return a valid idToken: {js}")

		token: str = raw_token.strip()
		self.log.debug("[✓] Gigya login successful.")
		return token

	def login(self, email: str, password: str, force_refresh: bool = False) -> None:
		"""
		Perform the full DeLonghi login flow via Gigya → Ayla.

		Args:
			email (str): User's DeLonghi Coffee Link login email.
			password (str): Account password.
			force_refresh (bool, optional): If True, forces a new login even if cached.

		Raises:
			RuntimeError: On authentication failure or network error.
		"""
		self._email, self._password = email, password

		if not force_refresh and self._has_valid_cached_token():
			cached = self.cache.get("ayla_auth")
			if cached:
				remaining = int(cached.get("expires_at", 0) - time.time())
				self.log.debug(f"[+] Using cached Ayla token (valid {max(0, remaining)}s)")
				return

		self.log.debug("[*] Performing fresh login…")

		try:
			_, _, _ = self._gigya_get_session_ids()
			id_token = self._gigya_login(email, password)
			print_jwt_info(id_token)

			ayla_token = self.ayla_token_sign_in(id_token)
			self._store_token(ayla_token, lifetime_seconds=23 * 3600)

			self.log.debug("[✓] Cached new Ayla token.")
		except Exception as exc:
			self.log.error(f"[!] Login sequence failed: {exc}")
			raise

	def _refresh_auth(self) -> None:
		"""
		Refresh authentication token using stored credentials.

		This overrides the parent implementation to ensure Gigya → Ayla refresh flow.

		Raises:
			RuntimeError: If stored credentials are missing.
		"""
		if not (self._email and self._password):
			raise RuntimeError("Cannot refresh auth: missing credentials.")
		self.log.debug("[→] Refreshing DeLonghi authentication token...")
		self.login(self._email, self._password, force_refresh=True)
		self.log.debug("[✓] Authentication successfully refreshed.")

	# ─────────────────────── Reply.it Helpers ───────────────────────

	def _reply_body(self, endpoint: str) -> Dict[str, Any]:
		"""
		Construct the correct JSON body for a given Reply.it API endpoint.

		The payload mirrors the structure used by the Coffee Link app (v4.9.1,
		tableVersion 1.510) for compatibility with DeLonghi backend APIs.

		Args:
			endpoint (str): Name of the Reply.it endpoint, e.g. "getTranscodeTable".

		Returns:
			Dict[str, Any]: The properly formatted request body.
		"""
		body: Dict[str, Any] = {
			"locale": "en_US",
			"appVersion": "4.9.1",
			"tableVersion": "1.510",
			"helpVersion": 0,
		}

		if endpoint == "getAddictionalCommonData":
			body["countryCode"] = "DE"
		elif endpoint == "getTranscodeTable":
			body["currentVersion"] = "1.510"

		self.log.debug(f"[→] Constructed Reply.it request body for '{endpoint}': {body}")
		return body

	def _reply_post(self, endpoint: str) -> Dict[str, Any]:
		"""
		Send a POST request to a Reply.it API endpoint with the correct headers and body.

		Args:
			endpoint (str): Name of the endpoint (without `.sr` suffix),
				e.g. ``"getTranscodeTable"``.

		Returns:
			Dict[str, Any]: Parsed JSON response from the Reply.it API.

		Raises:
			requests.HTTPError: If the response status code is not successful.
			ValueError: If the response body is not valid JSON.
		"""
		url = f"{self.REPLY_BASE}/{endpoint}.sr"
		headers = {
			"Content-Type": "application/json",
			"Accept": "application/json",
			"User-Agent": "DeLonghiCoffeeLink/4.9.1 (Android)",
		}

		body = self._reply_body(endpoint)
		self.log.info(f"[→] POST {endpoint}.sr with {body}")

		r = self.main_session.post(url, headers=headers, json=body, timeout=25)
		print(headers)
		print(body)
		r.raise_for_status()

		try:
			js: Dict[str, Any] = r.json()
			self.log.debug(f"[✓] {endpoint}.sr → {len(json.dumps(js))} bytes JSON")
			return js
		except Exception as exc:
			self.log.error(f"[!] Failed to parse Reply.it response for {endpoint}: {exc}")
			raise ValueError(f"Invalid JSON in Reply.it response for {endpoint}") from exc

	def get_transcode_table(self, *, force_refresh: bool = False) -> Dict[str, Any]:
		"""
		Retrieve and cache the DeLonghi transcode table from Reply.it.

		Loads the ``getTranscodeTable.sr`` JSON from the Reply.it API and stores
		it in the persistent cache. The cached copy is valid for 7 days by default.

		Args:
			force_refresh (bool, optional): If True, forces a fresh fetch
				even if a valid cached table exists.

		Returns:
			Dict[str, Any]: The transcode table dictionary.

		Raises:
			Exception: If no valid cache or online data is available.
		"""
		self.log.debug("Fetching transcode table from Reply.it...")
		TTL = 7 * 24 * 3600  # 7 days

		if not force_refresh:
			cached = self.cache.get("reply_transcode_table", ttl=TTL)
			if cached:
				self._transcode_table = cached
				self.log.debug("[✓] Using cached transcode table.")
				return cached

		try:			
			self.cache.set("reply_transcode_table", js)
			self._transcode_table = js
			self.log.debug("[✓] Transcode table fetched and cached.")
			return js
		except Exception as exc:
			# fallback logic
			if self._transcode_table:
				self.log.warning(f"[!] Using in-memory transcode table after fetch error: {exc}")
				return self._transcode_table

			cached_no_ttl = self.cache.get("reply_transcode_table", ttl=None)
			if cached_no_ttl:
				self.log.warning(f"[!] Using expired cached transcode table after fetch error: {exc}")
				self._transcode_table = cached_no_ttl
				return cached_no_ttl

			raise RuntimeError("Failed to fetch or restore transcode table") from exc

# ─────────────────────────── Device Model ───────────────────────────

_ALNUM_RE = re.compile(rb"[0-9A-Z]{10,}")

@dataclass
class MachineSettings:
	"""Represents core machine settings (decoded from base64)."""

	brewing_temperature: Optional[Dict[str, Any]] = None
	auto_shutoff_duration: Optional[Dict[str, Any]] = None
	water_hardness_level: Optional[Dict[str, Any]] = None
	profile: Optional[int] = None

	# ─────────────────────────── Helpers ───────────────────────────

	@staticmethod
	def _b64decode(val: Optional[str]) -> Optional[bytes]:
		if not val:
			return None
		try:
			return base64.b64decode(val)
		except Exception:
			return None

	@staticmethod
	def _b64text(val: Optional[str]) -> Optional[str]:
		data = MachineSettings._b64decode(val)
		if not data:
			return None
		return data.decode(errors="ignore").strip("\x00")

	@staticmethod
	def _b64byte_level(val: Optional[str]) -> Optional[int]:
		data = MachineSettings._b64decode(val)
		if data and len(data) >= 10:
			return data[9]
		return None

	@staticmethod
	def _safe_json_from_b64(val: Optional[str]) -> Optional[Dict[str, Any]]:
		text = MachineSettings._b64text(val)
		if not text:
			return None
		try:
			return json.loads(text)
		except Exception:
			return {"raw": text}

	@staticmethod
	def _adjust_level(level: Optional[int]) -> Optional[int]:
		"""Convert 0-based byte levels to 1-based human levels."""
		if level is None:
			return None
		return level + 1

	@staticmethod
	def _decode_active_profile_index(val: Optional[str]) -> Optional[int]:
		"""Decode active user profile index from d286_mach_sett_profile (base64)."""
		data = MachineSettings._b64decode(val)
		if not data or len(data) < 7:
			return None
		idx = data[6]
		if 0 <= idx < 4:
			return idx + 1  # Convert 0-based index to 1–4
		return None

	# ─────────────────────────── Factory ───────────────────────────

	@classmethod
	def from_properties(cls, props: Dict[str, Dict[str, Any]]) -> "MachineSettings":
		"""
		Parse sanitized Ayla property dict into structured MachineSettings.
		Example structure:
			{"d281_mach_sett_temperature": {"value": "base64string"}, ...}
		"""
		def val(key: str) -> Optional[str]:
			entry = props.get(key)
			if isinstance(entry, dict):
				return entry.get("value")
			return None

		# Decode 3 main numeric settings
		temp_level = cls._adjust_level(cls._b64byte_level(val("d281_mach_sett_temperature")))
		auto_level = cls._adjust_level(cls._b64byte_level(val("d282_mach_sett_auto_off")))
		hard_level = cls._adjust_level(cls._b64byte_level(val("d283_mach_sett_water_hard")))

		# Label maps
		temp_label = {1: "low", 2: "medium", 3: "high"}.get(temp_level or 0)
		auto_label = {1: "15 min", 2: "30 min", 3: "1 h", 4: "3 h"}.get(auto_level or 0)
		hard_label = {
			1: "very soft", 2: "soft", 3: "medium", 4: "hard", 5: "very hard"
		}.get(hard_level or 0)

		# Decode active profile index
		profile = cls._decode_active_profile_index(val("d286_mach_sett_profile"))

		return cls(
			brewing_temperature={"level": temp_level, "label": temp_label},
			auto_shutoff_duration={"level": auto_level, "label": auto_label},
			water_hardness_level={"level": hard_level, "label": hard_label},
			profile=profile,
		)

	def to_dict(self) -> Dict[str, Any]:
		"""Return settings as plain dict (JSON-friendly)."""
		return asdict(self)

@dataclass
class TotBevOther:
	"""Parsed structure from d702_tot_bev_other."""
	TotalBevBW: Optional[int] = None
	TotalBevOther: Optional[int] = None
	TotalBevW: Optional[int] = None

	@classmethod
	def from_properties(cls, props: Dict[str, Dict[str, Any]]) -> "TotBevOther":
		entry = props.get("d702_tot_bev_other")
		if not isinstance(entry, dict):
			return cls()
		val = entry.get("value")
		js = _parse_json_safe(val)
		if not js:
			return cls()
		def _int(k): return int(js.get(k, 0) or 0)
		return cls(
			TotalBevBW=_int("tot_bev_bw"),
			TotalBevOther=_int("tot_bev_other"),
			TotalBevW=_int("tot_bev_w"),
		)

@dataclass
class TotBevCounters:
	"""Parsed structure from d733_tot_bev_counters."""
	MugHotSmall: Optional[int] = None
	MugHotMedium: Optional[int] = None
	MugHotLarge: Optional[int] = None
	MugColdSmall: Optional[int] = None
	MugColdMedium: Optional[int] = None
	MugColdLarge: Optional[int] = None
	BevBIced: Optional[int] = None
	BevBWIced: Optional[int] = None
	BevOtherIced: Optional[int] = None
	BevWIced: Optional[int] = None

	@classmethod
	def from_properties(cls, props: Dict[str, Dict[str, Any]]) -> "TotBevCounters":
		entry = props.get("d733_tot_bev_counters")
		if not isinstance(entry, dict):
			return cls()
		val = entry.get("value")
		js = _parse_json_safe(val)
		if not js:
			return cls()
		def _int(k): return int(js.get(k, 0) or 0)
		return cls(
			MugHotSmall=_int("tot_mug_hot_small"),
			MugHotMedium=_int("tot_mug_hot_medium"),
			MugHotLarge=_int("tot_mug_hot_large"),
			MugColdSmall=_int("tot_mug_cold_small"),
			MugColdMedium=_int("tot_mug_cold_medium"),
			MugColdLarge=_int("tot_mug_cold_large"),
			BevBIced=_int("tot_bev_b_iced"),
			BevBWIced=_int("tot_bev_bw_iced"),
			BevOtherIced=_int("tot_bev_other_iced"),
			BevWIced=_int("tot_bev_w_iced"),
		)

@dataclass
class TotBevUsage:
	"""Parsed structure from d734_tot_bev_usage."""
	PregrCoffCnt: Optional[int] = None
	TasteBBW: Optional[int] = None
	TotCustomBBW: Optional[int] = None
	TotBevNoOriginal: Optional[int] = None
	TotBevProfileStd: Optional[int] = None
	TotBevProfileOther: Optional[int] = None
	ProfileSet: Optional[int] = None
	TotBevBBWNoMySize: Optional[int] = None
	BevAbortCnt: Optional[int] = None
	Bev2xCnt: Optional[int] = None

	@classmethod
	def from_properties(cls, props: Dict[str, Dict[str, Any]]) -> "TotBevUsage":
		entry = props.get("d734_tot_bev_usage")
		if not isinstance(entry, dict):
			return cls()
		val = entry.get("value")
		js = _parse_json_safe(val)
		if not js:
			return cls()
		def _int(k): return int(js.get(k, 0) or 0)
		return cls(
			PregrCoffCnt=_int("pregr_coff_cnt"),
			TasteBBW=_int("taste_b_bw"),
			TotCustomBBW=_int("tot_custom_b_bw"),
			TotBevNoOriginal=_int("tot_bev_no_original"),
			TotBevProfileStd=_int("tot_bev_profile_std_profile1"),
			TotBevProfileOther=_int("tot_bev_profile_other"),
			ProfileSet=_int("profile_set"),
			TotBevBBWNoMySize=_int("tot_bev_b_bw_no_my_size"),
			BevAbortCnt=_int("bev_abort_cnt"),
			Bev2xCnt=_int("bev_2x_cnt"),
		)

@dataclass
class BeverageCounters:
	"""Represents all beverage counters (cups dispensed per beverage type)."""

	CoffeeBeans: int = 0
	OtherBeverages: int = 0
	EspressoTotal: int = 0
	Espresso: int = 0
	Coffee: int = 0
	Long: int = 0
	DoppioPlus: int = 0
	Americano: int = 0
	Cappuccino: int = 0
	LatteMacchiato: int = 0
	CaffeLatte: int = 0
	FlatWhite: int = 0
	EspressoMacchiato: int = 0
	HotMilk: int = 0
	CappuccinoDoppioPlus: int = 0
	CappuccinoReverse: int = 0
	HotWater: int = 0
	Tea: int = 0
	CoffeePot: int = 0
	BrewOverIce: int = 0
	MugHot: int = 0
	MugCold: int = 0
	TotalBeverages: int = 0
	TotalUsage: int = 0
	IcedBeverage: int = 0
	MugBeverage: int = 0
	MugIcedBeverage: int = 0
	ColdBrew: int = 0
	TasteBeverage: int = 0

	# ───────────── Structured sub-sections ─────────────
	totBevOther: Optional[Any] = None
	totBevCounters: Optional[Any] = None
	totBevUsage: Optional[Any] = None
	waterQuantities: Optional[Any] = None

	@staticmethod
	def _key_map() -> Dict[str, str]:
		"""Map raw Ayla property names → field names for numeric counters."""
		return {
			"d701_tot_bev_b": "CoffeeBeans",
			"d704_tot_bev_espressi": "EspressoTotal",
			"d705_tot_id1_espr": "Espresso",
			"d706_tot_id2_coffee": "Coffee",
			"d707_tot_id3_long": "Long",
			"d708_tot_id5_doppio_p": "DoppioPlus",
			"d709_id6_americano": "Americano",
			"d710_tot_id7_capp": "Cappuccino",
			"d711_id8_lattmacc": "LatteMacchiato",
			"d712_id9_cafflatt": "CaffeLatte",
			"d713_id10_flatwhite": "FlatWhite",
			"d714_id11_esprmacc": "EspressoMacchiato",
			"d715_id12_hotmilk": "HotMilk",
			"d716_id13_cappdoppio_p": "CappuccinoDoppioPlus",
			"d717_id15_caprev": "CappuccinoReverse",
			"d718_id16_hotwater": "HotWater",
			"d719_id22_tea": "Tea",
			"d720_tot_id23_coffee_pot": "CoffeePot",
			"d730_tot_id27_brew_over_ice": "BrewOverIce",
			"d731_tot_mug_hot": "MugHot",
			"d732_tot_mug_cold": "MugCold",
			"d735_iced_bev": "IcedBeverage",
			"d736_mug_bev": "MugBeverage",
			"d737_mug_iced_bev": "MugIcedBeverage",
			"d738_cold_brew_bev": "ColdBrew",
			"d739_taste_bev": "TasteBeverage",
		}

	# ───────────── Factory ─────────────

	@classmethod
	def from_properties(cls, properties: Dict[str, Dict[str, Any]]) -> "BeverageCounters":
		"""Build BeverageCounters from a flat dict of properties."""
		keymap = cls._key_map()
		data: Dict[str, Any] = {}

		# map numeric counters directly
		for key, attr in keymap.items():
			val = properties.get(key, {}).get("value")
			if isinstance(val, (int, float)) or (isinstance(val, str) and val.isdigit()):
				data[attr] = int(val)

		# structured sub-sections (JSON-based)
		if "d702_tot_bev_other" in properties:
			data["totBevOther"] = TotBevOther.from_properties(properties)

		if "d733_tot_bev_counters" in properties:
			data["totBevCounters"] = TotBevCounters.from_properties(properties)

		if "d734_tot_bev_usage" in properties:
			data["totBevUsage"] = TotBevUsage.from_properties(properties)

		if "d740_water_qty_bev" in properties:
			data["waterQuantities"] = WaterQuantityBeverage.from_properties(properties)

		return cls(**data)

	# ───────────── Derived aggregates ─────────────

	def aggregates(self) -> Dict[str, Any]:
		"""Return aggregate counts for convenience."""
		return {
			"espresso_like": self.EspressoTotal + self.Espresso,
			"milk_based": (
				self.Cappuccino
				+ self.LatteMacchiato
				+ self.CaffeLatte
				+ self.HotMilk
			),
			"hot_vs_cold": {"hot": self.MugHot, "cold": self.MugCold},
		}

	def to_dict(self, include_aggregates: bool = True) -> Dict[str, Any]:
		"""Return dictionary representation, including nested sub-objects."""
		data = asdict(self)
		if include_aggregates:
			data["_aggregates"] = self.aggregates()
		return data

@dataclass
class ServiceCounters:
	"""Represents maintenance and service counters."""

	# ─────────── Base numeric counters ───────────
	WaterSinceDescale: Optional[int] = None                # d550_water_calc_qty
	GroundsTrayEmpties: Optional[int] = None               # d551_cnt_coffee_fondi
	TotalDescalingCycles: Optional[int] = None             # d552_cnt_calc_tot
	LifetimeWaterUsage: Optional[int] = None               # d553_water_tot_qty
	FiltersReplaced: Optional[int] = None                  # d554_cnt_filter_tot
	WaterThroughFilter: Optional[int] = None               # d555_water_filter_qty
	WaterHardnessLevel: Optional[int] = None               # d556_water_hardness
	BeveragesSinceDescaleWarn: Optional[int] = None        # d558_bev_cnt_desc_on
	GroundsFullnessPercent: Optional[int] = None           # d510_ground_cnt_percentage
	DescaleProgressPercent: Optional[int] = None           # d512_percentage_to_deca
	FilterUsagePercent: Optional[int] = None               # d513_percentage_usage_fltr
	DescaleAlertsTriggered: Optional[int] = None           # d524_ix_calcare_alm_qty
	BrewGroupMaxPosition: Optional[int] = None             # d525_max_pos

	# ─────────── Extended service parameters (from d580/d581 JSON) ───────────
	WaterMisuseCalcAbsQty: Optional[int] = None
	DescaleStatus: Optional[int] = None
	Last4WaterCalcQty: Optional[int] = None
	Last4CalcThreshold: Optional[int] = None
	WaterSteamerCalcRelQty: Optional[int] = None
	WaterHeaterCalcAbsQty: Optional[int] = None
	WaterSteamerCalcAbsQty: Optional[int] = None
	WaterColdBranchCalcRelQty: Optional[int] = None
	WaterColdBranchCalcAbsQty: Optional[int] = None

	# ─────────── Derived ───────────
	DerivedLifetimeLiters: Optional[float] = None

	@staticmethod
	def _to_int(val: Any) -> Optional[int]:
		"""Convert to int if possible."""
		if isinstance(val, (int, float)):
			return int(val)
		if isinstance(val, str) and val.strip().isdigit():
			return int(val.strip())
		return None

	# ─────────── Main Factory ───────────
	@classmethod
	def from_properties(cls, properties: Dict[str, Dict[str, Any]]) -> "ServiceCounters":
		"""Build ServiceCounters directly from flat Ayla property dictionary."""
		get = lambda key: properties.get(key, {}).get("value")

		obj = cls(
			WaterSinceDescale=cls._to_int(get("d550_water_calc_qty")),
			GroundsTrayEmpties=cls._to_int(get("d551_cnt_coffee_fondi")),
			TotalDescalingCycles=cls._to_int(get("d552_cnt_calc_tot")),
			LifetimeWaterUsage=cls._to_int(get("d553_water_tot_qty")),
			FiltersReplaced=cls._to_int(get("d554_cnt_filter_tot")),
			WaterThroughFilter=cls._to_int(get("d555_water_filter_qty")),
			WaterHardnessLevel=cls._to_int(get("d556_water_hardness")),
			BeveragesSinceDescaleWarn=cls._to_int(get("d558_bev_cnt_desc_on")),
			GroundsFullnessPercent=cls._to_int(get("d510_ground_cnt_percentage")),
			DescaleProgressPercent=cls._to_int(get("d512_percentage_to_deca")),
			FilterUsagePercent=cls._to_int(get("d513_percentage_usage_fltr")),
			DescaleAlertsTriggered=cls._to_int(get("d524_ix_calcare_alm_qty")),
			BrewGroupMaxPosition=cls._to_int(get("d525_max_pos")),
		)

		# Merge both service parameter JSON blobs if present
		merged_params: Dict[str, Any] = {}
		for key in ("d580_service_parameters", "d581_service_parameters"):
			js = _parse_json_safe(get(key))
			if js:
				merged_params.update(js)

		# Populate known fields
		mapping = {
			"water_misuse_calc_abs_qty": "WaterMisuseCalcAbsQty",
			"descale_status": "DescaleStatus",
			"last_4_water_calc_qty": "Last4WaterCalcQty",
			"last_4_calc_threshold": "Last4CalcThreshold",
			"water_steamer_calc_rel_qty": "WaterSteamerCalcRelQty",
			"water_heater_calc_abs_qty": "WaterHeaterCalcAbsQty",
			"water_steamer_calc_abs_qty": "WaterSteamerCalcAbsQty",
			"water_cold_branch_calc_rel_qty": "WaterColdBranchCalcRelQty",
			"water_cold_branch_calc_abs_qty": "WaterColdBranchCalcAbsQty",
		}

		for k, v in merged_params.items():
			attr = mapping.get(k.lower())
			if attr and hasattr(obj, attr):
				val = cls._to_int(v)
				if val is not None:
					setattr(obj, attr, val)

		# Compute derived liters
		if isinstance(obj.LifetimeWaterUsage, (int, float)):
			obj.DerivedLifetimeLiters = round(obj.LifetimeWaterUsage / 1000, 3)

		return obj

	def to_dict(self) -> Dict[str, Any]:
		"""Return a serializable dictionary."""
		return asdict(self)
	

@dataclass
class WaterQuantityBeverage:
	"""Represents per-beverage water quantity statistics (decoded from d740_water_qty_bev)."""

	BaseWaterQty: Optional[int] = None              # b_water_qty
	BrewOverCoffeeWaterQty: Optional[int] = None    # bw_coff_water_wty
	BrewOverMilkTimeQty: Optional[int] = None       # bw_milk_time_qty
	EspressiWaterQty: Optional[int] = None          # espressi_water_qty
	EspressoWaterQty: Optional[int] = None          # id1_espr_water_qty
	CoffeeWaterQty: Optional[int] = None            # id2_coffee_water_qty

	@classmethod
	def from_properties(cls, properties: Dict[str, Dict[str, Any]]) -> "WaterQuantityBeverage":
		"""Build WaterQuantityBeverage from a flat property dict."""
		raw_value = properties.get("d740_water_qty_bev", {}).get("value")
		js =_parse_json_safe(raw_value)
		if not js:
			return cls()

		mapping = {
			"b_water_qty": "BaseWaterQty",
			"bw_coff_water_wty": "BrewOverCoffeeWaterQty",
			"bw_milk_time_qty": "BrewOverMilkTimeQty",
			"espressi_water_qty": "EspressiWaterQty",
			"id1_espr_water_qty": "EspressoWaterQty",
			"id2_coffee_water_qty": "CoffeeWaterQty",
		}

		obj = cls()
		for k, v in js.items():
			attr = mapping.get(k.lower())
			if attr and hasattr(obj, attr):
				try:
					setattr(obj, attr, int(v))
				except (TypeError, ValueError):
					pass

		return obj

	def to_dict(self) -> Dict[str, Any]:
		"""Return a serializable dictionary representation."""
		return asdict(self)

class DeLonghiDevice:
	"""
	Represents a single DeLonghi Coffee Link device registered via Ayla Cloud.

	This class provides helper methods for fetching device properties, decoding
	structured property values, and resolving device metadata such as model and
	serial number.

	Attributes:
		_api (AylaClient): The parent API instance for network access.
		_data (Dict[str, Any]): Raw device data returned by Ayla.
		_properties_cache (Optional[List[Dict[str, Any]]]): Cached property list.

	Lazily resolved metadata:
		serial (Optional[str])
		product_code (Optional[str])
		friendly_model_name (Optional[str])
		model_type (Optional[str])
		app_model_id (Optional[str])
		image_url (Optional[str])
		image2_url (Optional[str])
		protocol_version (Optional[str])
		recipes (Optional[Any])
		type (Optional[str])
	"""

	def __init__(self, api: Any, data: Dict[str, Any]):
		"""
		Initialize a new DeLonghi device wrapper.

		Args:
			api (AylaClient): The parent authenticated API instance.
			data (Dict[str, Any]): The raw JSON device entry from Ayla.
		"""

		self.log = getattr(api, "log", make_default_logger("DeLonghiAPI.Device"))
		self._api = api
		self._data = data
		self._properties_cache:Dict[str, Any] = None

		# Lazily resolved metadata
		self.serial: Optional[str] = None
		self.product_code: Optional[str] = None
		self.friendly_model_name: Optional[str] = None
		self.model_type: Optional[str] = None
		self.app_model_id: Optional[str] = None
		self.image_url: Optional[str] = None
		self.image2_url: Optional[str] = None
		self.protocol_version: Optional[str] = None
		self.recipes: Optional[Any] = None
		self.type: Optional[str] = None

		self.update_properties()


		self.machine_settings: MachineSettings = MachineSettings.from_properties(self._properties_cache)
		self.beverage_counters = BeverageCounters.from_properties(self._properties_cache)
		self.service_counters: ServiceCounters = ServiceCounters.from_properties(self._properties_cache)
		self.water_quantity_beverage: WaterQuantityBeverage = WaterQuantityBeverage.from_properties(self._properties_cache)

		self._last_properties_fetch: float = 0.0
		self._properties_cache_ttl: int = 60  # seconds
	
	_SERIAL_RE = re.compile(r"(\d{6})([A-Z0-9]{2})(\d{2})(\d{2})(\d{2})([A-Z0-9]{5})")
	_serial_cache: Optional[Dict[str, Any]] = None

	def __repr__(self) -> str:
		"""Return a concise string representation of the device."""
		return f"<DeLonghiDevice {self.dsn} ({self.model}) {self.connection_status}>"

	@property
	def key(self) -> Optional[str]:
		"""Return the device key (unique identifier)."""
		return self._data.get("key")

	@property
	def serial_meta(self) -> Dict[str, Any]:
		"""Return parsed serial components (date, sku, etc)."""
		return getattr(self, "_serial_meta", {})

	@property
	def serial_production_date(self) -> Optional[datetime.date]:
		"""Return production date extracted from serial number, if available."""
		return self.serial_meta.get("production_date")

	@property
	def dsn(self) -> Optional[str]:
		"""Return the device DSN (Device Serial Number)."""
		return self._data.get("dsn")

	@property
	def model(self) -> Optional[str]:
		"""Return the reported model name."""
		return self._data.get("model")

	@property
	def oem_model(self) -> Optional[str]:
		"""Return the OEM model name, if provided."""
		return self._data.get("oem_model")

	@property
	def product_name(self) -> Optional[str]:
		"""Return the marketing product name, if available."""
		return self._data.get("product_name")

	@property
	def sw_version(self) -> Optional[str]:
		"""Return the software/firmware version string."""
		return self._data.get("sw_version")

	@property
	def lan_ip(self) -> Optional[str]:
		"""Return the device's LAN IP address (if in local mode)."""
		return self._data.get("lan_ip")

	@property
	def lan_enabled(self) -> bool:
		"""Return True if the device supports direct LAN communication."""
		return self._data.get("lan_enabled", False)

	@property
	def connection_status(self) -> Optional[str]:
		"""Return the device’s current Ayla Cloud connection status."""
		return self._data.get("connection_status")
	
	@property
	def software_version(self) -> Optional[str]:
		"""Return the reported software version string (if available)."""
		prop = self._properties_cache.get("software_version")
		if isinstance(prop, dict):
			val = prop.get("value")
			return str(val).strip() if val else None
		return None

	def _decode_serial(self) -> None:
		"""Decode base64 serial number and cache it internally."""
		if self._serial_cache is not None:
			return  # already cached

		entry = self._properties_cache.get("d270_serialnumber")
		serial_b64 = entry.get("value") if isinstance(entry, dict) else entry
		if not serial_b64:
			self._serial_cache = {}
			return

		try:
			ascii_block = base64.b64decode(serial_b64).decode(errors="ignore")
			# Clean up artifacts like trailing slashes, nulls, or control chars
			ascii_block = re.sub(r"[^A-Z0-9]", "", ascii_block.upper())
		except Exception:
			self._serial_cache = {}
			return

		m = self._SERIAL_RE.search(ascii_block)
		if not m:
			self._serial_cache = {"raw_ascii": ascii_block or None}
			return
	
		sku, unk2, yy, mm, dd, tail = m.groups()
		try:
			year = 2000 + int(yy)
			date = datetime.date(year, int(mm), int(dd))
		except Exception:
			date = None

		self._serial_cache = {
			"sku_last6": sku,
			"unknown2": unk2,
			"yy": yy,
			"mm": mm,
			"dd": dd,
			"tail5": tail,
			"production_date": date,
			"production_tuple": (int(yy), int(mm), int(dd)),
			"raw_ascii": ascii_block,
		}

	# ──────────────────────────────
	# Public serial properties
	# ──────────────────────────────
	@property
	def production_date(self) -> Optional[datetime.date]:
		return self._serial_cache.get("production_date") if self._serial_cache else None

	@property
	def sku_last6(self) -> Optional[str]:
		return self._serial_cache.get("sku_last6") if self._serial_cache else None

	@property
	def tail5(self) -> Optional[str]:
		return self._serial_cache.get("tail5") if self._serial_cache else None

	@property
	def production_tuple(self) -> Optional[tuple]:
		return self._serial_cache.get("production_tuple") if self._serial_cache else None

	@property
	def raw_serial_ascii(self) -> Optional[str]:
		return self._serial_cache.get("raw_ascii") if self._serial_cache else None
	
	@staticmethod
	def _normalize_oem_model(oem: Optional[str]) -> Optional[str]:
		if not oem:
			return None
		om = oem.strip().upper().replace("DL-", "").replace("_", "-")
		return om.replace("STRIKER-CB", "STRIKER_COLD-BREW")

	def _match_table_entry(self) -> Optional[Dict[str, Any]]:
		"""
		Find a matching machine entry in the Reply.it transcode table based on
		the cached serial number and OEM model.
		"""
		table = self._api.get_transcode_table() or {}
		machines = table.get("machines", []) or []

		ascii_upper = (self.raw_serial_ascii or "").upper()
		product_suffix = (self.sku_last6 or "")[-5:]

		# 1️⃣ Match by last 5 digits of product code inside serial ASCII
		matches = [m for m in machines if str(m.get("product_code", ""))[-5:] in ascii_upper]
		if not matches:
			return None

		# 2️⃣ If multiple matches, try normalized OEM model (appModelId)
		if len(matches) > 1:
			desired = self._normalize_oem_model(self.oem_model)
			if desired:
				for m in matches:
					app_model = str(m.get("appModelId", "")).upper()
					if app_model == desired:
						return m

		# 3️⃣ Fallback: return first candidate
		return matches[0]
	
	def _resolve_from_serial_and_table(self) -> None:
		"""Populate known fields using the serial info and the transcode table."""
		self._decode_serial()  # ensure serial cache is populated

		entry = self._match_table_entry()
		if not entry:
			self.log.debug("[→] No matching transcode table entry found.")
			return

		self.product_code = str(entry.get("product_code") or "").strip() or None
		self.friendly_model_name = entry.get("name")
		self.model_type = entry.get("type")
		self.app_model_id = entry.get("appModelId")
		self.image_url = entry.get("image_url") or entry.get("imageUrl")
		self.image2_url = entry.get("image2_url") or entry.get("image2Url")
		self.protocol_version = entry.get("protocolVersion")
		self.recipes = entry.get("recipes")

		self.log.debug(f"[✓] Matched transcode entry: {self.friendly_model_name} ({self.product_code})")

	# ─────────────────────── Property Fetch Helpers ───────────────────────

	def update_properties(self) -> None:
		"""
		Ensure that device properties are loaded and fresh.

		Automatically fetches properties if none are cached or if the cache
		is older than `_properties_cache_ttl` seconds.
		"""
		now = time.time()
		if (self._properties_cache is None or (now - self._last_properties_fetch) > self._properties_cache_ttl):
			self.log.debug("[→] Refreshing properties cache (stale or missing).")
			self.get_properties(refresh=True)
			self._resolve_from_serial_and_table()

	def _sanitize_properties(self, props: List[Dict[str, Any]]) -> Dict[str, Any]:
		sanitzedDict = {}
		name_count = {}
		
		for prop_entry in props:
			original_name = prop_entry['property']['name']
			
			# Generate unique name with counter
			if original_name not in name_count:
				name_count[original_name] = 1
				final_name = original_name
			else:
				name_count[original_name] += 1
				final_name = f"{original_name}_dup{name_count[original_name]}"
			
			del prop_entry['property']['name']
			sanitzedDict[final_name] = prop_entry['property']
		
		return sanitzedDict


	def get_properties(self, refresh: bool = False) -> Dict[str, Any]:
		"""
		Fetch all device properties from Ayla Cloud.

		If cached properties exist and `refresh=False`, the cached copy is used.
		Otherwise, a fresh API request is made.

		Args:
			refresh (bool, optional): If True, forces a full reload from the API.

		Returns:
			List[Dict[str, Any]]: A list of property entries.
		"""
		if self._properties_cache and not refresh:
			return self._properties_cache
		props = self._api.ayla_list_device_properties(self.key)
		props = self._sanitize_properties(props)
		self._properties_cache = props
		self._last_properties_fetch = time.time()
		return props
	
	def _decode_utf16_b64(self, name: str) -> list[str] | None:
		"""
		Decode Base64-encoded UTF-16LE profile-name blocks (d051_profile_name1_3, d052_profile_name4).
		Format:
		[7-byte header] + UTF-16LE names + separators (\x08\xff\x00 | \x01\xff\x00 | \xff\xff) + [tiny footer]
		"""
		import base64, re

		# Fetch value from cache (supports either dict entry or raw string)
		entry = self._properties_cache.get(name)
		val = entry.get("value") if isinstance(entry, dict) else (entry if isinstance(entry, str) else None)
		if not val:
			return None

		try:
			raw = base64.b64decode(val)
		except Exception:
			return None

		# 1) Strip fixed 7-byte header if present
		if len(raw) >= 7:
			raw = raw[7:]

		# 2) Strip common tiny footers/trailers observed in the wild
		raw = raw.rstrip(b"\x00\xff\xd8\x81`+")

		# 3) Split strictly on delimiters that separate names
		sep = re.compile(b"(?:\x08\xff\x00|\x01\xff\x00|\xff\xff)")
		chunks = [c for c in sep.split(raw) if c]

		names: list[str] = []
		for c in chunks:
			# Do NOT strip trailing zeros (they belong to UTF-16LE pairs).
			# Only strip leading pad zeros, then ensure even length.
			c = c.lstrip(b"\x00")
			if not c:
				continue
			if len(c) % 2 == 1:
				c += b"\x00"

			try:
				txt = c.decode("utf-16le", errors="ignore")
			except Exception:
				continue

			# Tidy spacing/nulls; keep digits and normal letters
			txt = txt.replace("\x00", "").strip()
			txt = re.sub(r"\s{2,}", " ", txt)
			if txt:
				names.append(txt)

		return names or None

	@property
	def profile_names(self) -> List[str]:
		"""Return list of user profile names (decoded and cleaned)."""
		names = []
		part1 = self._decode_utf16_b64("d051_profile_name1_3") or []
		part4 = self._decode_utf16_b64("d052_profile_name4") or []
		names.extend(part1)
		names.extend(part4)
		return names

	@property
	def profile_name_1(self) -> Optional[str]:
		return self.profile_names[0] if len(self.profile_names) > 0 else None

	@property
	def profile_name_2(self) -> Optional[str]:
		return self.profile_names[1] if len(self.profile_names) > 1 else None

	@property
	def profile_name_3(self) -> Optional[str]:
		return self.profile_names[2] if len(self.profile_names) > 2 else None

	@property
	def profile_name_4(self) -> Optional[str]:
		return self.profile_names[3] if len(self.profile_names) > 3 else None

	@property
	def device_connected(self) -> bool | None:
		"""Return True if the app reports device connected, False if not, None if unknown."""
		prop = self._properties_cache.get("app_device_connected")
		if not isinstance(prop, dict):
			return None
		val = prop.get("value")
		if not val:
			return None
		try:
			raw = base64.b64decode(val)
			# Connected if any non-zero byte
			return any(b != 0 for b in raw)
		except Exception:
			return None


	@property
	def device_status(self) -> str | None:
		"""Return current device status (RUN, IDLE, etc.)."""
		prop = self._properties_cache.get("app_device_status")
		val = prop.get("value") if isinstance(prop, dict) else None
		return str(val).strip() if val else None

	def _decode_beansystem_slot(self, index: int) -> Optional[str]:
		"""Decode a single base64 bean system slot into readable ASCII text."""
		name = f"d25{index}_beansystem_{index}"
		prop = self._properties_cache.get(name)
		if not isinstance(prop, dict):
			return None
		val = prop.get("value")
		if not val:
			return None
		try:
			return base64.b64decode(val).decode("utf-16", errors="ignore").strip("\x00")
		except Exception:
			try:
				return base64.b64decode(val).decode(errors="ignore").strip("\x00")
			except Exception:
				return None

	@property
	def bean_system_slots(self) -> Dict[str, Optional[str]]:
		"""Return all bean system slot texts (decoded)."""
		slots = {}
		for i in range(7):
			txt = self._decode_beansystem_slot(i)
			if txt:
				slots[str(i)] = txt
		return slots

	@property
	def bean_system_parameters(self) -> Optional[str]:
		"""Return decoded bean system parameter block."""
		prop = self._properties_cache.get("d260_beansystem_par")
		if not isinstance(prop, dict):
			return None
		val = prop.get("value")
		if not val:
			return None
		try:
			return base64.b64decode(val).decode(errors="ignore").strip("\x00")
		except Exception:
			return None

	# ─────────── User Settings ───────────


	def _decode_user_flags(self) -> int:
		"""Return raw user flag byte (cached)."""
		prop = self._properties_cache.get("d284_mach_sett_user_conf")
		if not isinstance(prop, dict):
			return 0
		val = prop.get("value")
		if not val:
			return 0
		try:
			b = base64.b64decode(val)
			return b[8] if len(b) >= 9 else 0
		except Exception:
			return 0

	@property
	def beep_enabled(self) -> bool:
		"""True if acoustic signals are enabled."""
		flags = self._decode_user_flags()
		return bool(flags & 0b00000100)

	@property
	def energy_saving_enabled(self) -> bool:
		"""True if energy-saving mode is enabled."""
		flags = self._decode_user_flags()
		return bool(flags & 0b00010000)

	# ─────────── Structured Property Decoding ───────────

	def _decode_param_array(self, data_b64: Optional[str]) -> Optional[List[Dict[str, Any]]]:
		"""
		Decode a base64-encoded parameter array containing 12-byte blocks.

		Each block encodes a header and four 16-bit values (H, param1–3).

		Args:
			data_b64 (Optional[str]): Base64-encoded binary payload.

		Returns:
			Optional[List[Dict[str, Any]]]: List of decoded parameter entries,
			or None if decoding fails.
		"""
		if not data_b64:
			return None
		try:
			data = base64.b64decode(data_b64)
		except Exception:
			self.log.debug("[!] Base64 decode failed for parameter array")
			return None

		out: List[Dict[str, Any]] = []
		for i in range(0, len(data) - 11, 12):
			try:
				hdr, val1, p1, p2, p3 = struct.unpack_from("<4sHHHH", data, i)
				out.append({
					"block": i // 12,
					"raw_header": hdr.hex(),
					"value": val1,
					"param1": p1,
					"param2": p2,
					"param3": p3,
				})
			except struct.error:
				break

		self.log.debug(f"[→] Decoded {len(out)} blocks from parameter array")
		return out or None

	def print_summary(self) -> None:
		"""
		Print a detailed, colorized, CLI-friendly summary of the device state.

		Includes identification, connection, software, user settings,
		beverage counters, and service statistics.
		"""
		import shutil
		# ───────────── Color + formatting helpers ─────────────
		def color(txt, fg=None, style=None):
			colors = {
				"red": 31, "green": 32, "yellow": 33, "blue": 34,
				"magenta": 35, "cyan": 36, "white": 37, "grey": 90
			}
			styles = {"bold": 1, "dim": 2}
			cseq = []
			if fg in colors:
				cseq.append(str(colors[fg]))
			if style in styles:
				cseq.append(str(styles[style]))
			if not cseq:
				return txt
			return f"\033[{';'.join(cseq)}m{txt}\033[0m"

		def section(title: str):
			width = shutil.get_terminal_size((100, 20)).columns
			line = color("─" * (width - len(title) - 3), "grey")
			print(color(f"\n{title} ", "cyan", "bold") + line)

		def kv(label: str, val: Any, indent=2, colorize=True):
			prefix = " " * indent + color(f"{label:<28}", "blue", "bold")
			if val is None or val == "":
				print(f"{prefix}{color('—', 'grey')}")
			elif isinstance(val, (dict, list)):
				js = json.dumps(val, indent=indent + 2, ensure_ascii=False)
				print(f"{prefix}\n{color(js, 'grey')}")
			else:
				if colorize and isinstance(val, (int, float)):
					print(f"{prefix}{color(str(val), 'yellow')}")
				elif isinstance(val, bool):
					col = "green" if val else "red"
					print(f"{prefix}{color(str(val), col, 'bold')}")
				else:
					print(f"{prefix}{str(val)}")

		# ───────────── Section: Identification ─────────────
		section("IDENTIFICATION")
		kv("Model name", getattr(self, "friendly_model_name", None))
		kv("Model type", getattr(self, "model_type", None))
		kv("App model ID", getattr(self, "app_model_id", None))
		kv("Product code", getattr(self, "product_code", None))
		kv("DSN", getattr(self, "dsn", None))
		kv("Serial (ASCII)", getattr(self, "raw_serial_ascii", None))
		kv("SKU last 6", self.sku_last6)
		kv("Tail code", self.tail5)
		kv("Production date", self.production_date)

		# ───────────── Section: Connection ─────────────
		section("CONNECTION")
		kv("Device connected", getattr(self, "device_connected", None))
		kv("Device status", getattr(self, "device_status", None))

		# ───────────── Section: Software ─────────────
		section("SOFTWARE / CONFIGURATION")
		kv("Software version", getattr(self, "sw_version", None))
		kv("Beep enabled", getattr(self, "beep_enabled", None))
		kv("Energy saving", getattr(self, "energy_saving_enabled", None))

		# ───────────── Section: Machine Settings ─────────────
		if getattr(self, "machine_settings", None):
			section("MACHINE SETTINGS")
			ms = self.machine_settings.to_dict()
			for k, v in ms.items():
				kv(k, v)

		# ───────────── Section: Beverage Counters ─────────────
		if getattr(self, "beverage_counters", None):
			section("BEVERAGE COUNTERS")
			bc = self.beverage_counters.to_dict(include_aggregates=True)
			for k, v in bc.items():
				kv(k, v)

		# ───────────── Section: Service Counters ─────────────
		if getattr(self, "service_counters", None):
			section("SERVICE COUNTERS")
			sc = self.service_counters.to_dict()
			for k, v in sc.items():
				kv(k, v)

		# ───────────── Section: Water Quantities ─────────────
		if getattr(self, "water_quantity_beverage", None):
			section("WATER QUANTITIES PER BEVERAGE")
			wq = self.water_quantity_beverage.to_dict()
			for k, v in wq.items():
				kv(k, v)

		# ───────────── Section: Bean System ─────────────
		if getattr(self, "bean_system", None):
			section("BEAN SYSTEM")
			bs = self.bean_system
			if isinstance(bs, dict):
				for k, v in bs.items():
					kv(k, v)
			else:
				print(bs)

		print()  # final newline