"""Diagnostics sans accès réseau pour l'intégration Arkteos.""" from __future__ import annotations import json from collections.abc import Mapping from pathlib import Path from typing import TYPE_CHECKING, Any from homeassistant.components.diagnostics import async_redact_data from . import _get_client from .const import CONF_HOST, CONF_PORT, DEFAULT_PORT, DOMAIN if TYPE_CHECKING: from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant _SENSITIVE_KEYS = frozenset({"host", "password", "token"}) _CLIENT_STATE_ATTRIBUTES = ( "available", "running", "connected", "bytes_received", "frames_received", "frames_rejected", "reconnect_count", "last_frame_type", ) _MISSING = object() def _manifest_version() -> str | None: """Retourne la version déclarée sans contacter de ressource externe.""" try: manifest = json.loads(Path(__file__).with_name("manifest.json").read_text()) except (OSError, json.JSONDecodeError): return None version = manifest.get("version") return version if isinstance(version, str) else None def _simple_values(data: object) -> dict[str, str | int | float | bool | None]: """Conserve uniquement les valeurs décodées, simples et non sensibles.""" if not isinstance(data, dict): return {} result: dict[str, str | int | float | bool | None] = {} for key, value in data.items(): if not isinstance(key, str) or key.lower() in _SENSITIVE_KEYS: continue if value is None or isinstance(value, (str, int, float, bool)): result[key] = value return result def _received_data(client: object | None, attribute: str) -> dict[str, Any]: """Prépare une section de données sans bytes, buffer ni objet asyncio.""" data = getattr(client, attribute, None) if client is not None else None values = _simple_values(data) return {"received": data is not None, "data": values} def _entry_state(entry: ConfigEntry) -> str | None: """Convertit l'état optionnel de l'entrée en valeur sérialisable.""" state = getattr(entry, "state", None) if state is None: return None value = getattr(state, "value", state) return value if isinstance(value, str) else None async def async_get_config_entry_diagnostics( hass: HomeAssistant, entry: ConfigEntry, ) -> dict[str, Any]: """Retourne des diagnostics locaux, redacted et uniquement sérialisables. Le client conserve uniquement ``frames_rejected`` de manière agrégée : les valeurs rejetées par champ ne sont donc pas inventées dans ce résultat. """ entry_data = getattr(entry, "data", {}) if not isinstance(entry_data, Mapping): entry_data = {} host = entry_data.get(CONF_HOST) title = getattr(entry, "title", None) if isinstance(title, str) and isinstance(host, str): title = title.replace(host, "**REDACTED**") configured = async_redact_data( { CONF_HOST: host, CONF_PORT: entry_data.get(CONF_PORT, DEFAULT_PORT), }, [CONF_HOST], ) client = _get_client(hass, entry) client_state: dict[str, str | int | float | bool | None] = {} if client is not None: for attribute in _CLIENT_STATE_ATTRIBUTES: value = getattr(client, attribute, _MISSING) if value is _MISSING: continue if value is None or isinstance(value, (str, int, float, bool)): client_state[attribute] = value return { "integration": { "domain": DOMAIN, "version": _manifest_version(), "title": title if isinstance(title, str) else None, "config_entry_state": _entry_state(entry), "configuration": configured, }, "client": client_state, "received_data": { "metadata": _received_data(client, "latest_metadata"), "frigo": _received_data(client, "latest_frigo_data"), "regulation": _received_data(client, "latest_regulation_data"), }, }