Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 05:06:13 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add support for Comelit Vedo system connected via Comelit Serial bridge (#156301)

Browse Source
This commit is contained in:
Simone Chemelli
2025-12-29 16:59:52 +01:00
committed by GitHub
parent d90e72c6d4
commit 8ca87ef1cb
20 changed files with 366 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/components/comelit/__init__.py
View File

@@ -5,7 +5,7 @@ from aiocomelit.const import BRIDGE
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE, Platform
from homeassistant.core import HomeAssistant
from .const import DEFAULT_PORT
from .const import CONF_VEDO_PIN, DEFAULT_PORT
from .coordinator import (
ComelitBaseCoordinator,
ComelitConfigEntry,
@@ -22,6 +22,16 @@ BRIDGE_PLATFORMS = [
Platform.SENSOR,
Platform.SWITCH,
]
BRIDGE_AND_VEDO_PLATFORMS = [
Platform.ALARM_CONTROL_PANEL,
Platform.BINARY_SENSOR,
Platform.CLIMATE,
Platform.COVER,
Platform.HUMIDIFIER,
Platform.LIGHT,
Platform.SENSOR,
Platform.SWITCH,
]
VEDO_PLATFORMS = [
Platform.ALARM_CONTROL_PANEL,
Platform.BINARY_SENSOR,
@@ -37,15 +47,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: ComelitConfigEntry) -> b
session = await async_client_session(hass)
if entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
vedo_pin = entry.data.get(CONF_VEDO_PIN)
coordinator = ComelitSerialBridge(
hass,
entry,
entry.data[CONF_HOST],
entry.data.get(CONF_PORT, DEFAULT_PORT),
entry.data[CONF_PIN],
vedo_pin,
session,
)
platforms = BRIDGE_PLATFORMS
# Add VEDO platforms if vedo_pin is configured
if vedo_pin:
platforms = BRIDGE_AND_VEDO_PLATFORMS
else:
coordinator = ComelitVedoSystem(
hass,
@@ -71,6 +86,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ComelitConfigEntry) ->
if entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
platforms = BRIDGE_PLATFORMS
# Add VEDO platforms if vedo_pin was configured
if entry.data.get(CONF_VEDO_PIN):
platforms = BRIDGE_AND_VEDO_PLATFORMS
else:
platforms = VEDO_PLATFORMS

34
homeassistant/components/comelit/alarm_control_panel.py
View File

@@ -3,10 +3,10 @@
from __future__ import annotations
import logging
from typing import cast
from typing import TYPE_CHECKING, cast
from aiocomelit.api import ComelitVedoAreaObject
from aiocomelit.const import AlarmAreaState
from aiocomelit.const import ALARM_AREA, AlarmAreaState
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
@@ -18,7 +18,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import ComelitConfigEntry, ComelitVedoSystem
from .coordinator import ComelitConfigEntry, ComelitSerialBridge, ComelitVedoSystem
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@@ -56,15 +56,25 @@ async def async_setup_entry(
) -> None:
"""Set up the Comelit VEDO system alarm control panel devices."""
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
coordinator = config_entry.runtime_data
is_bridge = isinstance(coordinator, ComelitSerialBridge)
async_add_entities(
ComelitAlarmEntity(coordinator, device, config_entry.entry_id)
for device in coordinator.data["alarm_areas"].values()
)
if TYPE_CHECKING:
if is_bridge:
assert isinstance(coordinator, ComelitSerialBridge)
else:
assert isinstance(coordinator, ComelitVedoSystem)
if data := coordinator.data[ALARM_AREA]:
async_add_entities(
ComelitAlarmEntity(coordinator, device, config_entry.entry_id)
for device in data.values()
)
class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanelEntity):
class ComelitAlarmEntity(
CoordinatorEntity[ComelitVedoSystem | ComelitSerialBridge], AlarmControlPanelEntity
):
"""Representation of a Ness alarm panel."""
_attr_has_entity_name = True
@@ -78,7 +88,7 @@ class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanel
def __init__(
self,
coordinator: ComelitVedoSystem,
coordinator: ComelitVedoSystem | ComelitSerialBridge,
area: ComelitVedoAreaObject,
config_entry_entry_id: str,
) -> None:
@@ -95,7 +105,9 @@ class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanel
@property
def _area(self) -> ComelitVedoAreaObject:
"""Return area object."""
return self.coordinator.data["alarm_areas"][self._area_index]
return cast(
ComelitVedoAreaObject, self.coordinator.data[ALARM_AREA][self._area_index]
)
@property
def available(self) -> bool:

46
homeassistant/components/comelit/binary_sensor.py
View File

@@ -2,9 +2,10 @@
from __future__ import annotations
from typing import cast
from typing import TYPE_CHECKING, cast
from aiocomelit import ComelitVedoZoneObject
from aiocomelit.api import ComelitVedoZoneObject
from aiocomelit.const import ALARM_ZONE, AlarmZoneState
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@@ -15,7 +16,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ObjectClassType
from .coordinator import ComelitConfigEntry, ComelitVedoSystem
from .coordinator import ComelitConfigEntry, ComelitSerialBridge, ComelitVedoSystem
from .utils import new_device_listener
# Coordinator is used to centralize the data updates
@@ -29,25 +30,32 @@ async def async_setup_entry(
) -> None:
"""Set up Comelit VEDO presence sensors."""
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
coordinator = config_entry.runtime_data
is_bridge = isinstance(coordinator, ComelitSerialBridge)
if TYPE_CHECKING:
if is_bridge:
assert isinstance(coordinator, ComelitSerialBridge)
else:
assert isinstance(coordinator, ComelitVedoSystem)
def _add_new_entities(new_devices: list[ObjectClassType], dev_type: str) -> None:
"""Add entities for new monitors."""
entities = [
ComelitVedoBinarySensorEntity(coordinator, device, config_entry.entry_id)
for device in coordinator.data["alarm_zones"].values()
for device in coordinator.data[dev_type].values()
if device in new_devices
]
if entities:
async_add_entities(entities)
config_entry.async_on_unload(
new_device_listener(coordinator, _add_new_entities, "alarm_zones")
new_device_listener(coordinator, _add_new_entities, ALARM_ZONE)
)
class ComelitVedoBinarySensorEntity(
CoordinatorEntity[ComelitVedoSystem], BinarySensorEntity
CoordinatorEntity[ComelitVedoSystem | ComelitSerialBridge], BinarySensorEntity
):
"""Sensor device."""
@@ -56,7 +64,7 @@ class ComelitVedoBinarySensorEntity(
def __init__(
self,
coordinator: ComelitVedoSystem,
coordinator: ComelitVedoSystem | ComelitSerialBridge,
zone: ComelitVedoZoneObject,
config_entry_entry_id: str,
) -> None:
@@ -68,9 +76,25 @@ class ComelitVedoBinarySensorEntity(
self._attr_unique_id = f"{config_entry_entry_id}-presence-{zone.index}"
self._attr_device_info = coordinator.platform_device_info(zone, "zone")
@property
def _zone(self) -> ComelitVedoZoneObject:
"""Return zone object."""
return cast(
ComelitVedoZoneObject, self.coordinator.data[ALARM_ZONE][self._zone_index]
)
@property
def available(self) -> bool:
"""Return True if alarm is available."""
if self._zone.human_status in [
AlarmZoneState.FAULTY,
AlarmZoneState.UNAVAILABLE,
AlarmZoneState.UNKNOWN,
]:
return False
return super().available
@property
def is_on(self) -> bool:
"""Presence detected."""
return (
self.coordinator.data["alarm_zones"][self._zone_index].status_api == "0001"
)
return self._zone.status_api == "0001"

42
homeassistant/components/comelit/config_flow.py
View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from asyncio.exceptions import TimeoutError
from collections.abc import Mapping
import re
from typing import Any
from typing import TYPE_CHECKING, Any
from aiocomelit import (
ComeliteSerialBridgeApi,
@@ -22,7 +22,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from .const import _LOGGER, DEFAULT_PORT, DEVICE_TYPE_LIST, DOMAIN
from .const import _LOGGER, CONF_VEDO_PIN, DEFAULT_PORT, DEVICE_TYPE_LIST, DOMAIN
from .utils import async_client_session
DEFAULT_HOST = "192.168.1.252"
@@ -34,9 +34,12 @@ USER_SCHEMA = vol.Schema(
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_PIN, default=DEFAULT_PIN): cv.string,
vol.Required(CONF_TYPE, default=BRIDGE): vol.In(DEVICE_TYPE_LIST),
vol.Optional(CONF_VEDO_PIN): cv.string,
}
)
STEP_REAUTH_DATA_SCHEMA = vol.Schema({vol.Required(CONF_PIN): cv.string})
STEP_REAUTH_DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_PIN): cv.string, vol.Optional(CONF_VEDO_PIN): cv.string}
)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, str]:
@@ -72,6 +75,18 @@ async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str,
finally:
await api.logout()
# Validate VEDO PIN if provided and device type is BRIDGE
if data.get(CONF_VEDO_PIN) and data.get(CONF_TYPE, BRIDGE) == BRIDGE:
if not re.fullmatch(r"[0-9]{4,10}", data[CONF_VEDO_PIN]):
raise InvalidVedoPin
if TYPE_CHECKING:
assert isinstance(api, ComeliteSerialBridgeApi)
# Verify VEDO is enabled with the provided PIN
if not await api.vedo_enabled(data[CONF_VEDO_PIN]):
raise InvalidVedoAuth
return {"title": data[CONF_HOST]}
@@ -99,6 +114,10 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
errors["base"] = "invalid_auth"
except InvalidPin:
errors["base"] = "invalid_pin"
except InvalidVedoPin:
errors["base"] = "invalid_vedo_pin"
except InvalidVedoAuth:
errors["base"] = "invalid_vedo_auth"
except Exception: # noqa: BLE001
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
@@ -182,6 +201,8 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_PIN: user_input[CONF_PIN],
CONF_TYPE: reconfigure_entry.data.get(CONF_TYPE, BRIDGE),
}
if CONF_VEDO_PIN in user_input:
data_to_validate[CONF_VEDO_PIN] = user_input[CONF_VEDO_PIN]
await validate_input(self.hass, data_to_validate)
except CannotConnect:
errors["base"] = "cannot_connect"
@@ -189,6 +210,10 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
errors["base"] = "invalid_auth"
except InvalidPin:
errors["base"] = "invalid_pin"
except InvalidVedoPin:
errors["base"] = "invalid_vedo_pin"
except InvalidVedoAuth:
errors["base"] = "invalid_vedo_auth"
except Exception: # noqa: BLE001
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
@@ -198,6 +223,8 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_PORT: user_input[CONF_PORT],
CONF_PIN: user_input[CONF_PIN],
}
if CONF_VEDO_PIN in user_input:
data_updates[CONF_VEDO_PIN] = user_input[CONF_VEDO_PIN]
return self.async_update_reload_and_abort(
reconfigure_entry, data_updates=data_updates
)
@@ -211,6 +238,7 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_PORT, default=reconfigure_entry.data[CONF_PORT]
): cv.port,
vol.Optional(CONF_PIN): cv.string,
vol.Optional(CONF_VEDO_PIN): cv.string,
}
)
@@ -231,3 +259,11 @@ class InvalidAuth(HomeAssistantError):
class InvalidPin(HomeAssistantError):
"""Error to indicate an invalid pin."""
class InvalidVedoPin(HomeAssistantError):
"""Error to indicate an invalid VEDO pin."""
class InvalidVedoAuth(HomeAssistantError):
"""Error to indicate VEDO authentication failed."""

1
homeassistant/components/comelit/const.py
View File

@@ -19,6 +19,7 @@ ObjectClassType = (
DOMAIN = "comelit"
DEFAULT_PORT = 80
DEVICE_TYPE_LIST = [BRIDGE, VEDO]
CONF_VEDO_PIN = "vedo_pin"
SCAN_INTERVAL = 5

53
homeassistant/components/comelit/coordinator.py
View File

@@ -1,17 +1,14 @@
"""Support for Comelit."""
from abc import abstractmethod
from collections.abc import Mapping
from datetime import timedelta
from typing import Any, TypeVar
from typing import TypeVar, cast
from aiocomelit.api import (
AlarmDataObject,
ComelitCommonApi,
ComeliteSerialBridgeApi,
ComelitSerialBridgeObject,
ComelitVedoApi,
)
from aiocomelit.api import ComelitCommonApi, ComeliteSerialBridgeApi, ComelitVedoApi
from aiocomelit.const import (
ALARM_AREA,
ALARM_ZONE,
BRIDGE,
CLIMATE,
COVER,
@@ -37,7 +34,10 @@ type ComelitConfigEntry = ConfigEntry[ComelitBaseCoordinator]
T = TypeVar(
"T",
bound=dict[str, dict[int, ComelitSerialBridgeObject]] | AlarmDataObject,
bound=dict[
str,
Mapping[int, ObjectClassType],
],
)
@@ -118,8 +118,8 @@ class ComelitBaseCoordinator(DataUpdateCoordinator[T]):
async def _async_remove_stale_devices(
self,
previous_list: dict[int, Any],
current_list: dict[int, Any],
previous_list: Mapping[int, ObjectClassType],
current_list: Mapping[int, ObjectClassType],
dev_type: str,
) -> None:
"""Remove stale devices."""
@@ -143,9 +143,7 @@ class ComelitBaseCoordinator(DataUpdateCoordinator[T]):
)
class ComelitSerialBridge(
ComelitBaseCoordinator[dict[str, dict[int, ComelitSerialBridgeObject]]]
):
class ComelitSerialBridge(ComelitBaseCoordinator[T]):
"""Queries Comelit Serial Bridge."""
_hw_version = "20003101"
@@ -158,17 +156,23 @@ class ComelitSerialBridge(
host: str,
port: int,
pin: str,
vedo_pin: str | None,
session: ClientSession,
) -> None:
"""Initialize the scanner."""
self.api = ComeliteSerialBridgeApi(host, port, pin, session)
self.vedo_pin = vedo_pin
super().__init__(hass, entry, BRIDGE, host)
async def _async_update_system_data(
self,
) -> dict[str, dict[int, ComelitSerialBridgeObject]]:
) -> T:
"""Specific method for updating data."""
data = await self.api.get_all_devices()
data: dict[
str,
Mapping[int, ObjectClassType],
] = {}
data.update(await self.api.get_all_devices())
if self.data:
for dev_type in (CLIMATE, COVER, LIGHT, IRRIGATION, OTHER, SCENARIO):
@@ -176,10 +180,14 @@ class ComelitSerialBridge(
self.data[dev_type], data[dev_type], dev_type
)
return data
# Get VEDO alarm data if vedo_pin is configured
if self.vedo_pin:
data.update(await self.api.get_all_areas_and_zones())
return cast(T, data)
class ComelitVedoSystem(ComelitBaseCoordinator[AlarmDataObject]):
class ComelitVedoSystem(ComelitBaseCoordinator[T]):
"""Queries Comelit VEDO system."""
_hw_version = "VEDO IP"
@@ -196,20 +204,21 @@ class ComelitVedoSystem(ComelitBaseCoordinator[AlarmDataObject]):
) -> None:
"""Initialize the scanner."""
self.api = ComelitVedoApi(host, port, pin, session)
self.vedo_pin = pin
super().__init__(hass, entry, VEDO, host)
async def _async_update_system_data(
self,
) -> AlarmDataObject:
) -> T:
"""Specific method for updating data."""
data = await self.api.get_all_areas_and_zones()
if self.data:
for obj_type in ("alarm_areas", "alarm_zones"):
for obj_type in (ALARM_AREA, ALARM_ZONE):
await self._async_remove_stale_devices(
self.data[obj_type],
data[obj_type],
"area" if obj_type == "alarm_areas" else "zone",
"area" if obj_type == ALARM_AREA else "zone",
)
return data
return cast(T, data)

4
homeassistant/components/comelit/cover.py
View File

@@ -72,7 +72,7 @@ class ComelitCoverEntity(ComelitBridgeBaseEntity, RestoreEntity, CoverEntity):
@property
def device_status(self) -> int:
"""Return current device status."""
return self.coordinator.data[COVER][self._device.index].status
return cast("int", self.coordinator.data[COVER][self._device.index].status)
@property
def is_closed(self) -> bool | None:
@@ -86,7 +86,7 @@ class ComelitCoverEntity(ComelitBridgeBaseEntity, RestoreEntity, CoverEntity):
@property
def is_closing(self) -> bool:
"""Return if the cover is closing."""
return self._current_action("closing")
return bool(self._current_action("closing"))
@property
def is_opening(self) -> bool:

2
homeassistant/components/comelit/light.py
View File

@@ -68,4 +68,4 @@ class ComelitLightEntity(ComelitBridgeBaseEntity, LightEntity):
@property
def is_on(self) -> bool:
"""Return True if light is on."""
return self.coordinator.data[LIGHT][self._device.index].status == STATE_ON
return bool(self.coordinator.data[LIGHT][self._device.index].status == STATE_ON)

2
homeassistant/components/comelit/manifest.json
View File

@@ -8,5 +8,5 @@
"iot_class": "local_polling",
"loggers": ["aiocomelit"],
"quality_scale": "platinum",
"requirements": ["aiocomelit==1.1.2"]
"requirements": ["aiocomelit==2.0.0"]
}

75
homeassistant/components/comelit/sensor.py
View File

@@ -2,17 +2,17 @@
from __future__ import annotations
from typing import Final, cast
from typing import TYPE_CHECKING, Final, cast
from aiocomelit.api import ComelitSerialBridgeObject, ComelitVedoZoneObject
from aiocomelit.const import BRIDGE, OTHER, AlarmZoneState
from aiocomelit.const import ALARM_ZONE, OTHER, AlarmZoneState
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import CONF_TYPE, UnitOfPower
from homeassistant.const import UnitOfPower
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
@@ -52,23 +52,20 @@ async def async_setup_entry(
) -> None:
"""Set up Comelit sensors."""
if config_entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
await async_setup_bridge_entry(hass, config_entry, async_add_entities)
else:
await async_setup_vedo_entry(hass, config_entry, async_add_entities)
coordinator = config_entry.runtime_data
is_bridge = isinstance(coordinator, ComelitSerialBridge)
if TYPE_CHECKING:
if is_bridge:
assert isinstance(coordinator, ComelitSerialBridge)
else:
assert isinstance(coordinator, ComelitVedoSystem)
async def async_setup_bridge_entry(
hass: HomeAssistant,
config_entry: ComelitConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Comelit Bridge sensors."""
coordinator = cast(ComelitSerialBridge, config_entry.runtime_data)
def _add_new_entities(new_devices: list[ObjectClassType], dev_type: str) -> None:
def _add_new_bridge_entities(
new_devices: list[ObjectClassType], dev_type: str
) -> None:
"""Add entities for new monitors."""
assert isinstance(coordinator, ComelitSerialBridge)
entities = [
ComelitBridgeSensorEntity(
coordinator, device, config_entry.entry_id, sensor_desc
@@ -80,36 +77,32 @@ async def async_setup_bridge_entry(
if entities:
async_add_entities(entities)
config_entry.async_on_unload(
new_device_listener(coordinator, _add_new_entities, OTHER)
)
async def async_setup_vedo_entry(
hass: HomeAssistant,
config_entry: ComelitConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Comelit VEDO sensors."""
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
def _add_new_entities(new_devices: list[ObjectClassType], dev_type: str) -> None:
def _add_new_vedo_entities(
new_devices: list[ObjectClassType], dev_type: str
) -> None:
"""Add entities for new monitors."""
entities = [
ComelitVedoSensorEntity(
coordinator, device, config_entry.entry_id, sensor_desc
)
for sensor_desc in SENSOR_VEDO_TYPES
for device in coordinator.data["alarm_zones"].values()
for device in coordinator.data[dev_type].values()
if device in new_devices
]
if entities:
async_add_entities(entities)
config_entry.async_on_unload(
new_device_listener(coordinator, _add_new_entities, "alarm_zones")
)
# Bridge native sensors
if is_bridge:
config_entry.async_on_unload(
new_device_listener(coordinator, _add_new_bridge_entities, OTHER)
)
# Alarm sensors (both via Bridge or VedoSystem)
if coordinator.vedo_pin:
config_entry.async_on_unload(
new_device_listener(coordinator, _add_new_vedo_entities, ALARM_ZONE)
)
class ComelitBridgeSensorEntity(ComelitBridgeBaseEntity, SensorEntity):
@@ -141,14 +134,16 @@ class ComelitBridgeSensorEntity(ComelitBridgeBaseEntity, SensorEntity):
)
class ComelitVedoSensorEntity(CoordinatorEntity[ComelitVedoSystem], SensorEntity):
class ComelitVedoSensorEntity(
CoordinatorEntity[ComelitVedoSystem | ComelitSerialBridge], SensorEntity
):
"""Sensor device."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: ComelitVedoSystem,
coordinator: ComelitVedoSystem | ComelitSerialBridge,
zone: ComelitVedoZoneObject,
config_entry_entry_id: str,
description: SensorEntityDescription,
@@ -166,7 +161,9 @@ class ComelitVedoSensorEntity(CoordinatorEntity[ComelitVedoSystem], SensorEntity
@property
def _zone_object(self) -> ComelitVedoZoneObject:
"""Zone object."""
return self.coordinator.data["alarm_zones"][self._zone_index]
return cast(
ComelitVedoZoneObject, self.coordinator.data[ALARM_ZONE][self._zone_index]
)
@property
def available(self) -> bool:

22
homeassistant/components/comelit/strings.json
View File

@@ -5,6 +5,8 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_pin": "The provided PIN is invalid. It must be a 4-10 digit number.",
"invalid_vedo_auth": "The provided VEDO PIN is incorrect or VEDO alarm is not enabled on this device.",
"invalid_vedo_pin": "The provided VEDO PIN is invalid. It must be a 4-10 digit number.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
@@ -13,28 +15,34 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_pin": "[%key:component::comelit::config::abort::invalid_pin%]",
"invalid_vedo_auth": "[%key:component::comelit::config::abort::invalid_vedo_auth%]",
"invalid_vedo_pin": "[%key:component::comelit::config::abort::invalid_vedo_pin%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"flow_title": "{host}",
"step": {
"reauth_confirm": {
"data": {
"pin": "[%key:common::config_flow::data::pin%]"
"pin": "[%key:common::config_flow::data::pin%]",
"vedo_pin": "[%key:component::comelit::config::step::user::data::vedo_pin%]"
},
"data_description": {
"pin": "The PIN of your Comelit device."
"pin": "The PIN of your Comelit device.",
"vedo_pin": "[%key:component::comelit::config::step::user::data_description::vedo_pin%]"
}
},
"reconfigure": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"pin": "[%key:common::config_flow::data::pin%]",
"port": "[%key:common::config_flow::data::port%]"
"port": "[%key:common::config_flow::data::port%]",
"vedo_pin": "[%key:component::comelit::config::step::user::data::vedo_pin%]"
},
"data_description": {
"host": "[%key:component::comelit::config::step::user::data_description::host%]",
"pin": "[%key:component::comelit::config::step::reauth_confirm::data_description::pin%]",
"port": "[%key:component::comelit::config::step::user::data_description::port%]"
"port": "[%key:component::comelit::config::step::user::data_description::port%]",
"vedo_pin": "[%key:component::comelit::config::step::user::data_description::vedo_pin%]"
}
},
"user": {
@@ -42,13 +50,15 @@
"host": "[%key:common::config_flow::data::host%]",
"pin": "[%key:common::config_flow::data::pin%]",
"port": "[%key:common::config_flow::data::port%]",
"type": "Device type"
"type": "Device type",
"vedo_pin": "VEDO alarm PIN (optional)"
},
"data_description": {
"host": "The hostname or IP address of your Comelit device.",
"pin": "[%key:component::comelit::config::step::reauth_confirm::data_description::pin%]",
"port": "The port of your Comelit device.",
"type": "The type of your Comelit device."
"type": "The type of your Comelit device.",
"vedo_pin": "Optional PIN for VEDO alarm system on Serial Bridge devices. Leave empty if you don't have VEDO alarm enabled."
}
}
}

2
homeassistant/components/comelit/switch.py
View File

@@ -82,7 +82,7 @@ class ComelitSwitchEntity(ComelitBridgeBaseEntity, SwitchEntity):
@property
def is_on(self) -> bool:
"""Return True if switch is on."""
return (
return bool(
self.coordinator.data[self._device.type][self._device.index].status
== STATE_ON
)

2
requirements_all.txt generated
View File

@@ -222,7 +222,7 @@ aiobafi6==0.9.0
aiobotocore==2.21.1
# homeassistant.components.comelit
aiocomelit==1.1.2
aiocomelit==2.0.0
# homeassistant.components.dhcp
aiodhcpwatcher==1.2.1

2
requirements_test_all.txt generated
View File

@@ -213,7 +213,7 @@ aiobafi6==0.9.0
aiobotocore==2.21.1
# homeassistant.components.comelit
aiocomelit==1.1.2
aiocomelit==2.0.0
# homeassistant.components.dhcp
aiodhcpwatcher==1.2.1

2
tests/components/comelit/conftest.py
View File

@@ -46,6 +46,8 @@ def mock_serial_bridge() -> Generator[AsyncMock]:
):
bridge = mock_comelit_serial_bridge.return_value
bridge.get_all_devices.return_value = deepcopy(BRIDGE_DEVICE_QUERY)
bridge.get_all_areas_and_zones.return_value = deepcopy(VEDO_DEVICE_QUERY)
bridge.vedo_enabled.return_value = True
bridge.host = BRIDGE_HOST
bridge.port = BRIDGE_PORT
bridge.device_pin = BRIDGE_PIN

12
tests/components/comelit/const.py
View File

@@ -1,12 +1,13 @@
"""Common stuff for Comelit SimpleHome tests."""
from aiocomelit.api import (
AlarmDataObject,
ComelitSerialBridgeObject,
ComelitVedoAreaObject,
ComelitVedoZoneObject,
)
from aiocomelit.const import (
ALARM_AREA,
ALARM_ZONE,
CLIMATE,
COVER,
IRRIGATION,
@@ -21,6 +22,7 @@ from aiocomelit.const import (
BRIDGE_HOST = "fake_bridge_host"
BRIDGE_PORT = 80
BRIDGE_PIN = "1234"
BRIDGE_VEDO_PIN = "5678"
VEDO_HOST = "fake_vedo_host"
VEDO_PORT = 8080
@@ -102,8 +104,8 @@ ZONE0 = ComelitVedoZoneObject(
status=0,
human_status=AlarmZoneState.REST,
)
VEDO_DEVICE_QUERY = AlarmDataObject(
alarm_areas={
VEDO_DEVICE_QUERY = {
ALARM_AREA: {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
@@ -120,7 +122,7 @@ VEDO_DEVICE_QUERY = AlarmDataObject(
human_status=AlarmAreaState.DISARMED,
)
},
alarm_zones={
ALARM_ZONE: {
0: ZONE0,
},
)
}

12
tests/components/comelit/test_alarm_control_panel.py
View File

@@ -2,8 +2,8 @@
from unittest.mock import AsyncMock
from aiocomelit.api import AlarmDataObject, ComelitVedoAreaObject
from aiocomelit.const import AlarmAreaState
from aiocomelit.api import ComelitVedoAreaObject
from aiocomelit.const import ALARM_AREA, ALARM_ZONE, AlarmAreaState
from freezegun.api import FrozenDateTimeFactory
import pytest
@@ -55,8 +55,8 @@ async def test_entity_availability(
assert (state := hass.states.get(ENTITY_ID))
assert state.state == AlarmControlPanelState.DISARMED
vedo_query = AlarmDataObject(
alarm_areas={
vedo_query = {
ALARM_AREA: {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
@@ -73,10 +73,10 @@ async def test_entity_availability(
human_status=human_status,
)
},
alarm_zones={
ALARM_ZONE: {
0: ZONE0,
},
)
}
mock_vedo.get_all_areas_and_zones.return_value = vedo_query

123
tests/components/comelit/test_config_flow.py
View File

@@ -6,8 +6,12 @@ from aiocomelit import CannotAuthenticate, CannotConnect
from aiocomelit.const import BRIDGE, VEDO
import pytest
from homeassistant.components.comelit.config_flow import InvalidPin
from homeassistant.components.comelit.const import DOMAIN
from homeassistant.components.comelit.config_flow import (
InvalidPin,
InvalidVedoAuth,
InvalidVedoPin,
)
from homeassistant.components.comelit.const import CONF_VEDO_PIN, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PIN, CONF_PORT, CONF_TYPE
from homeassistant.core import HomeAssistant
@@ -18,6 +22,7 @@ from .const import (
BRIDGE_HOST,
BRIDGE_PIN,
BRIDGE_PORT,
BRIDGE_VEDO_PIN,
FAKE_PIN,
VEDO_HOST,
VEDO_PIN,
@@ -99,6 +104,8 @@ async def test_flow_vedo(
(CannotAuthenticate, "invalid_auth"),
(ConnectionResetError, "unknown"),
(InvalidPin, "invalid_pin"),
(InvalidVedoPin, "invalid_vedo_pin"),
(InvalidVedoAuth, "invalid_vedo_auth"),
],
)
async def test_exception_connection(
@@ -248,6 +255,7 @@ async def test_reconfigure_successful(
CONF_HOST: new_host,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BRIDGE_VEDO_PIN,
},
)
@@ -265,6 +273,8 @@ async def test_reconfigure_successful(
(CannotAuthenticate, "invalid_auth"),
(ConnectionResetError, "unknown"),
(InvalidPin, "invalid_pin"),
(InvalidVedoPin, "invalid_vedo_pin"),
(InvalidVedoAuth, "invalid_vedo_auth"),
],
)
async def test_reconfigure_fails(
@@ -359,3 +369,112 @@ async def test_pin_format_serial_bridge(
}
assert not result["result"].unique_id
await hass.async_block_till_done()
async def test_flow_serial_bridge_with_vedo_pin(
hass: HomeAssistant,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test starting a flow by user with VEDO PIN."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# Mock vedo_enabled to return True
mock_serial_bridge.vedo_enabled.return_value = True
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: BRIDGE_HOST,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BRIDGE_VEDO_PIN,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_HOST: BRIDGE_HOST,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BRIDGE_VEDO_PIN,
CONF_TYPE: BRIDGE,
}
assert not result["result"].unique_id
await hass.async_block_till_done()
async def test_flow_serial_bridge_with_invalid_vedo_pin(
hass: HomeAssistant,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test starting a flow with invalid VEDO PIN."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: BRIDGE_HOST,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BAD_PIN,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_vedo_pin"}
# Test with correct VEDO PIN
mock_serial_bridge.vedo_enabled.return_value = True
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: BRIDGE_HOST,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BRIDGE_VEDO_PIN,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_flow_serial_bridge_with_vedo_auth_failure(
hass: HomeAssistant,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test starting a flow with VEDO authentication failure."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# Mock vedo_enabled to return False (authentication failed)
mock_serial_bridge.vedo_enabled.return_value = False
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: BRIDGE_HOST,
CONF_PORT: BRIDGE_PORT,
CONF_PIN: BRIDGE_PIN,
CONF_VEDO_PIN: BRIDGE_VEDO_PIN,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_vedo_auth"}

19
tests/components/comelit/test_coordinator.py
View File

@@ -3,12 +3,13 @@
from unittest.mock import AsyncMock
from aiocomelit.api import (
AlarmDataObject,
ComelitSerialBridgeObject,
ComelitVedoAreaObject,
ComelitVedoZoneObject,
)
from aiocomelit.const import (
ALARM_AREA,
ALARM_ZONE,
CLIMATE,
COVER,
IRRIGATION,
@@ -139,8 +140,8 @@ async def test_coordinator_stale_device_vedo(
entity_id_0 = "sensor.zone0"
entity_id_1 = "sensor.zone1"
mock_vedo.get_all_areas_and_zones.return_value = AlarmDataObject(
alarm_areas={
mock_vedo.get_all_areas_and_zones.return_value = {
ALARM_AREA: {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
@@ -157,7 +158,7 @@ async def test_coordinator_stale_device_vedo(
human_status=AlarmAreaState.DISARMED,
)
},
alarm_zones={
ALARM_ZONE: {
0: ZONE0,
1: ComelitVedoZoneObject(
index=1,
@@ -167,7 +168,7 @@ async def test_coordinator_stale_device_vedo(
human_status=AlarmZoneState.REST,
),
},
)
}
await setup_integration(hass, mock_vedo_config_entry)
assert (state := hass.states.get(entity_id_0))
@@ -175,8 +176,8 @@ async def test_coordinator_stale_device_vedo(
assert (state := hass.states.get(entity_id_1))
assert state.state == AlarmZoneState.REST.value
mock_vedo.get_all_areas_and_zones.return_value = AlarmDataObject(
alarm_areas={
mock_vedo.get_all_areas_and_zones.return_value = {
ALARM_AREA: {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
@@ -193,10 +194,10 @@ async def test_coordinator_stale_device_vedo(
human_status=AlarmAreaState.DISARMED,
)
},
alarm_zones={
ALARM_ZONE: {
0: ZONE0,
},
)
}
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)

20
tests/components/comelit/test_sensor.py
View File

@@ -3,12 +3,18 @@
from unittest.mock import AsyncMock, patch
from aiocomelit.api import (
AlarmDataObject,
ComelitSerialBridgeObject,
ComelitVedoAreaObject,
ComelitVedoZoneObject,
)
from aiocomelit.const import OTHER, WATT, AlarmAreaState, AlarmZoneState
from aiocomelit.const import (
ALARM_AREA,
ALARM_ZONE,
OTHER,
WATT,
AlarmAreaState,
AlarmZoneState,
)
from freezegun.api import FrozenDateTimeFactory
from syrupy.assertion import SnapshotAssertion
@@ -56,8 +62,8 @@ async def test_sensor_state_unknown(
assert (state := hass.states.get(ENTITY_ID))
assert state.state == AlarmZoneState.REST.value
vedo_query = AlarmDataObject(
alarm_areas={
vedo_query = {
ALARM_AREA: {
0: ComelitVedoAreaObject(
index=0,
name="Area0",
@@ -74,7 +80,7 @@ async def test_sensor_state_unknown(
human_status=AlarmAreaState.UNKNOWN,
)
},
alarm_zones={
ALARM_ZONE: {
0: ComelitVedoZoneObject(
index=0,
name="Zone0",
@@ -83,7 +89,7 @@ async def test_sensor_state_unknown(
human_status=AlarmZoneState.UNKNOWN,
)
},
)
}
mock_vedo.get_all_areas_and_zones.return_value = vedo_query
@@ -160,7 +166,7 @@ async def test_vedo_sensor_dynamic(
entity_id_2 = "sensor.zone1"
mock_vedo.get_all_areas_and_zones.return_value["alarm_zones"] = {
mock_vedo.get_all_areas_and_zones.return_value[ALARM_ZONE] = {
0: ComelitVedoZoneObject(
index=0,
name="Zone0",