starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add binary_sensor platform to IntelliClima (#178154)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
dvdinth
2026-08-18 16:56:49 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent dd58d215e1
commit e2990016e4
13 changed files with 479 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/intelliclima/__init__.py
+18 -7
View File
@@ -7,9 +7,14 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import LOGGER
from .coordinator import IntelliClimaConfigEntry, IntelliClimaCoordinator
from .coordinator import (
IntelliClimaConfigEntry,
IntelliClimaCoordinator,
IntelliClimaData,
IntelliClimaFilterCoordinator,
)
PLATFORMS = [Platform.FAN, Platform.SELECT, Platform.SENSOR]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.FAN, Platform.SELECT, Platform.SENSOR]
async def async_setup_entry(
@@ -25,18 +30,24 @@ async def async_setup_entry(
)
# Create coordinator
coordinator = IntelliClimaCoordinator(hass, entry, api)
devices_coordinator = IntelliClimaCoordinator(hass, entry, api)
# Fetch initial data
await coordinator.async_config_entry_first_refresh()
await devices_coordinator.async_config_entry_first_refresh()
LOGGER.debug(
"Discovered %d IntelliClima VMC device(s)",
len(coordinator.data.ecocomfort2_devices),
len(devices_coordinator.data.ecocomfort2_devices),
)
# Store coordinator
entry.runtime_data = coordinator
device_serials = [
device.crono_sn
for device in devices_coordinator.data.ecocomfort2_devices.values()
]
filter_coordinator = IntelliClimaFilterCoordinator(hass, entry, api, device_serials)
await filter_coordinator.async_refresh()
entry.runtime_data = IntelliClimaData(devices_coordinator, filter_coordinator)
# Set up platforms
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
homeassistant/components/intelliclima/binary_sensor.py
+75
View File
@@ -0,0 +1,75 @@
"""Support for IntelliClima Binary Sensors."""
from typing import override
from pyintelliclima.intelliclima_types import IntelliClimaECO
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import IntelliClimaConfigEntry, IntelliClimaFilterCoordinator
from .entity import eco_device_info
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: IntelliClimaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the IntelliClima binary sensor platform."""
data = entry.runtime_data
async_add_entities(
IntelliClimaFilterCleaningBinarySensor(
coordinator=data.filter_coordinator, device=ecocomfort2
)
for ecocomfort2 in data.devices_coordinator.data.ecocomfort2_devices.values()
)
class IntelliClimaFilterCleaningBinarySensor(
CoordinatorEntity[IntelliClimaFilterCoordinator], BinarySensorEntity
):
"""Binary sensor indicating whether the device's filter needs cleaning."""
_attr_has_entity_name = True
_attr_translation_key = "filter_cleaning"
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_device_class = BinarySensorDeviceClass.PROBLEM
def __init__(
self,
coordinator: IntelliClimaFilterCoordinator,
device: IntelliClimaECO,
) -> None:
"""Class initializer."""
super().__init__(coordinator)
self._attr_device_info = eco_device_info(device)
self._device_sn = device.crono_sn
self._attr_unique_id = f"{device.id}_filter_cleaning"
@property
@override
def available(self) -> bool:
"""Return if entity is available."""
device_data = (self.coordinator.data or {}).get(self._device_sn)
return super().available and device_data is not None and device_data.is_active
@property
@override
def is_on(self) -> bool | None:
"""Return true if the filter needs cleaning."""
device_data = (self.coordinator.data or {}).get(self._device_sn)
if device_data is None or not device_data.is_active:
return None
return device_data.change_filter
homeassistant/components/intelliclima/const.py
+3
View File
@@ -9,3 +9,6 @@ DOMAIN = "intelliclima"
# Update interval
DEFAULT_SCAN_INTERVAL = timedelta(minutes=1)
# Filter status is expensive to compute cloud-side, so it's polled far less often.
FILTER_SCAN_INTERVAL = timedelta(days=1)
homeassistant/components/intelliclima/coordinator.py
+58 -2
View File
@@ -1,16 +1,19 @@
"""DataUpdateCoordinator for IntelliClima."""
import asyncio
from dataclasses import dataclass
from typing import override
from pyintelliclima import IntelliClimaAPI, IntelliClimaAPIError, IntelliClimaDevices
from pyintelliclima.intelliclima_types import IntelliClimaFilterStatus
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, FILTER_SCAN_INTERVAL, LOGGER
type IntelliClimaConfigEntry = ConfigEntry[IntelliClimaCoordinator]
type IntelliClimaConfigEntry = ConfigEntry[IntelliClimaData]
class IntelliClimaCoordinator(DataUpdateCoordinator[IntelliClimaDevices]):
@@ -47,3 +50,56 @@ class IntelliClimaCoordinator(DataUpdateCoordinator[IntelliClimaDevices]):
except IntelliClimaAPIError as err:
raise UpdateFailed(f"Failed to update data: {err}") from err
class IntelliClimaFilterCoordinator(
DataUpdateCoordinator[dict[str, IntelliClimaFilterStatus]]
):
"""Coordinator to manage fetching IntelliClima filter status, polled once a day."""
def __init__(
self,
hass: HomeAssistant,
entry: IntelliClimaConfigEntry,
api: IntelliClimaAPI,
device_serials: list[str],
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
LOGGER,
name=f"{DOMAIN}_filter",
update_interval=FILTER_SCAN_INTERVAL,
config_entry=entry,
)
self.api = api
self._device_serials = device_serials
@override
async def _async_update_data(self) -> dict[str, IntelliClimaFilterStatus]:
"""Fetch filter status for all devices, isolating per-device failures."""
results = await asyncio.gather(
*(self.api.get_filter_status(serial) for serial in self._device_serials),
return_exceptions=True,
)
statuses: dict[str, IntelliClimaFilterStatus] = {}
for serial, result in zip(self._device_serials, results, strict=True):
if isinstance(result, IntelliClimaAPIError):
LOGGER.warning(
"Failed to update filter status for %s: %s", serial, result
)
continue
if isinstance(result, BaseException):
raise result
statuses[serial] = result
return statuses
@dataclass
class IntelliClimaData:
"""Runtime data for the IntelliClima config entry."""
devices_coordinator: IntelliClimaCoordinator
filter_coordinator: IntelliClimaFilterCoordinator
homeassistant/components/intelliclima/entity.py
+17 -7
View File
@@ -4,7 +4,6 @@ from typing import override
from pyintelliclima.intelliclima_types import IntelliClimaC800, IntelliClimaECO
from homeassistant.const import ATTR_CONNECTIONS, ATTR_MODEL, ATTR_SW_VERSION
from homeassistant.helpers.device_registry import (
CONNECTION_BLUETOOTH,
CONNECTION_NETWORK_MAC,
@@ -16,6 +15,22 @@ from .const import DOMAIN
from .coordinator import IntelliClimaCoordinator
def eco_device_info(device: IntelliClimaECO) -> DeviceInfo:
"""Return the device info shared by all entities of an ECOCOMFORT 2.0."""
return DeviceInfo(
identifiers={(DOMAIN, device.id)},
manufacturer="Fantini Cosmi",
name=device.name,
serial_number=device.crono_sn,
model="ECOCOMFORT 2.0",
sw_version=device.fw,
connections={
(CONNECTION_BLUETOOTH, device.mac),
(CONNECTION_NETWORK_MAC, device.macwifi),
},
)
class IntelliClimaEntity(CoordinatorEntity[IntelliClimaCoordinator]):
"""Define a generic class for IntelliClima entities."""
@@ -53,12 +68,7 @@ class IntelliClimaECOEntity(IntelliClimaEntity):
"""Class initializer."""
super().__init__(coordinator, device)
self._attr_device_info[ATTR_MODEL] = "ECOCOMFORT 2.0"
self._attr_device_info[ATTR_SW_VERSION] = device.fw
self._attr_device_info[ATTR_CONNECTIONS] = {
(CONNECTION_BLUETOOTH, device.mac),
(CONNECTION_NETWORK_MAC, device.macwifi),
}
self._attr_device_info = eco_device_info(device)
@property
def _device_data(self) -> IntelliClimaECO:
homeassistant/components/intelliclima/fan.py
+1 -1
View File
@@ -28,7 +28,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up IntelliClima VMC fans."""
coordinator = entry.runtime_data
coordinator = entry.runtime_data.devices_coordinator
entities: list[IntelliClimaVMCFan] = [
IntelliClimaVMCFan(
homeassistant/components/intelliclima/select.py
+1 -1
View File
@@ -31,7 +31,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up IntelliClima VMC fan mode select."""
coordinator = entry.runtime_data
coordinator = entry.runtime_data.devices_coordinator
entities: list[IntelliClimaVMCFanModeSelect] = [
IntelliClimaVMCFanModeSelect(
homeassistant/components/intelliclima/sensor.py
+1 -1
View File
@@ -61,7 +61,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up a IntelliClima Sensors."""
coordinator = entry.runtime_data
coordinator = entry.runtime_data.devices_coordinator
entities: list[IntelliClimaSensor] = [
IntelliClimaSensor(
homeassistant/components/intelliclima/strings.json
+5
View File
@@ -24,6 +24,11 @@
}
},
"entity": {
"binary_sensor": {
"filter_cleaning": {
"name": "Filter cleaning required"
}
},
"select": {
"fan_mode": {
"name": "Fan direction mode",
tests/components/intelliclima/conftest.py
+44 -8
View File
@@ -8,6 +8,8 @@ from pyintelliclima.const import FanMode, FanSpeed
from pyintelliclima.intelliclima_types import (
IntelliClimaDevices,
IntelliClimaECO,
IntelliClimaFilterStatsEntry,
IntelliClimaFilterStatus,
IntelliClimaModelType,
)
import pytest
@@ -39,17 +41,16 @@ def mock_config_entry() -> MockConfigEntry:
)
@pytest.fixture
def single_eco_device() -> IntelliClimaDevices:
"""Create IntelliClimaDevices with one ECOCOMFORT 2.0 and no C800."""
eco = IntelliClimaECO(
id="56789",
crono_sn="11223344",
def create_eco_device(device_id: str, crono_sn: str, name: str) -> IntelliClimaECO:
"""Create an ECOCOMFORT 2.0 device."""
return IntelliClimaECO(
id=device_id,
crono_sn=crono_sn,
status="OK",
online="OK",
command="OK",
model=IntelliClimaModelType(modello="ECO", tipo="wifi"),
name="Test VMC",
name=name,
houses_id="12345",
mode_set=FanMode.inward,
mode_state="1",
@@ -109,11 +110,30 @@ def single_eco_device() -> IntelliClimaDevices:
online_status_debug="mock",
)
@pytest.fixture
def single_eco_device() -> IntelliClimaDevices:
"""Create IntelliClimaDevices with one ECOCOMFORT 2.0 and no C800."""
eco = create_eco_device("56789", "11223344", "Test VMC")
return IntelliClimaDevices(ecocomfort2_devices={eco.id: eco}, c800_devices={})
@pytest.fixture
def mock_cloud_interface(single_eco_device) -> Generator[AsyncMock]:
def two_eco_devices() -> IntelliClimaDevices:
"""Create IntelliClimaDevices with two ECOCOMFORT 2.0 devices and no C800."""
first = create_eco_device("56789", "11223344", "Test VMC")
second = create_eco_device("98765", "55667788", "Other VMC")
return IntelliClimaDevices(
ecocomfort2_devices={first.id: first, second.id: second}, c800_devices={}
)
@pytest.fixture
def mock_cloud_interface(
single_eco_device: IntelliClimaDevices,
) -> Generator[AsyncMock]:
"""Mock IntelliClimaAPI for tests."""
with (
@@ -132,6 +152,22 @@ def mock_cloud_interface(single_eco_device) -> Generator[AsyncMock]:
# Mock other async methods if needed
mock_client.authenticate.return_value = True
mock_client.get_all_device_status.return_value = single_eco_device
mock_client.get_filter_status.return_value = IntelliClimaFilterStatus(
serial="11223344",
is_active=True,
from_date="2025-11-18 10:22:51",
stats=[
IntelliClimaFilterStatsEntry(
night_tot_hour="10",
low_tot_hour="20",
medium_tot_hour="30",
high_tot_hour="5",
boost_tot_hour="1",
)
],
totale=66.0,
change_filter=True,
)
# Sub-API used by the fan entity
mock_client.ecocomfort = SimpleNamespace(
tests/components/intelliclima/snapshots/test_binary_sensor.ambr
+90
View File
@@ -0,0 +1,90 @@
# serializer version: 1
# name: test_all_binary_sensor_entities.2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'bluetooth',
'00:11:22:33:44:55',
),
tuple(
'mac',
'00:11:22:33:44:55',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'intelliclima',
'56789',
),
}),
'labels': set({
}),
'manufacturer': 'Fantini Cosmi',
'model': 'ECOCOMFORT 2.0',
'model_id': None,
'name': 'Test VMC',
'name_by_user': None,
'serial_number': '11223344',
'sw_version': '0.6.8',
'via_device_id': None,
})
# ---
# name: test_all_binary_sensor_entities[binary_sensor.test_vmc_filter_cleaning_required-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.test_vmc_filter_cleaning_required',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Filter cleaning required',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'Filter cleaning required',
'platform': 'intelliclima',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'filter_cleaning',
'unique_id': '56789_filter_cleaning',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensor_entities[binary_sensor.test_vmc_filter_cleaning_required-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'problem',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test VMC Filter cleaning required',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_vmc_filter_cleaning_required',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/intelliclima/test_binary_sensor.py
+84
View File
@@ -0,0 +1,84 @@
"""Test IntelliClima Binary Sensors."""
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, patch
from pyintelliclima.intelliclima_types import IntelliClimaFilterStatus
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
async def setup_intelliclima_binary_sensor_only(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_cloud_interface: AsyncMock,
) -> AsyncGenerator[None]:
"""Set up IntelliClima integration with only the binary sensor platform."""
with (
patch(
"homeassistant.components.intelliclima.PLATFORMS", [Platform.BINARY_SENSOR]
),
):
await setup_integration(hass, mock_config_entry)
yield
async def test_all_binary_sensor_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_cloud_interface: AsyncMock,
) -> None:
"""Test all entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
binary_sensor_entries = [
entry
for entry in entity_registry.entities.values()
if entry.platform == "intelliclima" and entry.domain == BINARY_SENSOR_DOMAIN
]
assert len(binary_sensor_entries) == 1
for entity_entry in binary_sensor_entries:
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert device_entry == snapshot
async def test_filter_cleaning_unavailable_when_tracking_disabled(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_cloud_interface: AsyncMock,
) -> None:
"""Test the filter_cleaning sensor is unavailable when the vendor disables filter tracking.
The vendor API keeps returning `change_filter: false` in this state, which
would otherwise misreport a "clean filter" the integration can't actually vouch for.
"""
mock_cloud_interface.get_filter_status.return_value = IntelliClimaFilterStatus(
serial="11223344",
is_active=False,
from_date="2025-11-18 10:22:51",
stats=[],
totale=0,
change_filter=False,
)
await hass.config_entries.async_reload(mock_config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_vmc_filter_cleaning_required")
assert state is not None
assert state.state == STATE_UNAVAILABLE
tests/components/intelliclima/test_init.py
+82
View File
@@ -0,0 +1,82 @@
"""Test the IntelliClima integration setup."""
from unittest.mock import AsyncMock
from pyintelliclima.api import IntelliClimaAPIError
from pyintelliclima.intelliclima_types import (
IntelliClimaDevices,
IntelliClimaFilterStatus,
)
from homeassistant.components.intelliclima.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry
async def test_setup_succeeds_when_filter_status_unavailable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_cloud_interface: AsyncMock,
) -> None:
"""Test the config entry still loads when the filter-status endpoint fails.
Filter status only backs a diagnostic binary sensor, so a transient
failure of that ancillary endpoint must not block the fan, select, and
sensor platforms from being set up.
"""
mock_cloud_interface.get_filter_status.side_effect = IntelliClimaAPIError(
"cannot compute filter status"
)
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert hass.states.get("fan.test_vmc") is not None
state = hass.states.get("binary_sensor.test_vmc_filter_cleaning_required")
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_filter_status_failure_isolated_per_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_cloud_interface: AsyncMock,
entity_registry: er.EntityRegistry,
two_eco_devices: IntelliClimaDevices,
) -> None:
"""Test a filter-status failure for one device does not affect the others."""
mock_cloud_interface.get_all_device_status.return_value = two_eco_devices
working_device, failing_device = two_eco_devices.ecocomfort2_devices.values()
filter_status = mock_cloud_interface.get_filter_status.return_value
def _get_filter_status(serial: str) -> IntelliClimaFilterStatus:
if serial == failing_device.crono_sn:
raise IntelliClimaAPIError("cannot compute filter status")
return filter_status
mock_cloud_interface.get_filter_status.side_effect = _get_filter_status
await setup_integration(hass, mock_config_entry)
working_entity_id = entity_registry.async_get_entity_id(
Platform.BINARY_SENSOR, DOMAIN, f"{working_device.id}_filter_cleaning"
)
assert working_entity_id is not None
state = hass.states.get(working_entity_id)
assert state is not None
assert state.state == STATE_ON
failing_entity_id = entity_registry.async_get_entity_id(
Platform.BINARY_SENSOR, DOMAIN, f"{failing_device.id}_filter_cleaning"
)
assert failing_entity_id is not None
state = hass.states.get(failing_entity_id)
assert state is not None
assert state.state == STATE_UNAVAILABLE