Compare commits

...

4 Commits

Author SHA1 Message Date
copilot-swe-agent[bot]
9f79a1831e Fix linting issues and format code
Co-authored-by: chemelli74 <57354320+chemelli74@users.noreply.github.com>
2025-10-25 10:27:42 +00:00
copilot-swe-agent[bot]
b541934a75 Add comprehensive tests for VEDO PIN functionality
Co-authored-by: chemelli74 <57354320+chemelli74@users.noreply.github.com>
2025-10-25 10:24:59 +00:00
copilot-swe-agent[bot]
fa5c0c0bf2 Add optional VEDO PIN support to Comelit BRIDGE integration
Co-authored-by: chemelli74 <57354320+chemelli74@users.noreply.github.com>
2025-10-25 10:22:50 +00:00
copilot-swe-agent[bot]
1effbf3326 Initial plan 2025-10-25 10:07:33 +00:00
12 changed files with 576 additions and 36 deletions

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,
@@ -43,9 +43,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ComelitConfigEntry) -> b
entry.data[CONF_HOST],
entry.data.get(CONF_PORT, DEFAULT_PORT),
entry.data[CONF_PIN],
entry.data.get(CONF_VEDO_PIN),
session,
)
platforms = BRIDGE_PLATFORMS
platforms = list(BRIDGE_PLATFORMS)
# Add VEDO platforms if vedo_pin is configured
if entry.data.get(CONF_VEDO_PIN):
platforms.extend(VEDO_PLATFORMS)
else:
coordinator = ComelitVedoSystem(
hass,
@@ -70,7 +74,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ComelitConfigEntry) ->
"""Unload a config entry."""
if entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
platforms = BRIDGE_PLATFORMS
platforms = list(BRIDGE_PLATFORMS)
# Add VEDO platforms if vedo_pin was configured
if entry.data.get(CONF_VEDO_PIN):
platforms.extend(VEDO_PLATFORMS)
else:
platforms = VEDO_PLATFORMS

View File

@@ -6,7 +6,7 @@ import logging
from typing import cast
from aiocomelit.api import ComelitVedoAreaObject
from aiocomelit.const import AlarmAreaState
from aiocomelit.const import BRIDGE, AlarmAreaState
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
@@ -14,11 +14,13 @@ from homeassistant.components.alarm_control_panel import (
AlarmControlPanelState,
CodeFormat,
)
from homeassistant.const import CONF_TYPE
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
from .utils import DeviceType, alarm_device_listener
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@@ -56,12 +58,34 @@ async def async_setup_entry(
) -> None:
"""Set up the Comelit VEDO system alarm control panel devices."""
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
if config_entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
coordinator = cast(ComelitSerialBridge, config_entry.runtime_data)
# Only setup if bridge has VEDO alarm enabled
if not coordinator.vedo_pin:
return
async_add_entities(
ComelitAlarmEntity(coordinator, device, config_entry.entry_id)
for device in coordinator.data["alarm_areas"].values()
)
def _add_new_entities(new_devices: list[DeviceType], dev_type: str) -> None:
"""Add entities for new monitors."""
entities = [
ComelitBridgeAlarmEntity(coordinator, device, config_entry.entry_id)
for device in (coordinator.alarm_data or {})
.get("alarm_areas", {})
.values()
if device in new_devices
]
if entities:
async_add_entities(entities)
config_entry.async_on_unload(
alarm_device_listener(coordinator, _add_new_entities, "alarm_areas")
)
else:
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
async_add_entities(
ComelitAlarmEntity(coordinator, device, config_entry.entry_id)
for device in coordinator.data["alarm_areas"].values()
)
class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanelEntity):
@@ -171,3 +195,133 @@ class ComelitAlarmEntity(CoordinatorEntity[ComelitVedoSystem], AlarmControlPanel
await self._async_update_state(
AlarmAreaState.ARMED, ALARM_AREA_ARMED_STATUS[NIGHT]
)
class ComelitBridgeAlarmEntity(
CoordinatorEntity[ComelitSerialBridge], AlarmControlPanelEntity
):
"""Representation of a VEDO alarm panel on a Serial Bridge."""
_attr_has_entity_name = True
_attr_name = None
_attr_code_format = CodeFormat.NUMBER
_attr_code_arm_required = False
_attr_supported_features = (
AlarmControlPanelEntityFeature.ARM_AWAY
| AlarmControlPanelEntityFeature.ARM_HOME
)
def __init__(
self,
coordinator: ComelitSerialBridge,
area: ComelitVedoAreaObject,
config_entry_entry_id: str,
) -> None:
"""Initialize the alarm panel."""
self._area_index = area.index
super().__init__(coordinator)
# Use config_entry.entry_id as base for unique_id
# because no serial number or mac is available
self._attr_unique_id = f"{config_entry_entry_id}-{area.index}"
self._attr_device_info = coordinator.platform_device_info(area, "area")
if area.p2:
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_NIGHT
@property
def _area(self) -> ComelitVedoAreaObject:
"""Return area object."""
if self.coordinator.alarm_data:
return self.coordinator.alarm_data["alarm_areas"][self._area_index]
# Return a default area object if no alarm data
return ComelitVedoAreaObject(
index=self._area_index,
name="Unknown",
p1=False,
p2=False,
ready=False,
armed=0,
alarm=False,
alarm_memory=False,
sabotage=False,
anomaly=False,
in_time=False,
out_time=False,
human_status=AlarmAreaState.UNKNOWN,
)
@property
def available(self) -> bool:
"""Return True if alarm is available."""
if not self.coordinator.alarm_data:
return False
if self._area.human_status in [AlarmAreaState.ANOMALY, AlarmAreaState.UNKNOWN]:
return False
return super().available
@property
def alarm_state(self) -> AlarmControlPanelState | None:
"""Return the state of the alarm."""
_LOGGER.debug(
"Area %s status is: %s. Armed is %s",
self._area.name,
self._area.human_status,
self._area.armed,
)
if self._area.human_status == AlarmAreaState.ARMED:
if self._area.armed == ALARM_AREA_ARMED_STATUS[AWAY]:
return AlarmControlPanelState.ARMED_AWAY
if self._area.armed == ALARM_AREA_ARMED_STATUS[NIGHT]:
return AlarmControlPanelState.ARMED_NIGHT
return AlarmControlPanelState.ARMED_HOME
return {
AlarmAreaState.DISARMED: AlarmControlPanelState.DISARMED,
AlarmAreaState.ENTRY_DELAY: AlarmControlPanelState.DISARMING,
AlarmAreaState.EXIT_DELAY: AlarmControlPanelState.ARMING,
AlarmAreaState.TRIGGERED: AlarmControlPanelState.TRIGGERED,
}.get(self._area.human_status)
async def _async_update_state(self, area_state: AlarmAreaState, armed: int) -> None:
"""Update state after action."""
self._area.human_status = area_state
self._area.armed = armed
await self.async_update_ha_state()
async def async_alarm_disarm(self, code: str | None = None) -> None:
"""Send disarm command."""
if code != str(self.coordinator.vedo_pin):
return
await self.coordinator.api.set_zone_status(
self._area.index, ALARM_ACTIONS[DISABLE]
)
await self._async_update_state(
AlarmAreaState.DISARMED, ALARM_AREA_ARMED_STATUS[DISABLE]
)
async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command."""
await self.coordinator.api.set_zone_status(
self._area.index, ALARM_ACTIONS[AWAY]
)
await self._async_update_state(
AlarmAreaState.ARMED, ALARM_AREA_ARMED_STATUS[AWAY]
)
async def async_alarm_arm_home(self, code: str | None = None) -> None:
"""Send arm home command."""
await self.coordinator.api.set_zone_status(
self._area.index, ALARM_ACTIONS[HOME]
)
await self._async_update_state(
AlarmAreaState.ARMED, ALARM_AREA_ARMED_STATUS[HOME_P1]
)
async def async_alarm_arm_night(self, code: str | None = None) -> None:
"""Send arm night command."""
await self.coordinator.api.set_zone_status(
self._area.index, ALARM_ACTIONS[NIGHT]
)
await self._async_update_state(
AlarmAreaState.ARMED, ALARM_AREA_ARMED_STATUS[NIGHT]
)

View File

@@ -5,17 +5,19 @@ from __future__ import annotations
from typing import cast
from aiocomelit import ComelitVedoZoneObject
from aiocomelit.const import BRIDGE
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_TYPE
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 .utils import DeviceType, new_device_listener
from .coordinator import ComelitConfigEntry, ComelitSerialBridge, ComelitVedoSystem
from .utils import DeviceType, alarm_device_listener, new_device_listener
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@@ -28,21 +30,47 @@ async def async_setup_entry(
) -> None:
"""Set up Comelit VEDO presence sensors."""
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
if config_entry.data.get(CONF_TYPE, BRIDGE) == BRIDGE:
coordinator = cast(ComelitSerialBridge, config_entry.runtime_data)
# Only setup if bridge has VEDO alarm enabled
if not coordinator.vedo_pin:
return
def _add_new_entities(new_devices: list[DeviceType], 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()
if device in new_devices
]
if entities:
async_add_entities(entities)
def _add_new_entities(new_devices: list[DeviceType], dev_type: str) -> None:
"""Add entities for new monitors."""
entities = [
ComelitVedoBridgeBinarySensorEntity(
coordinator, device, config_entry.entry_id
)
for device in (coordinator.alarm_data or {})
.get("alarm_zones", {})
.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")
)
config_entry.async_on_unload(
alarm_device_listener(coordinator, _add_new_entities, "alarm_zones")
)
else:
coordinator = cast(ComelitVedoSystem, config_entry.runtime_data)
def _add_new_entities(new_devices: list[DeviceType], 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()
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")
)
class ComelitVedoBinarySensorEntity(
@@ -73,3 +101,41 @@ class ComelitVedoBinarySensorEntity(
return (
self.coordinator.data["alarm_zones"][self._zone_index].status_api == "0001"
)
class ComelitVedoBridgeBinarySensorEntity(
CoordinatorEntity[ComelitSerialBridge], BinarySensorEntity
):
"""VEDO sensor device on a Serial Bridge."""
_attr_has_entity_name = True
_attr_device_class = BinarySensorDeviceClass.MOTION
def __init__(
self,
coordinator: ComelitSerialBridge,
zone: ComelitVedoZoneObject,
config_entry_entry_id: str,
) -> None:
"""Init sensor entity."""
self._zone_index = zone.index
super().__init__(coordinator)
# Use config_entry.entry_id as base for unique_id
# because no serial number or mac is available
self._attr_unique_id = f"{config_entry_entry_id}-presence-{zone.index}"
self._attr_device_info = coordinator.platform_device_info(zone, "zone")
@property
def available(self) -> bool:
"""Sensor availability."""
return self.coordinator.alarm_data is not None
@property
def is_on(self) -> bool:
"""Presence detected."""
if not self.coordinator.alarm_data:
return False
return (
self.coordinator.alarm_data["alarm_zones"][self._zone_index].status_api
== "0001"
)

View File

@@ -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,6 +34,7 @@ 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})
@@ -42,6 +43,7 @@ STEP_RECONFIGURE = vol.Schema(
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORT): cv.port,
vol.Optional(CONF_PIN, default=DEFAULT_PIN): cv.string,
vol.Optional(CONF_VEDO_PIN): cv.string,
}
)
@@ -79,6 +81,27 @@ 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
# Verify VEDO is enabled with the provided PIN
try:
if not await api.vedo_enabled(data[CONF_VEDO_PIN]):
raise InvalidVedoAuth
except (aiocomelit_exceptions.CannotConnect, TimeoutError) as err:
raise CannotConnect(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
except aiocomelit_exceptions.CannotAuthenticate:
raise InvalidVedoAuth(
translation_domain=DOMAIN,
translation_key="invalid_vedo_auth",
) from None
return {"title": data[CONF_HOST]}
@@ -106,6 +129,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"
@@ -187,19 +214,38 @@ class ComelitConfigFlow(ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
try:
await validate_input(self.hass, user_input)
data_to_validate = {
CONF_HOST: updated_host,
CONF_PORT: user_input[CONF_PORT],
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"
except InvalidAuth:
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"
else:
data_updates = {
CONF_HOST: updated_host,
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={CONF_HOST: updated_host}
reconfigure_entry, data_updates=data_updates
)
return self.async_show_form(
@@ -219,3 +265,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."""

View File

@@ -9,6 +9,7 @@ _LOGGER = logging.getLogger(__package__)
DOMAIN = "comelit"
DEFAULT_PORT = 80
DEVICE_TYPE_LIST = [BRIDGE, VEDO]
CONF_VEDO_PIN = "vedo_pin"
SCAN_INTERVAL = 5

View File

@@ -154,6 +154,8 @@ class ComelitSerialBridge(
_hw_version = "20003101"
api: ComeliteSerialBridgeApi
vedo_pin: str | None
alarm_data: AlarmDataObject | None = None
def __init__(
self,
@@ -162,25 +164,49 @@ 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]]:
"""Specific method for updating data."""
data = await self.api.get_all_devices()
devices = await self.api.get_all_devices()
if self.data:
for dev_type in (CLIMATE, COVER, LIGHT, IRRIGATION, OTHER, SCENARIO):
await self._async_remove_stale_devices(
self.data[dev_type], data[dev_type], dev_type
self.data[dev_type], devices[dev_type], dev_type
)
return data
# Get VEDO alarm data if vedo_pin is configured
if self.vedo_pin:
try:
if await self.api.vedo_enabled(self.vedo_pin):
self.alarm_data = await self.api.get_all_areas_and_zones()
# Remove stale alarm devices
if self.alarm_data:
previous_alarm_data = getattr(
self, "_previous_alarm_data", None
)
if previous_alarm_data:
for obj_type in ("alarm_areas", "alarm_zones"):
await self._async_remove_stale_devices(
previous_alarm_data[obj_type],
self.alarm_data[obj_type],
"area" if obj_type == "alarm_areas" else "zone",
)
self._previous_alarm_data = self.alarm_data
except (CannotAuthenticate, CannotConnect, CannotRetrieveData):
_LOGGER.warning("Failed to retrieve VEDO alarm data")
return devices
class ComelitVedoSystem(ComelitBaseCoordinator[AlarmDataObject]):

View File

@@ -20,7 +20,7 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import ComelitConfigEntry, ComelitSerialBridge, ComelitVedoSystem
from .entity import ComelitBridgeBaseEntity
from .utils import DeviceType, new_device_listener
from .utils import DeviceType, alarm_device_listener, new_device_listener
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@@ -83,6 +83,30 @@ async def async_setup_bridge_entry(
new_device_listener(coordinator, _add_new_entities, OTHER)
)
# Add VEDO sensors if bridge has alarm data
if coordinator.vedo_pin:
def _add_new_alarm_entities(
new_devices: list[DeviceType], dev_type: str
) -> None:
"""Add entities for new alarm zones."""
entities = [
ComelitVedoBridgeSensorEntity(
coordinator, device, config_entry.entry_id, sensor_desc
)
for sensor_desc in SENSOR_VEDO_TYPES
for device in (coordinator.alarm_data or {})
.get("alarm_zones", {})
.values()
if device in new_devices
]
if entities:
async_add_entities(entities)
config_entry.async_on_unload(
alarm_device_listener(coordinator, _add_new_alarm_entities, "alarm_zones")
)
async def async_setup_vedo_entry(
hass: HomeAssistant,
@@ -179,3 +203,58 @@ class ComelitVedoSensorEntity(CoordinatorEntity[ComelitVedoSystem], SensorEntity
return None
return cast(str, status.value)
class ComelitVedoBridgeSensorEntity(
CoordinatorEntity[ComelitSerialBridge], SensorEntity
):
"""VEDO sensor device on a Serial Bridge."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: ComelitSerialBridge,
zone: ComelitVedoZoneObject,
config_entry_entry_id: str,
description: SensorEntityDescription,
) -> None:
"""Init sensor entity."""
self._zone_index = zone.index
super().__init__(coordinator)
# Use config_entry.entry_id as base for unique_id
# because no serial number or mac is available
self._attr_unique_id = f"{config_entry_entry_id}-{zone.index}"
self._attr_device_info = coordinator.platform_device_info(zone, "zone")
self.entity_description = description
@property
def _zone_object(self) -> ComelitVedoZoneObject:
"""Zone object."""
if self.coordinator.alarm_data:
return self.coordinator.alarm_data["alarm_zones"][self._zone_index]
# Return a default zone object if no alarm data
return ComelitVedoZoneObject(
index=self._zone_index,
name="Unknown",
status_api="0x000",
status=0,
human_status=AlarmZoneState.UNAVAILABLE,
)
@property
def available(self) -> bool:
"""Sensor availability."""
return (
self.coordinator.alarm_data is not None
and self._zone_object.human_status != AlarmZoneState.UNAVAILABLE
)
@property
def native_value(self) -> StateType:
"""Sensor value."""
if (status := self._zone_object.human_status) == AlarmZoneState.UNKNOWN:
return None
return cast(str, status.value)

View File

@@ -15,25 +15,29 @@
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"pin": "[%key:common::config_flow::data::pin%]",
"type": "Device type"
"type": "Device type",
"vedo_pin": "VEDO alarm PIN (optional)"
},
"data_description": {
"host": "The hostname or IP address of your Comelit device.",
"port": "The port of your Comelit device.",
"pin": "[%key:component::comelit::config::step::reauth_confirm::data_description::pin%]",
"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."
}
},
"reconfigure": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"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": {
"host": "[%key:component::comelit::config::step::user::data_description::host%]",
"port": "[%key:component::comelit::config::step::user::data_description::port%]",
"pin": "[%key:component::comelit::config::step::reauth_confirm::data_description::pin%]"
"pin": "[%key:component::comelit::config::step::reauth_confirm::data_description::pin%]",
"vedo_pin": "[%key:component::comelit::config::step::user::data_description::vedo_pin%]"
}
}
},
@@ -44,12 +48,16 @@
"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_pin": "The provided VEDO 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.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"error": {
"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_pin": "[%key:component::comelit::config::abort::invalid_vedo_pin%]",
"invalid_vedo_auth": "[%key:component::comelit::config::abort::invalid_vedo_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},

View File

@@ -158,3 +158,35 @@ def new_device_listener(
_check_devices()
return coordinator.async_add_listener(_check_devices)
def alarm_device_listener(
coordinator: ComelitBaseCoordinator,
new_devices_callback: Callable[
[list[ComelitVedoAreaObject | ComelitVedoZoneObject], str],
None,
],
data_type: str,
) -> Callable[[], None]:
"""Subscribe to coordinator updates to check for new alarm devices on bridge."""
known_devices: dict[str, list[int]] = {}
def _check_alarm_devices() -> None:
"""Check for new alarm devices and call callback with any new devices."""
# For ComelitSerialBridge with alarm_data
if not hasattr(coordinator, "alarm_data") or not coordinator.alarm_data:
return
new_devices: list[ComelitVedoAreaObject | ComelitVedoZoneObject] = []
for _id in coordinator.alarm_data[data_type]:
if _id not in (id_list := known_devices.get(data_type, [])):
known_devices.update({data_type: [*id_list, _id]})
new_devices.append(coordinator.alarm_data[data_type][_id])
if new_devices:
new_devices_callback(new_devices, data_type)
# Check for devices immediately
_check_alarm_devices()
return coordinator.async_add_listener(_check_alarm_devices)

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

View File

@@ -21,6 +21,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

View File

@@ -18,6 +18,7 @@ from .const import (
BRIDGE_HOST,
BRIDGE_PIN,
BRIDGE_PORT,
BRIDGE_VEDO_PIN,
FAKE_PIN,
VEDO_HOST,
VEDO_PIN,
@@ -359,3 +360,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,
"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,
"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,
"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,
"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,
"vedo_pin": BRIDGE_VEDO_PIN,
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_vedo_auth"}