Files
home-assistant-arkteos/custom_components/arkteos/diagnostics.py
T
2026-07-21 12:25:44 +02:00

127 lines
4.0 KiB
Python

"""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"),
},
}