Add native Arkteos sensors

This commit is contained in:
2026-07-21 12:05:30 +02:00
parent b76e8f3896
commit 8bae450bc3
7 changed files with 761 additions and 2 deletions
+1 -1
View File
@@ -7,5 +7,5 @@ DOMAIN = "arkteos"
DEFAULT_PORT = 9641 DEFAULT_PORT = 9641
CONF_HOST = "host" CONF_HOST = "host"
CONF_PORT = "port" CONF_PORT = "port"
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR] PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
RUNTIME_DATA = "runtime_data" RUNTIME_DATA = "runtime_data"
+396
View File
@@ -0,0 +1,396 @@
"""Capteurs natifs en lecture seule pour les données Arkteos."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Final, Literal
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
EntityCategory,
PERCENTAGE,
REVOLUTIONS_PER_MINUTE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
UnitOfElectricPotential,
UnitOfFrequency,
UnitOfPower,
UnitOfPressure,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import _get_client
from .client import ArkteosClient
from .const import DOMAIN
from .entity import ArkteosEntity
FRIGO_FRAME_TYPE: Final = "frigo"
REGULATION_FRAME_TYPE: Final = "regulation"
FRIGO_STATUS_OPTIONS: Final = (
"Arret",
"Refroidissement",
"Chauffage",
"Degivrage",
"Inconnu",
)
PAC_STATUS_OPTIONS: Final = (
"Arret",
"Attente",
"Chaud",
"Froid",
"Hors Gel",
"Ext Chaud",
"Ext Froid",
"Chaud Froid",
"ECS",
"Piscine",
"Inconnu",
)
@dataclass(frozen=True, kw_only=True)
class ArkteosSensorDescription(SensorEntityDescription):
"""Description d'un capteur et du type de trame qui l'alimente."""
frame_type: Literal["frigo", "regulation"]
FRIGO_SENSORS: Final[tuple[ArkteosSensorDescription, ...]] = (
ArkteosSensorDescription(
key="exterieur_temp",
translation_key="exterieur_temp",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=1,
),
ArkteosSensorDescription(
key="nb_degivrages",
translation_key="nb_degivrages",
frame_type=FRIGO_FRAME_TYPE,
state_class=SensorStateClass.TOTAL_INCREASING,
),
ArkteosSensorDescription(
key="temps_compresseur",
translation_key="temps_compresseur",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
ArkteosSensorDescription(
key="nb_cycles_compresseur",
translation_key="nb_cycles_compresseur",
frame_type=FRIGO_FRAME_TYPE,
state_class=SensorStateClass.TOTAL_INCREASING,
),
ArkteosSensorDescription(
key="freq_comp_actuelle",
translation_key="freq_comp_actuelle",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.FREQUENCY,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="freq_comp_cible",
translation_key="freq_comp_cible",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.FREQUENCY,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="fan_speed_evaporator_1",
translation_key="fan_speed_evaporator_1",
frame_type=FRIGO_FRAME_TYPE,
native_unit_of_measurement=REVOLUTIONS_PER_MINUTE,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="dc_voltage",
translation_key="dc_voltage",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="statut_frigo_s",
translation_key="statut_frigo_s",
frame_type=FRIGO_FRAME_TYPE,
device_class=SensorDeviceClass.ENUM,
options=FRIGO_STATUS_OPTIONS,
),
ArkteosSensorDescription(
key="statut_frigo",
translation_key="statut_frigo",
frame_type=FRIGO_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
ArkteosSensorDescription(
key="active_error_fri",
translation_key="active_error_fri",
frame_type=FRIGO_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
REGULATION_SENSORS: Final[tuple[ArkteosSensorDescription, ...]] = (
ArkteosSensorDescription(
key="puissance_inst_produite",
translation_key="puissance_inst_produite",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="puissance_inst_consommee",
translation_key="puissance_inst_consommee",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="temps_mise_sous_tension",
translation_key="temps_mise_sous_tension",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
ArkteosSensorDescription(
key="modele_pac_s",
translation_key="modele_pac_s",
frame_type=REGULATION_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
),
ArkteosSensorDescription(
key="modele_pac",
translation_key="modele_pac",
frame_type=REGULATION_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
ArkteosSensorDescription(
key="primaire_temp_eau_aller_consigne",
translation_key="primaire_temp_eau_aller_consigne",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="primaire_temp_eau_aller",
translation_key="primaire_temp_eau_aller",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="primaire_temp_eau_retour",
translation_key="primaire_temp_eau_retour",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="primaire_debit_eau",
translation_key="primaire_debit_eau",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.VOLUME_FLOW_RATE,
native_unit_of_measurement="L/h",
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="primaire_pression",
translation_key="primaire_pression",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.PRESSURE,
native_unit_of_measurement=UnitOfPressure.BAR,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="primaire_circulateur_consigne",
translation_key="primaire_circulateur_consigne",
frame_type=REGULATION_FRAME_TYPE,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="zone1_temp_interieur",
translation_key="zone1_temp_interieur",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
ArkteosSensorDescription(
key="zone1_consigne",
translation_key="zone1_consigne",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False,
),
ArkteosSensorDescription(
key="ecs_temp_eau_milieu",
translation_key="ecs_temp_eau_milieu",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="ecs_temp_eau_bas",
translation_key="ecs_temp_eau_bas",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="ecs_consigne",
translation_key="ecs_consigne",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
ArkteosSensorDescription(
key="nb_cycles_compresseur_reg",
translation_key="nb_cycles_compresseur_reg",
frame_type=REGULATION_FRAME_TYPE,
state_class=SensorStateClass.TOTAL_INCREASING,
),
ArkteosSensorDescription(
key="statut_pac_s",
translation_key="statut_pac_s",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.ENUM,
options=PAC_STATUS_OPTIONS,
),
ArkteosSensorDescription(
key="statut_pac",
translation_key="statut_pac",
frame_type=REGULATION_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
ArkteosSensorDescription(
key="active_error_reg",
translation_key="active_error_reg",
frame_type=REGULATION_FRAME_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
),
ArkteosSensorDescription(
key="signal_rf_sonde_1",
translation_key="signal_rf_sonde_1",
frame_type=REGULATION_FRAME_TYPE,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
SENSOR_DESCRIPTIONS: Final = FRIGO_SENSORS + REGULATION_SENSORS
async def async_setup_entry(hass, entry, async_add_entities: AddEntitiesCallback) -> None:
"""Ajoute les capteurs alimentés par le client partagé de l'entrée."""
client = _get_client(hass, entry)
if client is None:
return
async_add_entities(ArkteosSensor(client, description) for description in SENSOR_DESCRIPTIONS)
class ArkteosSensor(ArkteosEntity, SensorEntity):
"""Expose une valeur validée issue d'un seul type de trame Arkteos."""
_attr_should_poll = False
def __init__(self, client: ArkteosClient, description: ArkteosSensorDescription) -> None:
super().__init__(client)
self.entity_description = description
self._attr_unique_id = f"arkteos_zuran4_{description.key}"
latest_data = self._latest_data()
self._has_received_frame = latest_data is not None
self._last_value = latest_data.get(description.key) if latest_data is not None else None
@property
def available(self) -> bool:
"""N'est disponible qu'après la première trame de son propre type."""
return self._client.available and self._has_received_frame
@property
def native_value(self) -> int | float | str | None:
"""Retourne la dernière valeur acceptée, sans accès réseau."""
return self._last_value
@property
def device_info(self) -> DeviceInfo:
"""Regroupe tous les capteurs sous l'unique PAC Arkteos."""
regulation_data = self._client.latest_regulation_data
model = "Zuran 4"
if regulation_data is not None:
detected_model = regulation_data.get("modele_pac_s")
if isinstance(detected_model, str):
model = detected_model
return DeviceInfo(
identifiers={(DOMAIN, "arkteos_zuran4")},
manufacturer="Arkteos",
model=model,
name="PAC Arkteos Zuran 4",
)
async def async_added_to_hass(self) -> None:
"""Abonne le capteur aux seules données transmises par le client."""
await super().async_added_to_hass()
self._client.add_data_callback(self._handle_data)
async def async_will_remove_from_hass(self) -> None:
"""Désabonne le callback de données avant la destruction."""
self._client.remove_data_callback(self._handle_data)
await super().async_will_remove_from_hass()
def _latest_data(self) -> dict[str, int | float | str] | None:
if self.entity_description.frame_type == FRIGO_FRAME_TYPE:
return self._client.latest_frigo_data
return self._client.latest_regulation_data
def _handle_data(
self,
frame_type: str,
data: dict[str, int | float | str],
_rejected: dict[str, int | float],
) -> None:
if frame_type != self.entity_description.frame_type:
return
self._has_received_frame = True
if self.entity_description.key in data:
self._last_value = data[self.entity_description.key]
self.async_write_ha_state()
+34
View File
@@ -36,6 +36,40 @@
"connection": { "connection": {
"name": "PAC connectée" "name": "PAC connectée"
} }
},
"sensor": {
"exterieur_temp": {"name": "Température extérieure"},
"nb_degivrages": {"name": "Nombre de dégivrages"},
"temps_compresseur": {"name": "Temps fonctionnement compresseur"},
"nb_cycles_compresseur": {"name": "Nombre de cycles compresseur"},
"freq_comp_actuelle": {"name": "Fréquence compresseur actuelle"},
"freq_comp_cible": {"name": "Fréquence compresseur cible"},
"fan_speed_evaporator_1": {"name": "Vitesse ventilateur groupe frigo"},
"dc_voltage": {"name": "Tension DC compresseur"},
"statut_frigo_s": {"name": "Statut frigorifique"},
"statut_frigo": {"name": "Code statut frigorifique"},
"active_error_fri": {"name": "Erreur active frigorifique"},
"puissance_inst_produite": {"name": "Puissance instantanée produite"},
"puissance_inst_consommee": {"name": "Puissance instantanée consommée"},
"temps_mise_sous_tension": {"name": "Temps mise sous tension"},
"modele_pac_s": {"name": "Modèle PAC"},
"modele_pac": {"name": "Code modèle PAC"},
"primaire_temp_eau_aller_consigne": {"name": "Consigne départ eau primaire"},
"primaire_temp_eau_aller": {"name": "Température eau primaire aller"},
"primaire_temp_eau_retour": {"name": "Température eau primaire retour"},
"primaire_debit_eau": {"name": "Débit eau primaire"},
"primaire_pression": {"name": "Pression eau primaire"},
"primaire_circulateur_consigne": {"name": "Circulateur primaire"},
"zone1_temp_interieur": {"name": "Température intérieure zone 1"},
"zone1_consigne": {"name": "Consigne température zone 1"},
"ecs_temp_eau_milieu": {"name": "Température ballon ECS milieu"},
"ecs_temp_eau_bas": {"name": "Température ballon ECS bas"},
"ecs_consigne": {"name": "Consigne ECS"},
"nb_cycles_compresseur_reg": {"name": "Cycles compresseur régulation"},
"statut_pac_s": {"name": "Statut PAC"},
"statut_pac": {"name": "Code statut PAC"},
"active_error_reg": {"name": "Erreur active régulation"},
"signal_rf_sonde_1": {"name": "Signal RF sonde zone 1"}
} }
} }
} }
@@ -36,6 +36,40 @@
"connection": { "connection": {
"name": "PAC connectée" "name": "PAC connectée"
} }
},
"sensor": {
"exterieur_temp": {"name": "Température extérieure"},
"nb_degivrages": {"name": "Nombre de dégivrages"},
"temps_compresseur": {"name": "Temps fonctionnement compresseur"},
"nb_cycles_compresseur": {"name": "Nombre de cycles compresseur"},
"freq_comp_actuelle": {"name": "Fréquence compresseur actuelle"},
"freq_comp_cible": {"name": "Fréquence compresseur cible"},
"fan_speed_evaporator_1": {"name": "Vitesse ventilateur groupe frigo"},
"dc_voltage": {"name": "Tension DC compresseur"},
"statut_frigo_s": {"name": "Statut frigorifique"},
"statut_frigo": {"name": "Code statut frigorifique"},
"active_error_fri": {"name": "Erreur active frigorifique"},
"puissance_inst_produite": {"name": "Puissance instantanée produite"},
"puissance_inst_consommee": {"name": "Puissance instantanée consommée"},
"temps_mise_sous_tension": {"name": "Temps mise sous tension"},
"modele_pac_s": {"name": "Modèle PAC"},
"modele_pac": {"name": "Code modèle PAC"},
"primaire_temp_eau_aller_consigne": {"name": "Consigne départ eau primaire"},
"primaire_temp_eau_aller": {"name": "Température eau primaire aller"},
"primaire_temp_eau_retour": {"name": "Température eau primaire retour"},
"primaire_debit_eau": {"name": "Débit eau primaire"},
"primaire_pression": {"name": "Pression eau primaire"},
"primaire_circulateur_consigne": {"name": "Circulateur primaire"},
"zone1_temp_interieur": {"name": "Température intérieure zone 1"},
"zone1_consigne": {"name": "Consigne température zone 1"},
"ecs_temp_eau_milieu": {"name": "Température ballon ECS milieu"},
"ecs_temp_eau_bas": {"name": "Température ballon ECS bas"},
"ecs_consigne": {"name": "Consigne ECS"},
"nb_cycles_compresseur_reg": {"name": "Cycles compresseur régulation"},
"statut_pac_s": {"name": "Statut PAC"},
"statut_pac": {"name": "Code statut PAC"},
"active_error_reg": {"name": "Erreur active régulation"},
"signal_rf_sonde_1": {"name": "Signal RF sonde zone 1"}
} }
} }
} }
+35
View File
@@ -0,0 +1,35 @@
"""Tests de la base d'entité Arkteos."""
from __future__ import annotations
from unittest.mock import Mock
from custom_components.arkteos.entity import ArkteosEntity
class FakeClient:
def __init__(self, available: bool = False) -> None:
self.available = available
self.callbacks: list[object] = []
def add_availability_callback(self, callback) -> None:
self.callbacks.append(callback)
def remove_availability_callback(self, callback) -> None:
self.callbacks.remove(callback)
def test_base_entity_availability_reflects_client() -> None:
assert ArkteosEntity(FakeClient(True)).available
assert not ArkteosEntity(FakeClient(False)).available
def test_availability_callback_writes_state() -> None:
entity = ArkteosEntity(FakeClient())
entity.async_write_ha_state = Mock()
entity._handle_availability(True)
entity.async_write_ha_state.assert_called_once()
def test_base_entity_has_no_polling_method() -> None:
assert not hasattr(ArkteosEntity(FakeClient()), "async_update")
+2 -1
View File
@@ -116,7 +116,8 @@ async def test_double_unload_is_safe(hass) -> None:
def test_no_polling_platform_is_declared() -> None: def test_no_polling_platform_is_declared() -> None:
from custom_components.arkteos.const import PLATFORMS from custom_components.arkteos.const import PLATFORMS
assert len(PLATFORMS) == 1 assert len(PLATFORMS) == 2
assert {platform.value for platform in PLATFORMS} == {"binary_sensor", "sensor"}
def test_no_mqtt_dependency_in_manifest() -> None: def test_no_mqtt_dependency_in_manifest() -> None:
+259
View File
@@ -0,0 +1,259 @@
"""Tests hors ligne des capteurs Arkteos alimentés par callbacks."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import Mock
import pytest
from custom_components.arkteos.const import DOMAIN
from custom_components.arkteos.parser import filter_values, parse_frame
from custom_components.arkteos.sensor import (
FRIGO_SENSORS,
FRIGO_STATUS_OPTIONS,
PAC_STATUS_OPTIONS,
REGULATION_SENSORS,
SENSOR_DESCRIPTIONS,
ArkteosSensor,
async_setup_entry,
)
FIXTURES = Path(__file__).parent / "fixtures"
class FakeClient:
"""Client en mémoire, sans réseau, pour les capteurs."""
def __init__(self) -> None:
self.available = False
self.latest_frigo_data: dict[str, int | float | str] | None = None
self.latest_regulation_data: dict[str, int | float | str] | None = None
self.data_callbacks: list[object] = []
self.availability_callbacks: list[object] = []
def add_data_callback(self, callback) -> None:
self.data_callbacks.append(callback)
def remove_data_callback(self, callback) -> None:
self.data_callbacks.remove(callback)
def add_availability_callback(self, callback) -> None:
self.availability_callbacks.append(callback)
def remove_availability_callback(self, callback) -> None:
self.availability_callbacks.remove(callback)
def publish(
self,
frame_type: str,
data: dict[str, int | float | str],
rejected: dict[str, int | float] | None = None,
) -> None:
if frame_type == "frigo":
self.latest_frigo_data = dict(data)
elif frame_type == "regulation":
self.latest_regulation_data = dict(data)
availability_changed = not self.available
self.available = True
if availability_changed:
for callback in tuple(self.availability_callbacks):
callback(True)
for callback in tuple(self.data_callbacks):
callback(frame_type, dict(data), rejected or {})
def _frame_values(name: str) -> dict[str, int | float | str]:
filtered = filter_values(parse_frame((FIXTURES / name).read_bytes()))
return filtered.accepted
def _sensors(client: FakeClient) -> list[ArkteosSensor]:
return [ArkteosSensor(client, description) for description in SENSOR_DESCRIPTIONS]
async def _add_sensors_to_hass(sensors: list[ArkteosSensor]) -> None:
"""Exécute le cycle d'ajout qui inscrit les callbacks du client."""
for sensor in sensors:
sensor.async_write_ha_state = Mock()
await sensor.async_added_to_hass()
def _sensor(sensors: list[ArkteosSensor], key: str) -> ArkteosSensor:
return next(sensor for sensor in sensors if sensor.entity_description.key == key)
def test_total_sensor_count_and_unique_ids() -> None:
sensors = _sensors(FakeClient())
assert len(sensors) == 32
assert len({sensor.unique_id for sensor in sensors}) == len(sensors)
def test_all_node_red_keys_are_described() -> None:
assert {description.key for description in FRIGO_SENSORS} == {
"exterieur_temp",
"nb_degivrages",
"temps_compresseur",
"nb_cycles_compresseur",
"freq_comp_actuelle",
"freq_comp_cible",
"fan_speed_evaporator_1",
"dc_voltage",
"statut_frigo_s",
"statut_frigo",
"active_error_fri",
}
assert {description.key for description in REGULATION_SENSORS} == {
"puissance_inst_produite",
"puissance_inst_consommee",
"temps_mise_sous_tension",
"modele_pac_s",
"modele_pac",
"primaire_temp_eau_aller_consigne",
"primaire_temp_eau_aller",
"primaire_temp_eau_retour",
"primaire_debit_eau",
"primaire_pression",
"primaire_circulateur_consigne",
"zone1_temp_interieur",
"zone1_consigne",
"ecs_temp_eau_milieu",
"ecs_temp_eau_bas",
"ecs_consigne",
"nb_cycles_compresseur_reg",
"statut_pac_s",
"statut_pac",
"active_error_reg",
"signal_rf_sonde_1",
}
def test_initial_sensor_state_is_unavailable() -> None:
for sensor in _sensors(FakeClient()):
assert not sensor.available
assert sensor.native_value is None
def test_device_info_is_shared_by_all_sensors() -> None:
infos = [sensor.device_info for sensor in _sensors(FakeClient())]
assert all(info["identifiers"] == {(DOMAIN, "arkteos_zuran4")} for info in infos)
assert all(info["name"] == "PAC Arkteos Zuran 4" for info in infos)
assert all(info["manufacturer"] == "Arkteos" for info in infos)
async def test_frigo_frame_only_makes_frigo_sensors_available() -> None:
client = FakeClient()
sensors = _sensors(client)
await _add_sensors_to_hass(sensors)
client.publish("frigo", _frame_values("frigo_163.bin"))
assert all(sensor.available for sensor in sensors if sensor.entity_description.frame_type == "frigo")
assert not any(sensor.available for sensor in sensors if sensor.entity_description.frame_type == "regulation")
async def test_regulation_frame_keeps_frigo_available_and_enables_regulation() -> None:
client = FakeClient()
sensors = _sensors(client)
await _add_sensors_to_hass(sensors)
client.publish("frigo", _frame_values("frigo_163.bin"))
client.publish("regulation", _frame_values("regulation_227.bin"))
assert all(sensor.available for sensor in sensors)
@pytest.mark.parametrize(
("key", "fixture_name"),
(
("exterieur_temp", "frigo_163.bin"),
("primaire_temp_eau_aller", "regulation_227.bin"),
("primaire_temp_eau_retour", "regulation_227.bin"),
("puissance_inst_consommee", "regulation_227.bin"),
("puissance_inst_produite", "regulation_227.bin"),
("primaire_pression", "regulation_227.bin"),
("ecs_temp_eau_milieu", "regulation_227.bin"),
("ecs_temp_eau_bas", "regulation_227.bin"),
("statut_frigo_s", "frigo_163.bin"),
("statut_pac_s", "regulation_227.bin"),
),
)
async def test_fixture_values_match_parser(key: str, fixture_name: str) -> None:
client = FakeClient()
sensor = _sensor(_sensors(client), key)
await _add_sensors_to_hass([sensor])
values = _frame_values(fixture_name)
client.publish(values["frame_type"], values)
assert sensor.native_value == values[key]
def test_status_enum_options_are_exact() -> None:
descriptions = {description.key: description for description in SENSOR_DESCRIPTIONS}
assert descriptions["statut_frigo_s"].options == FRIGO_STATUS_OPTIONS
assert descriptions["statut_pac_s"].options == PAC_STATUS_OPTIONS
def test_diagnostic_and_disabled_descriptions() -> None:
descriptions = {description.key: description for description in SENSOR_DESCRIPTIONS}
for key in ("statut_frigo", "statut_pac", "modele_pac", "zone1_temp_interieur", "zone1_consigne"):
assert not descriptions[key].entity_registry_enabled_default
for key in ("active_error_fri", "active_error_reg", "signal_rf_sonde_1"):
assert descriptions[key].entity_category.value == "diagnostic"
async def test_rejected_value_does_not_replace_last_valid_value() -> None:
client = FakeClient()
sensor = _sensor(_sensors(client), "exterieur_temp")
await _add_sensors_to_hass([sensor])
valid = _frame_values("frigo_163.bin")
client.publish("frigo", valid)
previous_value = sensor.native_value
client.publish("frigo", {"frame_type": "frigo"}, {"exterieur_temp": 151})
assert sensor.native_value == previous_value
async def test_entity_added_after_frame_uses_latest_data() -> None:
client = FakeClient()
values = _frame_values("frigo_163.bin")
client.publish("frigo", values)
sensor = _sensor(_sensors(client), "exterieur_temp")
await _add_sensors_to_hass([sensor])
assert sensor.available
assert sensor.native_value == values["exterieur_temp"]
def test_callbacks_only_update_their_own_frame_type() -> None:
client = FakeClient()
sensors = _sensors(client)
frigo_sensor = _sensor(sensors, "exterieur_temp")
regulation_sensor = _sensor(sensors, "primaire_temp_eau_aller")
frigo_sensor.async_write_ha_state = Mock()
regulation_sensor.async_write_ha_state = Mock()
client.add_data_callback(frigo_sensor._handle_data)
client.add_data_callback(regulation_sensor._handle_data)
client.publish("frigo", _frame_values("frigo_163.bin"))
frigo_sensor.async_write_ha_state.assert_called_once()
regulation_sensor.async_write_ha_state.assert_not_called()
async def test_sensor_callback_is_removed_when_entity_is_removed() -> None:
client = FakeClient()
sensor = _sensor(_sensors(client), "exterieur_temp")
client.add_data_callback(sensor._handle_data)
client.add_availability_callback(sensor._handle_availability)
await sensor.async_will_remove_from_hass()
assert sensor._handle_data not in client.data_callbacks
assert sensor._handle_availability not in client.availability_callbacks
async def test_sensor_platform_creates_all_descriptions(monkeypatch: pytest.MonkeyPatch) -> None:
import custom_components.arkteos.sensor as sensor_platform
client = FakeClient()
created: list[ArkteosSensor] = []
monkeypatch.setattr(sensor_platform, "_get_client", lambda _hass, _entry: client)
await async_setup_entry(None, object(), lambda entities: created.extend(entities))
assert len(created) == len(SENSOR_DESCRIPTIONS)
def test_sensor_entities_do_not_poll_or_use_mqtt() -> None:
assert all(sensor._attr_should_poll is False for sensor in _sensors(FakeClient()))
assert "mqtt" not in Path(__file__).parents[1].joinpath("custom_components/arkteos/sensor.py").read_text().lower()