Add Compit fan (#164049)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Przemko92
2026-02-25 20:52:01 +01:00
committed by GitHub
parent 80574f7ae0
commit 02972579aa
7 changed files with 516 additions and 0 deletions
@@ -12,6 +12,7 @@ from .coordinator import CompitConfigEntry, CompitDataUpdateCoordinator
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.CLIMATE,
Platform.FAN,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
+172
View File
@@ -0,0 +1,172 @@
"""Fan platform for Compit integration."""
from typing import Any
from compit_inext_api import PARAM_VALUES
from compit_inext_api.consts import CompitParameter
from homeassistant.components.fan import (
FanEntity,
FanEntityDescription,
FanEntityFeature,
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from .const import DOMAIN, MANUFACTURER_NAME
from .coordinator import CompitConfigEntry, CompitDataUpdateCoordinator
PARALLEL_UPDATES = 0
COMPIT_GEAR_TO_HA = PARAM_VALUES[CompitParameter.VENTILATION_GEAR_TARGET]
HA_STATE_TO_COMPIT = {value: key for key, value in COMPIT_GEAR_TO_HA.items()}
DEVICE_DEFINITIONS: dict[int, FanEntityDescription] = {
223: FanEntityDescription(
key="Nano Color 2",
translation_key="ventilation",
),
12: FanEntityDescription(
key="Nano Color",
translation_key="ventilation",
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: CompitConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Compit fan entities from a config entry."""
coordinator = entry.runtime_data
async_add_entities(
CompitFan(
coordinator,
device_id,
device_definition,
)
for device_id, device in coordinator.connector.all_devices.items()
if (device_definition := DEVICE_DEFINITIONS.get(device.definition.code))
)
class CompitFan(CoordinatorEntity[CompitDataUpdateCoordinator], FanEntity):
"""Representation of a Compit fan entity."""
_attr_speed_count = len(COMPIT_GEAR_TO_HA)
_attr_has_entity_name = True
_attr_name = None
_attr_supported_features = (
FanEntityFeature.TURN_ON
| FanEntityFeature.TURN_OFF
| FanEntityFeature.SET_SPEED
)
def __init__(
self,
coordinator: CompitDataUpdateCoordinator,
device_id: int,
entity_description: FanEntityDescription,
) -> None:
"""Initialize the fan entity."""
super().__init__(coordinator)
self.device_id = device_id
self.entity_description = entity_description
self._attr_unique_id = f"{device_id}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, str(device_id))},
name=entity_description.key,
manufacturer=MANUFACTURER_NAME,
model=entity_description.key,
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self.coordinator.connector.get_device(self.device_id) is not None
)
@property
def is_on(self) -> bool | None:
"""Return true if the fan is on."""
value = self.coordinator.connector.get_current_option(
self.device_id, CompitParameter.VENTILATION_ON_OFF
)
return True if value == STATE_ON else False if value == STATE_OFF else None
async def async_turn_on(
self,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn on the fan."""
await self.coordinator.connector.select_device_option(
self.device_id, CompitParameter.VENTILATION_ON_OFF, STATE_ON
)
if percentage is None:
self.async_write_ha_state()
return
await self.async_set_percentage(percentage)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the fan."""
await self.coordinator.connector.select_device_option(
self.device_id, CompitParameter.VENTILATION_ON_OFF, STATE_OFF
)
self.async_write_ha_state()
@property
def percentage(self) -> int | None:
"""Return the current fan speed as a percentage."""
if self.is_on is False:
return 0
mode = self.coordinator.connector.get_current_option(
self.device_id, CompitParameter.VENTILATION_GEAR_TARGET
)
if mode is None:
return None
gear = COMPIT_GEAR_TO_HA.get(mode)
return (
None
if gear is None
else ordered_list_item_to_percentage(
list(COMPIT_GEAR_TO_HA.values()),
gear,
)
)
async def async_set_percentage(self, percentage: int) -> None:
"""Set the fan speed."""
if percentage == 0:
await self.async_turn_off()
return
gear = int(
percentage_to_ordered_list_item(
list(COMPIT_GEAR_TO_HA.values()),
percentage,
)
)
mode = HA_STATE_TO_COMPIT.get(gear)
if mode is None:
return
await self.coordinator.connector.select_device_option(
self.device_id, CompitParameter.VENTILATION_GEAR_TARGET, mode
)
self.async_write_ha_state()
@@ -20,6 +20,14 @@
"default": "mdi:alert"
}
},
"fan": {
"ventilation": {
"default": "mdi:fan",
"state": {
"off": "mdi:fan-off"
}
}
},
"number": {
"boiler_target_temperature": {
"default": "mdi:water-boiler"
@@ -53,6 +53,11 @@
"name": "Temperature alert"
}
},
"fan": {
"ventilation": {
"name": "[%key:component::fan::title%]"
}
},
"number": {
"boiler_target_temperature": {
"name": "Boiler target temperature"
+2
View File
@@ -74,6 +74,8 @@ def mock_connector():
MagicMock(
code="__tempzadpozadomem", value=18.5
), # Target temperature out of home
MagicMock(code="__aerowentylacjaon&off", value="on"),
MagicMock(code="__trybaero2", value="gear_2"),
]
mock_device_2.definition.code = 223 # Nano Color 2
@@ -0,0 +1,57 @@
# serializer version: 1
# name: test_fan_entities_snapshot[fan.nano_color_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'preset_modes': None,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'fan',
'entity_category': None,
'entity_id': 'fan.nano_color_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'compit',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <FanEntityFeature: 49>,
'translation_key': 'ventilation',
'unique_id': '2_Nano Color 2',
'unit_of_measurement': None,
})
# ---
# name: test_fan_entities_snapshot[fan.nano_color_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Nano Color 2',
'percentage': 60,
'percentage_step': 20.0,
'preset_mode': None,
'preset_modes': None,
'supported_features': <FanEntityFeature: 49>,
}),
'context': <ANY>,
'entity_id': 'fan.nano_color_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
+271
View File
@@ -0,0 +1,271 @@
"""Tests for the Compit fan platform."""
from typing import Any
from unittest.mock import MagicMock
from compit_inext_api.consts import CompitParameter
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.fan import (
ATTR_PERCENTAGE,
DOMAIN as FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration, snapshot_compit_entities
from tests.common import MockConfigEntry
async def test_fan_entities_snapshot(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Snapshot test for fan entities creation, unique IDs, and device info."""
await setup_integration(hass, mock_config_entry)
snapshot_compit_entities(hass, entity_registry, snapshot, Platform.FAN)
async def test_fan_turn_on(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test turning on the fan."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_OFF
)
await hass.services.async_call(
FAN_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "fan.nano_color_2"}, blocking=True
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_ON
async def test_fan_turn_on_with_percentage(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test turning on the fan with a percentage."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_OFF
)
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "fan.nano_color_2", ATTR_PERCENTAGE: 100},
blocking=True,
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_ON
assert state.attributes.get("percentage") == 100
async def test_fan_turn_off(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test turning off the fan."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_ON
)
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "fan.nano_color_2"},
blocking=True,
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_OFF
async def test_fan_set_speed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test setting the fan speed."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_ON
) # Ensure fan is on before setting speed
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{
ATTR_ENTITY_ID: "fan.nano_color_2",
ATTR_PERCENTAGE: 80,
},
blocking=True,
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.attributes.get("percentage") == 80
async def test_fan_set_speed_while_off(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test setting the fan speed while the fan is off."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_OFF
)
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{
ATTR_ENTITY_ID: "fan.nano_color_2",
ATTR_PERCENTAGE: 80,
},
blocking=True,
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_OFF # Fan should remain off until turned on
assert state.attributes.get("percentage") == 0
async def test_fan_set_speed_to_not_in_step_percentage(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test setting the fan speed to a percentage that is not in the step of the fan."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_ON
)
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{ATTR_ENTITY_ID: "fan.nano_color_2", ATTR_PERCENTAGE: 65},
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_ON
assert state.attributes.get("percentage") == 80
async def test_fan_set_speed_to_0(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
) -> None:
"""Test setting the fan speed to 0."""
await setup_integration(hass, mock_config_entry)
await mock_connector.select_device_option(
2, CompitParameter.VENTILATION_ON_OFF, STATE_ON
) # Turn on fan first
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{
ATTR_ENTITY_ID: "fan.nano_color_2",
ATTR_PERCENTAGE: 0,
},
blocking=True,
)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_OFF # Fan is turned off by setting the percentage to 0
assert state.attributes.get("percentage") == 0
@pytest.mark.parametrize(
"mock_return_value",
[
None,
"invalid",
],
)
async def test_fan_invalid_speed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
mock_return_value: Any,
) -> None:
"""Test setting an invalid speed."""
mock_connector.get_current_option.side_effect = lambda device_id, parameter_code: (
mock_return_value
)
await setup_integration(hass, mock_config_entry)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.state == STATE_UNKNOWN
@pytest.mark.parametrize(
("gear", "expected_percentage"),
[
("gear_0", 20),
("gear_1", 40),
("gear_2", 60),
("gear_3", 80),
("airing", 100),
],
)
async def test_fan_gear_to_percentage(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connector: MagicMock,
gear: str,
expected_percentage: int,
) -> None:
"""Test the gear to percentage conversion."""
mock_connector.get_current_option.side_effect = lambda device_id, parameter_code: (
gear
)
await setup_integration(hass, mock_config_entry)
state = hass.states.get("fan.nano_color_2")
assert state is not None
assert state.attributes.get("percentage") == expected_percentage