Compare commits

..

4 Commits

Author SHA1 Message Date
Ludovic BOUÉ
08eee8d479 Update tests/components/matter/test_switch.py
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-10-29 11:40:54 +01:00
Ludovic BOUÉ
6bbaae7235 Update homeassistant/components/matter/switch.py
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-10-29 11:40:39 +01:00
Ludovic BOUÉ
86a5dff3f5 Add test for MatterError handling in numeric switch with Eve Thermo fixture 2025-10-29 10:15:35 +00:00
Ludovic BOUÉ
34e137005d Add error handling for Matter switch commands 2025-10-29 10:08:43 +00:00
132 changed files with 2227 additions and 3487 deletions

View File

@@ -361,6 +361,7 @@ homeassistant.components.myuplink.*
homeassistant.components.nam.*
homeassistant.components.nanoleaf.*
homeassistant.components.nasweb.*
homeassistant.components.neato.*
homeassistant.components.nest.*
homeassistant.components.netatmo.*
homeassistant.components.network.*

View File

@@ -28,5 +28,5 @@
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/airthings_ble",
"iot_class": "local_polling",
"requirements": ["airthings-ble==1.2.0"]
"requirements": ["airthings-ble==1.1.1"]
}

View File

@@ -26,6 +26,3 @@ COUNTRY_DOMAINS = {
"us": DEFAULT_DOMAIN,
"za": "co.za",
}
CATEGORY_SENSORS = "sensors"
CATEGORY_NOTIFICATIONS = "notifications"

View File

@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["aioamazondevices"],
"quality_scale": "platinum",
"requirements": ["aioamazondevices==6.5.5"]
"requirements": ["aioamazondevices==6.4.6"]
}

View File

@@ -4,15 +4,9 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import Final
from aioamazondevices.api import AmazonDevice
from aioamazondevices.const import (
NOTIFICATION_ALARM,
NOTIFICATION_REMINDER,
NOTIFICATION_TIMER,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -25,7 +19,6 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import CATEGORY_NOTIFICATIONS, CATEGORY_SENSORS
from .coordinator import AmazonConfigEntry
from .entity import AmazonEntity
@@ -43,20 +36,6 @@ class AmazonSensorEntityDescription(SensorEntityDescription):
and (sensor := device.sensors.get(key)) is not None
and sensor.error is False
)
category: str = CATEGORY_SENSORS
@dataclass(frozen=True, kw_only=True)
class AmazonNotificationEntityDescription(SensorEntityDescription):
"""Amazon Devices notification entity description."""
native_unit_of_measurement_fn: Callable[[AmazonDevice, str], str] | None = None
is_available_fn: Callable[[AmazonDevice, str], bool] = lambda device, key: (
device.online
and (notification := device.notifications.get(key)) is not None
and notification.next_occurrence is not None
)
category: str = CATEGORY_NOTIFICATIONS
SENSORS: Final = (
@@ -77,23 +56,6 @@ SENSORS: Final = (
state_class=SensorStateClass.MEASUREMENT,
),
)
NOTIFICATIONS: Final = (
AmazonNotificationEntityDescription(
key=NOTIFICATION_ALARM,
translation_key="alarm",
device_class=SensorDeviceClass.TIMESTAMP,
),
AmazonNotificationEntityDescription(
key=NOTIFICATION_REMINDER,
translation_key="reminder",
device_class=SensorDeviceClass.TIMESTAMP,
),
AmazonNotificationEntityDescription(
key=NOTIFICATION_TIMER,
translation_key="timer",
device_class=SensorDeviceClass.TIMESTAMP,
),
)
async def async_setup_entry(
@@ -112,18 +74,12 @@ async def async_setup_entry(
new_devices = current_devices - known_devices
if new_devices:
known_devices.update(new_devices)
sensors_list = [
async_add_entities(
AmazonSensorEntity(coordinator, serial_num, sensor_desc)
for sensor_desc in SENSORS
for serial_num in new_devices
if coordinator.data[serial_num].sensors.get(sensor_desc.key) is not None
]
notifications_list = [
AmazonSensorEntity(coordinator, serial_num, notification_desc)
for notification_desc in NOTIFICATIONS
for serial_num in new_devices
]
async_add_entities(sensors_list + notifications_list)
)
_check_device()
entry.async_on_unload(coordinator.async_add_listener(_check_device))
@@ -132,9 +88,7 @@ async def async_setup_entry(
class AmazonSensorEntity(AmazonEntity, SensorEntity):
"""Sensor device."""
entity_description: (
AmazonSensorEntityDescription | AmazonNotificationEntityDescription
)
entity_description: AmazonSensorEntityDescription
@property
def native_unit_of_measurement(self) -> str | None:
@@ -147,13 +101,9 @@ class AmazonSensorEntity(AmazonEntity, SensorEntity):
return super().native_unit_of_measurement
@property
def native_value(self) -> StateType | datetime:
def native_value(self) -> StateType:
"""Return the state of the sensor."""
# Sensors
if self.entity_description.category == CATEGORY_SENSORS:
return self.device.sensors[self.entity_description.key].value
# Notifications
return self.device.notifications[self.entity_description.key].next_occurrence
return self.device.sensors[self.entity_description.key].value
@property
def available(self) -> bool:

View File

@@ -66,17 +66,6 @@
"name": "Speak"
}
},
"sensor": {
"alarm": {
"name": "Next alarm"
},
"reminder": {
"name": "Next reminder"
},
"timer": {
"name": "Next timer"
}
},
"switch": {
"do_not_disturb": {
"name": "Do not disturb"

View File

@@ -115,37 +115,26 @@ GACTIONS_SCHEMA = ASSISTANT_SCHEMA.extend(
{vol.Optional(CONF_ENTITY_CONFIG): {cv.entity_id: GOOGLE_ENTITY_SCHEMA}}
)
_BASE_CONFIG_SCHEMA = vol.Schema(
{
vol.Optional(CONF_COGNITO_CLIENT_ID): str,
vol.Optional(CONF_USER_POOL_ID): str,
vol.Optional(CONF_REGION): str,
vol.Optional(CONF_ALEXA): ALEXA_SCHEMA,
vol.Optional(CONF_GOOGLE_ACTIONS): GACTIONS_SCHEMA,
vol.Optional(CONF_ACCOUNT_LINK_SERVER): str,
vol.Optional(CONF_ACCOUNTS_SERVER): str,
vol.Optional(CONF_ACME_SERVER): str,
vol.Optional(CONF_API_SERVER): str,
vol.Optional(CONF_RELAYER_SERVER): str,
vol.Optional(CONF_REMOTESTATE_SERVER): str,
vol.Optional(CONF_SERVICEHANDLERS_SERVER): str,
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Any(
_BASE_CONFIG_SCHEMA.extend(
{
vol.Required(CONF_MODE): vol.In([MODE_DEV]),
vol.Required(CONF_API_SERVER): str,
}
),
_BASE_CONFIG_SCHEMA.extend(
{
vol.Optional(CONF_MODE, default=DEFAULT_MODE): vol.In([MODE_PROD]),
}
),
DOMAIN: vol.Schema(
{
vol.Optional(CONF_MODE, default=DEFAULT_MODE): vol.In(
[MODE_DEV, MODE_PROD]
),
vol.Optional(CONF_COGNITO_CLIENT_ID): str,
vol.Optional(CONF_USER_POOL_ID): str,
vol.Optional(CONF_REGION): str,
vol.Optional(CONF_ALEXA): ALEXA_SCHEMA,
vol.Optional(CONF_GOOGLE_ACTIONS): GACTIONS_SCHEMA,
vol.Optional(CONF_ACCOUNT_LINK_SERVER): str,
vol.Optional(CONF_ACCOUNTS_SERVER): str,
vol.Optional(CONF_ACME_SERVER): str,
vol.Optional(CONF_API_SERVER): str,
vol.Optional(CONF_RELAYER_SERVER): str,
vol.Optional(CONF_REMOTESTATE_SERVER): str,
vol.Optional(CONF_SERVICEHANDLERS_SERVER): str,
}
)
},
extra=vol.ALLOW_EXTRA,

View File

@@ -263,9 +263,6 @@ class Panel:
# Title to show in the sidebar
sidebar_title: str | None = None
# If the panel should be visible by default in the sidebar
sidebar_default_visible: bool = True
# Url to show the panel in the frontend
frontend_url_path: str
@@ -283,7 +280,6 @@ class Panel:
component_name: str,
sidebar_title: str | None,
sidebar_icon: str | None,
sidebar_default_visible: bool,
frontend_url_path: str | None,
config: dict[str, Any] | None,
require_admin: bool,
@@ -297,7 +293,6 @@ class Panel:
self.config = config
self.require_admin = require_admin
self.config_panel_domain = config_panel_domain
self.sidebar_default_visible = sidebar_default_visible
@callback
def to_response(self) -> PanelResponse:
@@ -306,7 +301,6 @@ class Panel:
"component_name": self.component_name,
"icon": self.sidebar_icon,
"title": self.sidebar_title,
"default_visible": self.sidebar_default_visible,
"config": self.config,
"url_path": self.frontend_url_path,
"require_admin": self.require_admin,
@@ -321,7 +315,6 @@ def async_register_built_in_panel(
component_name: str,
sidebar_title: str | None = None,
sidebar_icon: str | None = None,
sidebar_default_visible: bool = True,
frontend_url_path: str | None = None,
config: dict[str, Any] | None = None,
require_admin: bool = False,
@@ -334,7 +327,6 @@ def async_register_built_in_panel(
component_name,
sidebar_title,
sidebar_icon,
sidebar_default_visible,
frontend_url_path,
config,
require_admin,
@@ -887,7 +879,6 @@ class PanelResponse(TypedDict):
component_name: str
icon: str | None
title: str | None
default_visible: bool
config: dict[str, Any] | None
url_path: str
require_admin: bool

View File

@@ -2,35 +2,20 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
import os.path
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.usb import USBDevice, async_register_port_event_callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import DEVICE, DOMAIN, NABU_CASA_FIRMWARE_RELEASES_URL
from .const import DEVICE, DOMAIN
_LOGGER = logging.getLogger(__name__)
type HomeAssistantConnectZBT2ConfigEntry = ConfigEntry[HomeAssistantConnectZBT2Data]
@dataclass
class HomeAssistantConnectZBT2Data:
"""Runtime data definition."""
coordinator: FirmwareUpdateCoordinator
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@@ -64,9 +49,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantConnectZBT2ConfigEntry
) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant Connect ZBT-2 config entry."""
# Postpone loading the config entry if the device is missing
@@ -77,23 +60,12 @@ async def async_setup_entry(
translation_key="device_disconnected",
)
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantConnectZBT2Data(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
await hass.config_entries.async_forward_entry_setups(entry, ["update"])
return True
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantConnectZBT2ConfigEntry
) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True

View File

@@ -3,7 +3,7 @@
DOMAIN = "homeassistant_connect_zbt2"
NABU_CASA_FIRMWARE_RELEASES_URL = (
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest"
)
FIRMWARE = "firmware"

View File

@@ -1,9 +0,0 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -90,13 +90,6 @@
}
}
},
"entity": {
"switch": {
"beta_firmware": {
"name": "Beta firmware updates"
}
}
},
"exceptions": {
"device_disconnected": {
"message": "The device is not plugged in"

View File

@@ -1,54 +0,0 @@
"""Home Assistant Connect ZBT-2 switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantConnectZBT2ConfigEntry
from .const import DOMAIN, HARDWARE_NAME, SERIAL_NUMBER
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant Connect ZBT-2."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant Connect ZBT-2 beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
serial_number = self._config_entry.data[SERIAL_NUMBER]
self._attr_unique_id = f"{serial_number}_beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
name=f"{HARDWARE_NAME} ({serial_number})",
model=HARDWARE_NAME,
manufacturer="Nabu Casa",
serial_number=serial_number,
)

View File

@@ -4,6 +4,8 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -17,14 +19,22 @@ from homeassistant.components.homeassistant_hardware.util import (
ResetTarget,
)
from homeassistant.components.update import UpdateDeviceClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantConnectZBT2ConfigEntry
from .const import DOMAIN, FIRMWARE, FIRMWARE_VERSION, HARDWARE_NAME, SERIAL_NUMBER
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
HARDWARE_NAME,
NABU_CASA_FIRMWARE_RELEASES_URL,
SERIAL_NUMBER,
)
_LOGGER = logging.getLogger(__name__)
@@ -81,7 +91,8 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -100,7 +111,12 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=config_entry.data["device"],
config_entry=config_entry,
update_coordinator=config_entry.runtime_data.coordinator,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
entity_description=entity_description,
)
@@ -110,7 +126,11 @@ def _async_create_update_entity(
"""Replace the current entity when the firmware type changes."""
er.async_get(hass).async_remove(entity.entity_id)
async_add_entities(
[_async_create_update_entity(hass, config_entry, async_add_entities)]
[
_async_create_update_entity(
hass, config_entry, session, async_add_entities
)
]
)
entity.async_on_remove(
@@ -122,11 +142,14 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the firmware update config entry."""
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
session = async_get_clientsession(hass)
entity = _async_create_update_entity(
hass, config_entry, session, async_add_entities
)
async_add_entities([entity])
@@ -139,7 +162,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
config_entry: ConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

View File

@@ -7,7 +7,7 @@
"documentation": "https://www.home-assistant.io/integrations/homeassistant_hardware",
"integration_type": "system",
"requirements": [
"universal-silabs-flasher==0.0.37",
"ha-silabs-firmware-client==0.3.0"
"universal-silabs-flasher==0.0.35",
"ha-silabs-firmware-client==0.2.0"
]
}

View File

@@ -1,64 +0,0 @@
"""Home Assistant Hardware base beta firmware switch entity."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.helpers.restore_state import RestoreEntity
from .coordinator import FirmwareUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class BaseBetaFirmwareSwitch(SwitchEntity, RestoreEntity):
"""Base switch to enable beta firmware updates."""
_attr_has_entity_name = True
_attr_entity_category = EntityCategory.CONFIG
_attr_entity_registry_enabled_default = False
_attr_translation_key = "beta_firmware"
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: ConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
self._coordinator = coordinator
self._config_entry = config_entry
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added to hass."""
await super().async_added_to_hass()
# Restore the last state
last_state = await self.async_get_last_state()
if last_state is not None:
self._attr_is_on = last_state.state == "on"
else:
self._attr_is_on = False
# Apply the restored state to the coordinator
await self._update_coordinator_prerelease()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on beta firmware updates."""
self._attr_is_on = True
self.async_write_ha_state()
await self._update_coordinator_prerelease()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off beta firmware updates."""
self._attr_is_on = False
self.async_write_ha_state()
await self._update_coordinator_prerelease()
async def _update_coordinator_prerelease(self) -> None:
"""Update the coordinator with the current prerelease setting."""
self._coordinator.client.update_prerelease(bool(self._attr_is_on))
await self._coordinator.async_refresh()

View File

@@ -150,11 +150,6 @@ class BaseFirmwareUpdateEntity(
self._update_attributes()
# Fetch firmware info early to avoid prolonged "unknown" state when the device
# is initially set up
if self._latest_manifest is None:
await self.coordinator.async_request_refresh()
@property
def extra_restore_state_data(self) -> FirmwareUpdateExtraStoredData:
"""Return state data to be restored."""

View File

@@ -2,13 +2,9 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
import os.path
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.util import guess_firmware_info
from homeassistant.components.usb import (
USBDevice,
@@ -19,7 +15,6 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import (
@@ -29,7 +24,6 @@ from .const import (
FIRMWARE,
FIRMWARE_VERSION,
MANUFACTURER,
NABU_CASA_FIRMWARE_RELEASES_URL,
PID,
PRODUCT,
SERIAL_NUMBER,
@@ -38,16 +32,6 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
type HomeAssistantSkyConnectConfigEntry = ConfigEntry[HomeAssistantSkyConnectData]
@dataclass
class HomeAssistantSkyConnectData:
"""Runtime data definition."""
coordinator: FirmwareUpdateCoordinator
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@@ -81,9 +65,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant SkyConnect config entry."""
# Postpone loading the config entry if the device is missing
@@ -94,31 +76,18 @@ async def async_setup_entry(
translation_key="device_disconnected",
)
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantSkyConnectData(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
await hass.config_entries.async_forward_entry_setups(entry, ["update"])
return True
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True
async def async_migrate_entry(
hass: HomeAssistant, config_entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug(

View File

@@ -8,7 +8,7 @@ DOMAIN = "homeassistant_sky_connect"
DOCS_WEB_FLASHER_URL = "https://skyconnect.home-assistant.io/firmware-update/"
NABU_CASA_FIRMWARE_RELEASES_URL = (
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest"
)
FIRMWARE = "firmware"

View File

@@ -1,9 +0,0 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -90,13 +90,6 @@
}
}
},
"entity": {
"switch": {
"beta_firmware": {
"name": "Beta firmware updates"
}
}
},
"exceptions": {
"device_disconnected": {
"message": "The device is not plugged in"

View File

@@ -1,57 +0,0 @@
"""Home Assistant SkyConnect switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantSkyConnectConfigEntry
from .const import DOMAIN, PRODUCT, SERIAL_NUMBER, HardwareVariant
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantSkyConnectConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant SkyConnect."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant SkyConnect beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantSkyConnectConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
variant = HardwareVariant.from_usb_product_name(
self._config_entry.data[PRODUCT]
)
serial_number = self._config_entry.data[SERIAL_NUMBER]
self._attr_unique_id = f"{serial_number}_beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
name=f"{variant.full_name} ({serial_number[:8]})",
model=variant.full_name,
manufacturer="Nabu Casa",
serial_number=serial_number,
)

View File

@@ -4,6 +4,8 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -16,17 +18,19 @@ from homeassistant.components.homeassistant_hardware.util import (
FirmwareInfo,
)
from homeassistant.components.update import UpdateDeviceClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantSkyConnectConfigEntry
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
NABU_CASA_FIRMWARE_RELEASES_URL,
PRODUCT,
SERIAL_NUMBER,
HardwareVariant,
@@ -98,7 +102,8 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: HomeAssistantSkyConnectConfigEntry,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -117,7 +122,12 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=config_entry.data["device"],
config_entry=config_entry,
update_coordinator=config_entry.runtime_data.coordinator,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
entity_description=entity_description,
)
@@ -127,7 +137,11 @@ def _async_create_update_entity(
"""Replace the current entity when the firmware type changes."""
er.async_get(hass).async_remove(entity.entity_id)
async_add_entities(
[_async_create_update_entity(hass, config_entry, async_add_entities)]
[
_async_create_update_entity(
hass, config_entry, session, async_add_entities
)
]
)
entity.async_on_remove(
@@ -139,11 +153,14 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantSkyConnectConfigEntry,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the firmware update config entry."""
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
session = async_get_clientsession(hass)
entity = _async_create_update_entity(
hass, config_entry, session, async_add_entities
)
async_add_entities([entity])
@@ -157,7 +174,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: HomeAssistantSkyConnectConfigEntry,
config_entry: ConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

View File

@@ -2,13 +2,9 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
from homeassistant.components.hassio import get_os_info
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
check_multi_pan_addon,
)
@@ -20,34 +16,14 @@ from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.hassio import is_hassio
from .const import (
FIRMWARE,
FIRMWARE_VERSION,
NABU_CASA_FIRMWARE_RELEASES_URL,
RADIO_DEVICE,
ZHA_HW_DISCOVERY_DATA,
)
from .const import FIRMWARE, FIRMWARE_VERSION, RADIO_DEVICE, ZHA_HW_DISCOVERY_DATA
_LOGGER = logging.getLogger(__name__)
type HomeAssistantYellowConfigEntry = ConfigEntry[HomeAssistantYellowData]
@dataclass
class HomeAssistantYellowData:
"""Runtime data definition."""
coordinator: (
FirmwareUpdateCoordinator # Type from homeassistant_hardware.coordinator
)
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantYellowConfigEntry
) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant Yellow config entry."""
if not is_hassio(hass):
# Not running under supervisor, Home Assistant may have been migrated
@@ -80,31 +56,18 @@ async def async_setup_entry(
data=ZHA_HW_DISCOVERY_DATA,
)
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantYellowData(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
await hass.config_entries.async_forward_entry_setups(entry, ["update"])
return True
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantYellowConfigEntry
) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True
async def async_migrate_entry(
hass: HomeAssistant, config_entry: HomeAssistantYellowConfigEntry
) -> bool:
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug(

View File

@@ -22,5 +22,5 @@ FIRMWARE_VERSION = "firmware_version"
ZHA_DOMAIN = "zha"
NABU_CASA_FIRMWARE_RELEASES_URL = (
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest"
)

View File

@@ -1,9 +0,0 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -1,10 +1,5 @@
{
"entity": {
"switch": {
"beta_firmware": {
"name": "Radio beta firmware updates"
}
},
"update": {
"radio_firmware": {
"name": "Radio firmware"

View File

@@ -1,50 +0,0 @@
"""Home Assistant Yellow switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantYellowConfigEntry
from .const import DOMAIN, MANUFACTURER, MODEL
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantYellowConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant Yellow."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant Yellow beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantYellowConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
self._attr_unique_id = "beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, "yellow")},
name=MODEL,
model=MODEL,
manufacturer=MANUFACTURER,
)

View File

@@ -4,6 +4,8 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -17,14 +19,23 @@ from homeassistant.components.homeassistant_hardware.util import (
ResetTarget,
)
from homeassistant.components.update import UpdateDeviceClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantYellowConfigEntry
from .const import DOMAIN, FIRMWARE, FIRMWARE_VERSION, MANUFACTURER, MODEL, RADIO_DEVICE
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
MANUFACTURER,
MODEL,
NABU_CASA_FIRMWARE_RELEASES_URL,
RADIO_DEVICE,
)
_LOGGER = logging.getLogger(__name__)
@@ -97,7 +108,8 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: HomeAssistantYellowConfigEntry,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -116,7 +128,12 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=RADIO_DEVICE,
config_entry=config_entry,
update_coordinator=config_entry.runtime_data.coordinator,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
entity_description=entity_description,
)
@@ -126,7 +143,11 @@ def _async_create_update_entity(
"""Replace the current entity when the firmware type changes."""
er.async_get(hass).async_remove(entity.entity_id)
async_add_entities(
[_async_create_update_entity(hass, config_entry, async_add_entities)]
[
_async_create_update_entity(
hass, config_entry, session, async_add_entities
)
]
)
entity.async_on_remove(
@@ -138,11 +159,14 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantYellowConfigEntry,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the firmware update config entry."""
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
session = async_get_clientsession(hass)
entity = _async_create_update_entity(
hass, config_entry, session, async_add_entities
)
async_add_entities([entity])
@@ -155,7 +179,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: HomeAssistantYellowConfigEntry,
config_entry: ConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

View File

@@ -9,6 +9,7 @@ from typing import Any
from chip.clusters import Objects as clusters
from chip.clusters.Objects import ClusterCommand, NullValue
from matter_server.client.models import device_types
from matter_server.common.errors import MatterError
from homeassistant.components.switch import (
SwitchDeviceClass,
@@ -18,6 +19,7 @@ from homeassistant.components.switch import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .entity import MatterEntity, MatterEntityDescription
@@ -54,15 +56,21 @@ class MatterSwitch(MatterEntity, SwitchEntity):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn switch on."""
await self.send_device_command(
clusters.OnOff.Commands.On(),
)
try:
await self.send_device_command(
clusters.OnOff.Commands.On(),
)
except MatterError as err:
raise HomeAssistantError(f"Failed to set value: {err}") from err
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn switch off."""
await self.send_device_command(
clusters.OnOff.Commands.Off(),
)
try:
await self.send_device_command(
clusters.OnOff.Commands.Off(),
)
except MatterError as err:
raise HomeAssistantError(f"Failed to set value: {err}") from err
@callback
def _update_from_device(self) -> None:
@@ -83,18 +91,24 @@ class MatterGenericCommandSwitch(MatterSwitch):
"""Turn switch on."""
if self.entity_description.on_command:
# custom command defined to set the new value
await self.send_device_command(
self.entity_description.on_command(),
self.entity_description.command_timeout,
)
try:
await self.send_device_command(
self.entity_description.on_command(),
self.entity_description.command_timeout,
)
except MatterError as err:
raise HomeAssistantError(f"Failed to set value: {err}") from err
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn switch off."""
if self.entity_description.off_command:
await self.send_device_command(
self.entity_description.off_command(),
self.entity_description.command_timeout,
)
try:
await self.send_device_command(
self.entity_description.off_command(),
self.entity_description.command_timeout,
)
except MatterError as err:
raise HomeAssistantError(f"Failed to set value: {err}") from err
@callback
def _update_from_device(self) -> None:
@@ -111,13 +125,16 @@ class MatterGenericCommandSwitch(MatterSwitch):
**kwargs: Any,
) -> None:
"""Send device command with timeout."""
await self.matter_client.send_device_command(
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=command,
timed_request_timeout_ms=command_timeout,
**kwargs,
)
try:
await self.matter_client.send_device_command(
node_id=self._endpoint.node.node_id,
endpoint_id=self._endpoint.endpoint_id,
command=command,
timed_request_timeout_ms=command_timeout,
**kwargs,
)
except MatterError as err:
raise HomeAssistantError(f"Failed to set value: {err}") from err
@dataclass(frozen=True, kw_only=True)

View File

@@ -0,0 +1,76 @@
"""Support for Neato botvac connected vacuum cleaners."""
import logging
import aiohttp
from pybotvac import Account
from pybotvac.exceptions import NeatoException
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_entry_oauth2_flow
from . import api
from .const import NEATO_DOMAIN, NEATO_LOGIN
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BUTTON,
Platform.CAMERA,
Platform.SENSOR,
Platform.SWITCH,
Platform.VACUUM,
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up config entry."""
hass.data.setdefault(NEATO_DOMAIN, {})
if CONF_TOKEN not in entry.data:
raise ConfigEntryAuthFailed
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
try:
await session.async_ensure_token_valid()
except aiohttp.ClientResponseError as ex:
_LOGGER.debug("API error: %s (%s)", ex.code, ex.message)
if ex.code in (401, 403):
raise ConfigEntryAuthFailed("Token not valid, trigger renewal") from ex
raise ConfigEntryNotReady from ex
neato_session = api.ConfigEntryAuth(hass, entry, implementation)
hass.data[NEATO_DOMAIN][entry.entry_id] = neato_session
hub = NeatoHub(hass, Account(neato_session))
await hub.async_update_entry_unique_id(entry)
try:
await hass.async_add_executor_job(hub.update_robots)
except NeatoException as ex:
_LOGGER.debug("Failed to connect to Neato API")
raise ConfigEntryNotReady from ex
hass.data[NEATO_LOGIN] = hub
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[NEATO_DOMAIN].pop(entry.entry_id)
return unload_ok

View File

@@ -0,0 +1,58 @@
"""API for Neato Botvac bound to Home Assistant OAuth."""
from __future__ import annotations
from asyncio import run_coroutine_threadsafe
from typing import Any
import pybotvac
from homeassistant import config_entries, core
from homeassistant.components.application_credentials import AuthImplementation
from homeassistant.helpers import config_entry_oauth2_flow
class ConfigEntryAuth(pybotvac.OAuthSession): # type: ignore[misc]
"""Provide Neato Botvac authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
) -> None:
"""Initialize Neato Botvac Auth."""
self.hass = hass
self.session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
super().__init__(self.session.token, vendor=pybotvac.Neato())
def refresh_tokens(self) -> str:
"""Refresh and return new Neato Botvac tokens."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop
).result()
return self.session.token["access_token"] # type: ignore[no-any-return]
class NeatoImplementation(AuthImplementation):
"""Neato implementation of LocalOAuth2Implementation.
We need this class because we have to add client_secret
and scope to the authorization request.
"""
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {"client_secret": self.client_secret}
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Generate a url for the user to authorize.
We must make sure that the plus signs are not encoded.
"""
url = await super().async_generate_authorize_url(flow_id)
return f"{url}&scope=public_profile+control_robots+maps"

View File

@@ -0,0 +1,28 @@
"""Application credentials platform for neato."""
from pybotvac import Neato
from homeassistant.components.application_credentials import (
AuthorizationServer,
ClientCredential,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from . import api
async def async_get_auth_implementation(
hass: HomeAssistant, auth_domain: str, credential: ClientCredential
) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
"""Return auth implementation for a custom auth implementation."""
vendor = Neato()
return api.NeatoImplementation(
hass,
auth_domain,
credential,
AuthorizationServer(
authorize_url=vendor.auth_endpoint,
token_url=vendor.token_endpoint,
),
)

View File

@@ -0,0 +1,44 @@
"""Support for Neato buttons."""
from __future__ import annotations
from pybotvac import Robot
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_ROBOTS
from .entity import NeatoEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato button from config entry."""
entities = [NeatoDismissAlertButton(robot) for robot in hass.data[NEATO_ROBOTS]]
async_add_entities(entities, True)
class NeatoDismissAlertButton(NeatoEntity, ButtonEntity):
"""Representation of a dismiss_alert button entity."""
_attr_translation_key = "dismiss_alert"
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
robot: Robot,
) -> None:
"""Initialize a dismiss_alert Neato button entity."""
super().__init__(robot)
self._attr_unique_id = f"{robot.serial}_dismiss_alert"
async def async_press(self) -> None:
"""Press the button."""
await self.hass.async_add_executor_job(self.robot.dismiss_current_alert)

View File

@@ -0,0 +1,130 @@
"""Support for loading picture from Neato."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from urllib3.response import HTTPResponse
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_MAP_DATA, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
ATTR_GENERATED_AT = "generated_at"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato camera with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
dev = [
NeatoCleaningMap(neato, robot, mapdata)
for robot in hass.data[NEATO_ROBOTS]
if "maps" in robot.traits
]
if not dev:
return
_LOGGER.debug("Adding robots for cleaning maps %s", dev)
async_add_entities(dev, True)
class NeatoCleaningMap(NeatoEntity, Camera):
"""Neato cleaning map for last clean."""
_attr_translation_key = "cleaning_map"
def __init__(
self, neato: NeatoHub, robot: Robot, mapdata: dict[str, Any] | None
) -> None:
"""Initialize Neato cleaning map."""
super().__init__(robot)
Camera.__init__(self)
self.neato = neato
self._mapdata = mapdata
self._available = neato is not None
self._robot_serial: str = self.robot.serial
self._attr_unique_id = self.robot.serial
self._generated_at: str | None = None
self._image_url: str | None = None
self._image: bytes | None = None
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
self.update()
return self._image
def update(self) -> None:
"""Check the contents of the map list."""
_LOGGER.debug("Running camera update for '%s'", self.entity_id)
try:
self.neato.update_robots()
except NeatoRobotException as ex:
if self._available: # Print only once when available
_LOGGER.error(
"Neato camera connection error for '%s': %s", self.entity_id, ex
)
self._image = None
self._image_url = None
self._available = False
return
if self._mapdata:
map_data: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
if (image_url := map_data["url"]) == self._image_url:
_LOGGER.debug(
"The map image_url for '%s' is the same as old", self.entity_id
)
return
try:
image: HTTPResponse = self.neato.download_map(image_url)
except NeatoRobotException as ex:
if self._available: # Print only once when available
_LOGGER.error(
"Neato camera connection error for '%s': %s", self.entity_id, ex
)
self._image = None
self._image_url = None
self._available = False
return
self._image = image.read()
self._image_url = image_url
self._generated_at = map_data.get("generated_at")
self._available = True
@property
def available(self) -> bool:
"""Return if the robot is available."""
return self._available
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the vacuum cleaner."""
data: dict[str, Any] = {}
if self._generated_at is not None:
data[ATTR_GENERATED_AT] = self._generated_at
return data

View File

@@ -0,0 +1,64 @@
"""Config flow for Neato Botvac."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from .const import NEATO_DOMAIN
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=NEATO_DOMAIN
):
"""Config flow to handle Neato Botvac OAuth2 authentication."""
DOMAIN = NEATO_DOMAIN
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Create an entry for the flow."""
current_entries = self._async_current_entries()
if self.source != SOURCE_REAUTH and current_entries:
# Already configured
return self.async_abort(reason="already_configured")
return await super().async_step_user(user_input=user_input)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon migration of old entries."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm reauth upon migration of old entries."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
"""Create an entry for the flow. Update an entry if one already exist."""
current_entries = self._async_current_entries()
if self.source == SOURCE_REAUTH and current_entries:
# Update entry
self.hass.config_entries.async_update_entry(
current_entries[0], title=self.flow_impl.name, data=data
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(current_entries[0].entry_id)
)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self.flow_impl.name, data=data)

View File

@@ -0,0 +1,150 @@
"""Constants for Neato integration."""
NEATO_DOMAIN = "neato"
CONF_VENDOR = "vendor"
NEATO_LOGIN = "neato_login"
NEATO_MAP_DATA = "neato_map_data"
NEATO_PERSISTENT_MAPS = "neato_persistent_maps"
NEATO_ROBOTS = "neato_robots"
SCAN_INTERVAL_MINUTES = 1
MODE = {1: "Eco", 2: "Turbo"}
ACTION = {
0: "Invalid",
1: "House Cleaning",
2: "Spot Cleaning",
3: "Manual Cleaning",
4: "Docking",
5: "User Menu Active",
6: "Suspended Cleaning",
7: "Updating",
8: "Copying logs",
9: "Recovering Location",
10: "IEC test",
11: "Map cleaning",
12: "Exploring map (creating a persistent map)",
13: "Acquiring Persistent Map IDs",
14: "Creating & Uploading Map",
15: "Suspended Exploration",
}
ERRORS = {
"ui_error_battery_battundervoltlithiumsafety": "Replace battery",
"ui_error_battery_critical": "Replace battery",
"ui_error_battery_invalidsensor": "Replace battery",
"ui_error_battery_lithiumadapterfailure": "Replace battery",
"ui_error_battery_mismatch": "Replace battery",
"ui_error_battery_nothermistor": "Replace battery",
"ui_error_battery_overtemp": "Replace battery",
"ui_error_battery_overvolt": "Replace battery",
"ui_error_battery_undercurrent": "Replace battery",
"ui_error_battery_undertemp": "Replace battery",
"ui_error_battery_undervolt": "Replace battery",
"ui_error_battery_unplugged": "Replace battery",
"ui_error_brush_stuck": "Brush stuck",
"ui_error_brush_overloaded": "Brush overloaded",
"ui_error_bumper_stuck": "Bumper stuck",
"ui_error_check_battery_switch": "Check battery",
"ui_error_corrupt_scb": "Call customer service corrupt board",
"ui_error_deck_debris": "Deck debris",
"ui_error_dflt_app": "Check Neato app",
"ui_error_disconnect_chrg_cable": "Disconnected charge cable",
"ui_error_disconnect_usb_cable": "Disconnected USB cable",
"ui_error_dust_bin_missing": "Dust bin missing",
"ui_error_dust_bin_full": "Dust bin full",
"ui_error_dust_bin_emptied": "Dust bin emptied",
"ui_error_hardware_failure": "Hardware failure",
"ui_error_ldrop_stuck": "Clear my path",
"ui_error_lds_jammed": "Clear my path",
"ui_error_lds_bad_packets": "Check Neato app",
"ui_error_lds_disconnected": "Check Neato app",
"ui_error_lds_missed_packets": "Check Neato app",
"ui_error_lwheel_stuck": "Clear my path",
"ui_error_navigation_backdrop_frontbump": "Clear my path",
"ui_error_navigation_backdrop_leftbump": "Clear my path",
"ui_error_navigation_backdrop_wheelextended": "Clear my path",
"ui_error_navigation_noprogress": "Clear my path",
"ui_error_navigation_origin_unclean": "Clear my path",
"ui_error_navigation_pathproblems": "Cannot return to base",
"ui_error_navigation_pinkycommsfail": "Clear my path",
"ui_error_navigation_falling": "Clear my path",
"ui_error_navigation_noexitstogo": "Clear my path",
"ui_error_navigation_nomotioncommands": "Clear my path",
"ui_error_navigation_rightdrop_leftbump": "Clear my path",
"ui_error_navigation_undockingfailed": "Clear my path",
"ui_error_picked_up": "Picked up",
"ui_error_qa_fail": "Check Neato app",
"ui_error_rdrop_stuck": "Clear my path",
"ui_error_reconnect_failed": "Reconnect failed",
"ui_error_rwheel_stuck": "Clear my path",
"ui_error_stuck": "Stuck!",
"ui_error_unable_to_return_to_base": "Unable to return to base",
"ui_error_unable_to_see": "Clean vacuum sensors",
"ui_error_vacuum_slip": "Clear my path",
"ui_error_vacuum_stuck": "Clear my path",
"ui_error_warning": "Error check app",
"batt_base_connect_fail": "Battery failed to connect to base",
"batt_base_no_power": "Battery base has no power",
"batt_low": "Battery low",
"batt_on_base": "Battery on base",
"clean_tilt_on_start": "Clean the tilt on start",
"dustbin_full": "Dust bin full",
"dustbin_missing": "Dust bin missing",
"gen_picked_up": "Picked up",
"hw_fail": "Hardware failure",
"hw_tof_sensor_sensor": "Hardware sensor disconnected",
"lds_bad_packets": "Bad packets",
"lds_deck_debris": "Debris on deck",
"lds_disconnected": "Disconnected",
"lds_jammed": "Jammed",
"lds_missed_packets": "Missed packets",
"maint_brush_stuck": "Brush stuck",
"maint_brush_overload": "Brush overloaded",
"maint_bumper_stuck": "Bumper stuck",
"maint_customer_support_qa": "Contact customer support",
"maint_vacuum_stuck": "Vacuum is stuck",
"maint_vacuum_slip": "Vacuum is stuck",
"maint_left_drop_stuck": "Vacuum is stuck",
"maint_left_wheel_stuck": "Vacuum is stuck",
"maint_right_drop_stuck": "Vacuum is stuck",
"maint_right_wheel_stuck": "Vacuum is stuck",
"not_on_charge_base": "Not on the charge base",
"nav_robot_falling": "Clear my path",
"nav_no_path": "Clear my path",
"nav_path_problem": "Clear my path",
"nav_backdrop_frontbump": "Clear my path",
"nav_backdrop_leftbump": "Clear my path",
"nav_backdrop_wheelextended": "Clear my path",
"nav_floorplan_zone_path_blocked": "Clear my path",
"nav_mag_sensor": "Clear my path",
"nav_no_exit": "Clear my path",
"nav_no_movement": "Clear my path",
"nav_rightdrop_leftbump": "Clear my path",
"nav_undocking_failed": "Clear my path",
}
ALERTS = {
"ui_alert_dust_bin_full": "Please empty dust bin",
"ui_alert_recovering_location": "Returning to start",
"ui_alert_battery_chargebasecommerr": "Battery error",
"ui_alert_busy_charging": "Busy charging",
"ui_alert_charging_base": "Base charging",
"ui_alert_charging_power": "Charging power",
"ui_alert_connect_chrg_cable": "Connect charge cable",
"ui_alert_info_thank_you": "Thank you",
"ui_alert_invalid": "Invalid check app",
"ui_alert_old_error": "Old error",
"ui_alert_swupdate_fail": "Update failed",
"dustbin_full": "Please empty dust bin",
"maint_brush_change": "Change the brush",
"maint_filter_change": "Change the filter",
"clean_completed_to_start": "Cleaning completed",
"nav_floorplan_not_created": "No floorplan found",
"nav_floorplan_load_fail": "Failed to load floorplan",
"nav_floorplan_localization_fail": "Failed to load floorplan",
"clean_incomplete_to_start": "Cleaning incomplete",
"log_upload_failed": "Logs failed to upload",
}

View File

@@ -0,0 +1,24 @@
"""Base entity for Neato."""
from __future__ import annotations
from pybotvac import Robot
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import NEATO_DOMAIN
class NeatoEntity(Entity):
"""Base Neato entity."""
_attr_has_entity_name = True
def __init__(self, robot: Robot) -> None:
"""Initialize Neato entity."""
self.robot = robot
self._attr_device_info: DeviceInfo = DeviceInfo(
identifiers={(NEATO_DOMAIN, self.robot.serial)},
name=self.robot.name,
)

View File

@@ -0,0 +1,50 @@
"""Support for Neato botvac connected vacuum cleaners."""
from datetime import timedelta
import logging
from pybotvac import Account
from urllib3.response import HTTPResponse
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.util import Throttle
from .const import NEATO_MAP_DATA, NEATO_PERSISTENT_MAPS, NEATO_ROBOTS
_LOGGER = logging.getLogger(__name__)
class NeatoHub:
"""A My Neato hub wrapper class."""
def __init__(self, hass: HomeAssistant, neato: Account) -> None:
"""Initialize the Neato hub."""
self._hass = hass
self.my_neato: Account = neato
@Throttle(timedelta(minutes=1))
def update_robots(self) -> None:
"""Update the robot states."""
_LOGGER.debug("Running HUB.update_robots %s", self._hass.data.get(NEATO_ROBOTS))
self._hass.data[NEATO_ROBOTS] = self.my_neato.robots
self._hass.data[NEATO_PERSISTENT_MAPS] = self.my_neato.persistent_maps
self._hass.data[NEATO_MAP_DATA] = self.my_neato.maps
def download_map(self, url: str) -> HTTPResponse:
"""Download a new map image."""
map_image_data: HTTPResponse = self.my_neato.get_map_image(url)
return map_image_data
async def async_update_entry_unique_id(self, entry: ConfigEntry) -> str:
"""Update entry for unique_id."""
await self._hass.async_add_executor_job(self.my_neato.refresh_userdata)
unique_id: str = self.my_neato.unique_id
if entry.unique_id == unique_id:
return unique_id
_LOGGER.debug("Updating user unique_id for previous config entry")
self._hass.config_entries.async_update_entry(entry, unique_id=unique_id)
return unique_id

View File

@@ -0,0 +1,7 @@
{
"services": {
"custom_cleaning": {
"service": "mdi:broom"
}
}
}

View File

@@ -0,0 +1,11 @@
{
"domain": "neato",
"name": "Neato Botvac",
"codeowners": [],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/neato",
"iot_class": "cloud_polling",
"loggers": ["pybotvac"],
"requirements": ["pybotvac==0.0.28"]
}

View File

@@ -0,0 +1,81 @@
"""Support for Neato sensors."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
BATTERY = "Battery"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Neato sensor using config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
dev = [NeatoSensor(neato, robot) for robot in hass.data[NEATO_ROBOTS]]
if not dev:
return
_LOGGER.debug("Adding robots for sensors %s", dev)
async_add_entities(dev, True)
class NeatoSensor(NeatoEntity, SensorEntity):
"""Neato sensor."""
_attr_device_class = SensorDeviceClass.BATTERY
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_native_unit_of_measurement = PERCENTAGE
_attr_available: bool = False
def __init__(self, neato: NeatoHub, robot: Robot) -> None:
"""Initialize Neato sensor."""
super().__init__(robot)
self._robot_serial: str = self.robot.serial
self._attr_unique_id = self.robot.serial
self._state: dict[str, Any] | None = None
def update(self) -> None:
"""Update Neato Sensor."""
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available:
_LOGGER.error(
"Neato sensor connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
@property
def native_value(self) -> str | None:
"""Return the state."""
if self._state is not None:
return str(self._state["details"]["charge"])
return None

View File

@@ -0,0 +1,32 @@
custom_cleaning:
target:
entity:
integration: neato
domain: vacuum
fields:
mode:
default: 2
selector:
number:
min: 1
max: 2
mode: box
navigation:
default: 1
selector:
number:
min: 1
max: 3
mode: box
category:
default: 4
selector:
number:
min: 2
max: 4
step: 2
mode: box
zone:
example: "Kitchen"
selector:
text:

View File

@@ -0,0 +1,73 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"step": {
"pick_implementation": {
"data": {
"implementation": "[%key:common::config_flow::data::implementation%]"
},
"data_description": {
"implementation": "[%key:common::config_flow::description::implementation%]"
},
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::description::confirm_setup%]"
}
}
},
"entity": {
"button": {
"dismiss_alert": {
"name": "Dismiss alert"
}
},
"camera": {
"cleaning_map": {
"name": "Cleaning map"
}
},
"switch": {
"schedule": {
"name": "Schedule"
}
}
},
"services": {
"custom_cleaning": {
"description": "Starts a custom cleaning of your house.",
"fields": {
"category": {
"description": "Whether to use a persistent map or not for cleaning (i.e. No go lines): 2 for no map, 4 for map. Default to using map if not set (and fallback to no map if no map is found).",
"name": "Use cleaning map"
},
"mode": {
"description": "Sets the cleaning mode: 1 for eco and 2 for turbo. Defaults to turbo if not set.",
"name": "Cleaning mode"
},
"navigation": {
"description": "Sets the navigation mode: 1 for normal, 2 for extra care, 3 for deep. Defaults to normal if not set.",
"name": "Navigation mode"
},
"zone": {
"description": "Name of the zone to clean (only supported on the Botvac D7). Defaults to no zone i.e. complete house cleanup.",
"name": "Zone"
}
},
"name": "Custom cleaning"
}
}
}

View File

@@ -0,0 +1,118 @@
"""Support for Neato Connected Vacuums switches."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_OFF, STATE_ON, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
SWITCH_TYPE_SCHEDULE = "schedule"
SWITCH_TYPES = {SWITCH_TYPE_SCHEDULE: ["Schedule"]}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato switch with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
dev = [
NeatoConnectedSwitch(neato, robot, type_name)
for robot in hass.data[NEATO_ROBOTS]
for type_name in SWITCH_TYPES
]
if not dev:
return
_LOGGER.debug("Adding switches %s", dev)
async_add_entities(dev, True)
class NeatoConnectedSwitch(NeatoEntity, SwitchEntity):
"""Neato Connected Switches."""
_attr_translation_key = "schedule"
_attr_available = False
_attr_entity_category = EntityCategory.CONFIG
def __init__(self, neato: NeatoHub, robot: Robot, switch_type: str) -> None:
"""Initialize the Neato Connected switches."""
super().__init__(robot)
self.type = switch_type
self._state: dict[str, Any] | None = None
self._schedule_state: str | None = None
self._clean_state = None
self._attr_unique_id = self.robot.serial
def update(self) -> None:
"""Update the states of Neato switches."""
_LOGGER.debug("Running Neato switch update for '%s'", self.entity_id)
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available: # Print only once when available
_LOGGER.error(
"Neato switch connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
if self.type == SWITCH_TYPE_SCHEDULE:
_LOGGER.debug("State: %s", self._state)
if self._state is not None and self._state["details"]["isScheduleEnabled"]:
self._schedule_state = STATE_ON
else:
self._schedule_state = STATE_OFF
_LOGGER.debug(
"Schedule state for '%s': %s", self.entity_id, self._schedule_state
)
@property
def is_on(self) -> bool:
"""Return true if switch is on."""
return bool(
self.type == SWITCH_TYPE_SCHEDULE and self._schedule_state == STATE_ON
)
def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.enable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato switch connection error '%s': %s", self.entity_id, ex
)
def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.disable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato switch connection error '%s': %s", self.entity_id, ex
)

View File

@@ -0,0 +1,388 @@
"""Support for Neato Connected Vacuums."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac import Robot
from pybotvac.exceptions import NeatoRobotException
import voluptuous as vol
from homeassistant.components.vacuum import (
ATTR_STATUS,
StateVacuumEntity,
VacuumActivity,
VacuumEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_MODE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
ACTION,
ALERTS,
ERRORS,
MODE,
NEATO_LOGIN,
NEATO_MAP_DATA,
NEATO_PERSISTENT_MAPS,
NEATO_ROBOTS,
SCAN_INTERVAL_MINUTES,
)
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
ATTR_CLEAN_START = "clean_start"
ATTR_CLEAN_STOP = "clean_stop"
ATTR_CLEAN_AREA = "clean_area"
ATTR_CLEAN_BATTERY_START = "battery_level_at_clean_start"
ATTR_CLEAN_BATTERY_END = "battery_level_at_clean_end"
ATTR_CLEAN_SUSP_COUNT = "clean_suspension_count"
ATTR_CLEAN_SUSP_TIME = "clean_suspension_time"
ATTR_CLEAN_PAUSE_TIME = "clean_pause_time"
ATTR_CLEAN_ERROR_TIME = "clean_error_time"
ATTR_LAUNCHED_FROM = "launched_from"
ATTR_NAVIGATION = "navigation"
ATTR_CATEGORY = "category"
ATTR_ZONE = "zone"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato vacuum with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
persistent_maps: dict[str, Any] | None = hass.data.get(NEATO_PERSISTENT_MAPS)
dev = [
NeatoConnectedVacuum(neato, robot, mapdata, persistent_maps)
for robot in hass.data[NEATO_ROBOTS]
]
if not dev:
return
_LOGGER.debug("Adding vacuums %s", dev)
async_add_entities(dev, True)
platform = entity_platform.async_get_current_platform()
assert platform is not None
platform.async_register_entity_service(
"custom_cleaning",
{
vol.Optional(ATTR_MODE, default=2): cv.positive_int,
vol.Optional(ATTR_NAVIGATION, default=1): cv.positive_int,
vol.Optional(ATTR_CATEGORY, default=4): cv.positive_int,
vol.Optional(ATTR_ZONE): cv.string,
},
"neato_custom_cleaning",
)
class NeatoConnectedVacuum(NeatoEntity, StateVacuumEntity):
"""Representation of a Neato Connected Vacuum."""
_attr_supported_features = (
VacuumEntityFeature.BATTERY
| VacuumEntityFeature.PAUSE
| VacuumEntityFeature.RETURN_HOME
| VacuumEntityFeature.STOP
| VacuumEntityFeature.START
| VacuumEntityFeature.CLEAN_SPOT
| VacuumEntityFeature.STATE
| VacuumEntityFeature.MAP
| VacuumEntityFeature.LOCATE
)
_attr_name = None
def __init__(
self,
neato: NeatoHub,
robot: Robot,
mapdata: dict[str, Any] | None,
persistent_maps: dict[str, Any] | None,
) -> None:
"""Initialize the Neato Connected Vacuum."""
super().__init__(robot)
self._attr_available: bool = neato is not None
self._mapdata = mapdata
self._robot_has_map: bool = self.robot.has_persistent_maps
self._robot_maps = persistent_maps
self._robot_serial: str = self.robot.serial
self._attr_unique_id: str = self.robot.serial
self._status_state: str | None = None
self._state: dict[str, Any] | None = None
self._clean_time_start: str | None = None
self._clean_time_stop: str | None = None
self._clean_area: float | None = None
self._clean_battery_start: int | None = None
self._clean_battery_end: int | None = None
self._clean_susp_charge_count: int | None = None
self._clean_susp_time: int | None = None
self._clean_pause_time: int | None = None
self._clean_error_time: int | None = None
self._launched_from: str | None = None
self._robot_boundaries: list = []
self._robot_stats: dict[str, Any] | None = None
def update(self) -> None:
"""Update the states of Neato Vacuums."""
_LOGGER.debug("Running Neato Vacuums update for '%s'", self.entity_id)
try:
if self._robot_stats is None:
self._robot_stats = self.robot.get_general_info().json().get("data")
except NeatoRobotException:
_LOGGER.warning("Couldn't fetch robot information of %s", self.entity_id)
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available: # print only once when available
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
if self._state is None:
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
if "alert" in self._state:
robot_alert = ALERTS.get(self._state["alert"])
else:
robot_alert = None
if self._state["state"] == 1:
if self._state["details"]["isCharging"]:
self._attr_activity = VacuumActivity.DOCKED
self._status_state = "Charging"
elif (
self._state["details"]["isDocked"]
and not self._state["details"]["isCharging"]
):
self._attr_activity = VacuumActivity.DOCKED
self._status_state = "Docked"
else:
self._attr_activity = VacuumActivity.IDLE
self._status_state = "Stopped"
if robot_alert is not None:
self._status_state = robot_alert
elif self._state["state"] == 2:
if robot_alert is None:
self._attr_activity = VacuumActivity.CLEANING
self._status_state = (
f"{MODE.get(self._state['cleaning']['mode'])} "
f"{ACTION.get(self._state['action'])}"
)
if (
"boundary" in self._state["cleaning"]
and "name" in self._state["cleaning"]["boundary"]
):
self._status_state += (
f" {self._state['cleaning']['boundary']['name']}"
)
else:
self._status_state = robot_alert
elif self._state["state"] == 3:
self._attr_activity = VacuumActivity.PAUSED
self._status_state = "Paused"
elif self._state["state"] == 4:
self._attr_activity = VacuumActivity.ERROR
self._status_state = ERRORS.get(self._state["error"])
self._attr_battery_level = self._state["details"]["charge"]
if self._mapdata is None or not self._mapdata.get(self._robot_serial, {}).get(
"maps", []
):
return
mapdata: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
self._clean_time_start = mapdata["start_at"]
self._clean_time_stop = mapdata["end_at"]
self._clean_area = mapdata["cleaned_area"]
self._clean_susp_charge_count = mapdata["suspended_cleaning_charging_count"]
self._clean_susp_time = mapdata["time_in_suspended_cleaning"]
self._clean_pause_time = mapdata["time_in_pause"]
self._clean_error_time = mapdata["time_in_error"]
self._clean_battery_start = mapdata["run_charge_at_start"]
self._clean_battery_end = mapdata["run_charge_at_end"]
self._launched_from = mapdata["launched_from"]
if (
self._robot_has_map
and self._state
and self._state["availableServices"]["maps"] != "basic-1"
and self._robot_maps
):
allmaps: dict = self._robot_maps[self._robot_serial]
_LOGGER.debug(
"Found the following maps for '%s': %s", self.entity_id, allmaps
)
self._robot_boundaries = [] # Reset boundaries before refreshing boundaries
for maps in allmaps:
try:
robot_boundaries = self.robot.get_map_boundaries(maps["id"]).json()
except NeatoRobotException as ex:
_LOGGER.error(
"Could not fetch map boundaries for '%s': %s",
self.entity_id,
ex,
)
return
_LOGGER.debug(
"Boundaries for robot '%s' in map '%s': %s",
self.entity_id,
maps["name"],
robot_boundaries,
)
if "boundaries" in robot_boundaries["data"]:
self._robot_boundaries += robot_boundaries["data"]["boundaries"]
_LOGGER.debug(
"List of boundaries for '%s': %s",
self.entity_id,
self._robot_boundaries,
)
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the vacuum cleaner."""
data: dict[str, Any] = {}
if self._status_state is not None:
data[ATTR_STATUS] = self._status_state
if self._clean_time_start is not None:
data[ATTR_CLEAN_START] = self._clean_time_start
if self._clean_time_stop is not None:
data[ATTR_CLEAN_STOP] = self._clean_time_stop
if self._clean_area is not None:
data[ATTR_CLEAN_AREA] = self._clean_area
if self._clean_susp_charge_count is not None:
data[ATTR_CLEAN_SUSP_COUNT] = self._clean_susp_charge_count
if self._clean_susp_time is not None:
data[ATTR_CLEAN_SUSP_TIME] = self._clean_susp_time
if self._clean_pause_time is not None:
data[ATTR_CLEAN_PAUSE_TIME] = self._clean_pause_time
if self._clean_error_time is not None:
data[ATTR_CLEAN_ERROR_TIME] = self._clean_error_time
if self._clean_battery_start is not None:
data[ATTR_CLEAN_BATTERY_START] = self._clean_battery_start
if self._clean_battery_end is not None:
data[ATTR_CLEAN_BATTERY_END] = self._clean_battery_end
if self._launched_from is not None:
data[ATTR_LAUNCHED_FROM] = self._launched_from
return data
@property
def device_info(self) -> DeviceInfo:
"""Device info for neato robot."""
device_info = self._attr_device_info
if self._robot_stats:
device_info["manufacturer"] = self._robot_stats["battery"]["vendor"]
device_info["model"] = self._robot_stats["model"]
device_info["sw_version"] = self._robot_stats["firmware"]
return device_info
def start(self) -> None:
"""Start cleaning or resume cleaning."""
if self._state:
try:
if self._state["state"] == 1:
self.robot.start_cleaning()
elif self._state["state"] == 3:
self.robot.resume_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def pause(self) -> None:
"""Pause the vacuum."""
try:
self.robot.pause_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def return_to_base(self, **kwargs: Any) -> None:
"""Set the vacuum cleaner to return to the dock."""
try:
if self._attr_activity == VacuumActivity.CLEANING:
self.robot.pause_cleaning()
self._attr_activity = VacuumActivity.RETURNING
self.robot.send_to_base()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def stop(self, **kwargs: Any) -> None:
"""Stop the vacuum cleaner."""
try:
self.robot.stop_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def locate(self, **kwargs: Any) -> None:
"""Locate the robot by making it emit a sound."""
try:
self.robot.locate()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def clean_spot(self, **kwargs: Any) -> None:
"""Run a spot cleaning starting from the base."""
try:
self.robot.start_spot_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def neato_custom_cleaning(
self, mode: str, navigation: str, category: str, zone: str | None = None
) -> None:
"""Zone cleaning service call."""
boundary_id = None
if zone is not None:
for boundary in self._robot_boundaries:
if zone in boundary["name"]:
boundary_id = boundary["id"]
if boundary_id is None:
_LOGGER.error(
"Zone '%s' was not found for the robot '%s'", zone, self.entity_id
)
return
_LOGGER.debug(
"Start cleaning zone '%s' with robot %s", zone, self.entity_id
)
self._attr_activity = VacuumActivity.CLEANING
try:
self.robot.start_cleaning(mode, navigation, category, boundary_id)
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)

View File

@@ -49,44 +49,6 @@ class NSConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
MINOR_VERSION = 1
async def _validate_api_key(self, api_key: str) -> dict[str, str]:
"""Validate the API key by testing connection to NS API.
Returns a dict of errors, empty if validation successful.
"""
errors: dict[str, str] = {}
client = NSAPI(api_key)
try:
await self.hass.async_add_executor_job(client.get_stations)
except HTTPError:
errors["base"] = "invalid_auth"
except (RequestsConnectionError, Timeout):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception validating API key")
errors["base"] = "unknown"
return errors
def _is_api_key_already_configured(
self, api_key: str, exclude_entry_id: str | None = None
) -> dict[str, str]:
"""Check if the API key is already configured in another entry.
Args:
api_key: The API key to check.
exclude_entry_id: Optional entry ID to exclude from the check.
Returns:
A dict of errors, empty if not already configured.
"""
for entry in self._async_current_entries():
if (
entry.entry_id != exclude_entry_id
and entry.data.get(CONF_API_KEY) == api_key
):
return {"base": "already_configured"}
return {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@@ -94,7 +56,16 @@ class NSConfigFlow(ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match(user_input)
errors = await self._validate_api_key(user_input[CONF_API_KEY])
client = NSAPI(user_input[CONF_API_KEY])
try:
await self.hass.async_add_executor_job(client.get_stations)
except HTTPError:
errors["base"] = "invalid_auth"
except (RequestsConnectionError, Timeout):
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception validating API key")
errors["base"] = "unknown"
if not errors:
return self.async_create_entry(
title=INTEGRATION_TITLE,
@@ -106,33 +77,6 @@ class NSConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration to update the API key from the UI."""
errors: dict[str, str] = {}
reconfigure_entry = self._get_reconfigure_entry()
if user_input is not None:
# Check if this API key is already used by another entry
errors = self._is_api_key_already_configured(
user_input[CONF_API_KEY], exclude_entry_id=reconfigure_entry.entry_id
)
if not errors:
errors = await self._validate_api_key(user_input[CONF_API_KEY])
if not errors:
return self.async_update_reload_and_abort(
reconfigure_entry,
data_updates={CONF_API_KEY: user_input[CONF_API_KEY]},
)
return self.async_show_form(
step_id="reconfigure",
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Handle import from YAML configuration."""
self._async_abort_entries_match({CONF_API_KEY: import_data[CONF_API_KEY]})

View File

@@ -1,25 +1,14 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"already_configured": "This API key is already configured for another entry.",
"cannot_connect": "Could not connect to NS API. Check your API key.",
"invalid_auth": "[%key:common::config_flow::error::invalid_api_key%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"reconfigure": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "[%key:component::nederlandse_spoorwegen::config::step::user::data_description::api_key%]"
},
"description": "Update your Nederlandse Spoorwegen API key."
},
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"

View File

@@ -23,28 +23,6 @@ PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.SENSOR]
DATA_PRIVILEGED_KEY: HassKey[bool | None] = HassKey(DOMAIN)
async def async_migrate_entry(hass: HomeAssistant, entry: PingConfigEntry) -> bool:
"""Migrate old config entries."""
if entry.version == 1 and entry.minor_version == 1:
_LOGGER.debug("Migrating to minor version 2")
# Migrate device registry identifiers from homeassistant domain to ping domain
registry = dr.async_get(hass)
if (
device := registry.async_get_device(
identifiers={(HOMEASSISTANT_DOMAIN, entry.entry_id)}
)
) is not None and entry.entry_id in device.config_entries:
registry.async_update_device(
device_id=device.id,
new_identifiers={(DOMAIN, entry.entry_id)},
)
hass.config_entries.async_update_entry(entry, minor_version=2)
return True
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the ping integration."""
hass.data[DATA_PRIVILEGED_KEY] = await _can_use_icmp_lib_with_privilege()
@@ -54,6 +32,19 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: PingConfigEntry) -> bool:
"""Set up Ping (ICMP) from a config entry."""
# Migrate device registry identifiers from homeassistant domain to ping domain
registry = dr.async_get(hass)
if (
device := registry.async_get_device(
identifiers={(HOMEASSISTANT_DOMAIN, entry.entry_id)}
)
) is not None and entry.entry_id in device.config_entries:
registry.async_update_device(
device_id=device.id,
new_identifiers={(DOMAIN, entry.entry_id)},
)
privileged = hass.data[DATA_PRIVILEGED_KEY]
host: str = entry.options[CONF_HOST]

View File

@@ -37,7 +37,6 @@ class PingConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Ping."""
VERSION = 1
MINOR_VERSION = 2
async def async_step_user(
self, user_input: dict[str, Any] | None = None

View File

@@ -10,12 +10,11 @@ from homeassistant.components.device_tracker import (
ScannerEntity,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util
from .const import CONF_IMPORTED_BY, DOMAIN
from .const import CONF_IMPORTED_BY
from .coordinator import PingConfigEntry, PingUpdateCoordinator
@@ -25,7 +24,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up a Ping config entry."""
async_add_entities([PingDeviceTracker(hass, entry, entry.runtime_data)])
async_add_entities([PingDeviceTracker(entry, entry.runtime_data)])
class PingDeviceTracker(CoordinatorEntity[PingUpdateCoordinator], ScannerEntity):
@@ -34,10 +33,7 @@ class PingDeviceTracker(CoordinatorEntity[PingUpdateCoordinator], ScannerEntity)
_last_seen: datetime | None = None
def __init__(
self,
hass: HomeAssistant,
config_entry: PingConfigEntry,
coordinator: PingUpdateCoordinator,
self, config_entry: PingConfigEntry, coordinator: PingUpdateCoordinator
) -> None:
"""Initialize the Ping device tracker."""
super().__init__(coordinator)
@@ -50,13 +46,6 @@ class PingDeviceTracker(CoordinatorEntity[PingUpdateCoordinator], ScannerEntity)
)
)
if (
device := dr.async_get(hass).async_get_device(
identifiers={(DOMAIN, config_entry.entry_id)}
)
) is not None:
self.device_entry = device
@property
def ip_address(self) -> str:
"""Return the primary ip address of the device."""

View File

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
@@ -15,10 +16,11 @@ from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, PLATFORMS, PLATFORMS_WITH_AUTH
from .coordinator import SFRConfigEntry, SFRDataUpdateCoordinator, SFRRuntimeData
from .coordinator import SFRDataUpdateCoordinator
from .models import DomainData
async def async_setup_entry(hass: HomeAssistant, entry: SFRConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up SFR box as config entry."""
box = SFRBox(ip=entry.data[CONF_HOST], client=async_get_clientsession(hass))
platforms = PLATFORMS
@@ -33,7 +35,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: SFRConfigEntry) -> bool:
raise ConfigEntryNotReady from err
platforms = PLATFORMS_WITH_AUTH
data = SFRRuntimeData(
data = DomainData(
box=box,
dsl=SFRDataUpdateCoordinator(
hass, entry, box, "dsl", lambda b: b.dsl_get_info()
@@ -62,6 +64,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: SFRConfigEntry) -> bool:
tasks.append(data.ftth.async_config_entry_first_refresh())
await asyncio.gather(*tasks)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = data
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
@@ -73,12 +77,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: SFRConfigEntry) -> bool:
configuration_url=f"http://{entry.data[CONF_HOST]}",
)
entry.runtime_data = data
await hass.config_entries.async_forward_entry_setups(entry, platforms)
return True
async def async_unload_entry(hass: HomeAssistant, entry: SFRConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

View File

@@ -6,19 +6,23 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
from sfrbox_api.models import DslInfo, FtthInfo, WanInfo
from sfrbox_api.models import DslInfo, FtthInfo, SystemInfo, WanInfo
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
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 .coordinator import SFRConfigEntry
from .entity import SFRCoordinatorEntity
from .const import DOMAIN
from .coordinator import SFRDataUpdateCoordinator
from .models import DomainData
@dataclass(frozen=True, kw_only=True)
@@ -59,11 +63,11 @@ WAN_SENSOR_TYPES: tuple[SFRBoxBinarySensorEntityDescription[WanInfo], ...] = (
async def async_setup_entry(
hass: HomeAssistant,
entry: SFRConfigEntry,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sensors."""
data = entry.runtime_data
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
@@ -86,10 +90,29 @@ async def async_setup_entry(
async_add_entities(entities)
class SFRBoxBinarySensor[_T](SFRCoordinatorEntity[_T], BinarySensorEntity):
"""SFR Box binary sensor."""
class SFRBoxBinarySensor[_T](
CoordinatorEntity[SFRDataUpdateCoordinator[_T]], BinarySensorEntity
):
"""SFR Box sensor."""
entity_description: SFRBoxBinarySensorEntityDescription[_T]
_attr_has_entity_name = True
def __init__(
self,
coordinator: SFRDataUpdateCoordinator[_T],
description: SFRBoxBinarySensorEntityDescription,
system_info: SystemInfo,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = (
f"{system_info.mac_addr}_{coordinator.name}_{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, system_info.mac_addr)},
)
@property
def is_on(self) -> bool | None:

View File

@@ -16,13 +16,15 @@ from homeassistant.components.button import (
ButtonEntity,
ButtonEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SFRConfigEntry
from .entity import SFREntity
from .const import DOMAIN
from .models import DomainData
def with_error_wrapping[**_P, _R](
@@ -64,11 +66,11 @@ BUTTON_TYPES: tuple[SFRBoxButtonEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant,
entry: SFRConfigEntry,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the buttons."""
data = entry.runtime_data
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
@@ -79,10 +81,11 @@ async def async_setup_entry(
async_add_entities(entities)
class SFRBoxButton(SFREntity, ButtonEntity):
"""SFR Box button."""
class SFRBoxButton(ButtonEntity):
"""Mixin for button specific attributes."""
entity_description: SFRBoxButtonEntityDescription
_attr_has_entity_name = True
def __init__(
self,
@@ -90,9 +93,13 @@ class SFRBoxButton(SFREntity, ButtonEntity):
description: SFRBoxButtonEntityDescription,
system_info: SystemInfo,
) -> None:
"""Initialize the button."""
super().__init__(description, system_info)
"""Initialize the sensor."""
self.entity_description = description
self._box = box
self._attr_unique_id = f"{system_info.mac_addr}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, system_info.mac_addr)},
)
@with_error_wrapping
async def async_press(self) -> None:

View File

@@ -1,16 +1,12 @@
"""SFR Box coordinator."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxError
from sfrbox_api.models import DslInfo, FtthInfo, SystemInfo, WanInfo
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@@ -19,29 +15,16 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
_LOGGER = logging.getLogger(__name__)
_SCAN_INTERVAL = timedelta(minutes=1)
type SFRConfigEntry = ConfigEntry[SFRRuntimeData]
@dataclass
class SFRRuntimeData:
"""Runtime data for SFR Box."""
box: SFRBox
dsl: SFRDataUpdateCoordinator[DslInfo]
ftth: SFRDataUpdateCoordinator[FtthInfo]
system: SFRDataUpdateCoordinator[SystemInfo]
wan: SFRDataUpdateCoordinator[WanInfo]
class SFRDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT | None]):
"""Coordinator to manage data updates."""
config_entry: SFRConfigEntry
config_entry: ConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: SFRConfigEntry,
config_entry: ConfigEntry,
box: SFRBox,
name: str,
method: Callable[[SFRBox], Coroutine[Any, Any, _DataT | None]],

View File

@@ -6,9 +6,11 @@ import dataclasses
from typing import TYPE_CHECKING, Any
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .coordinator import SFRConfigEntry
from .const import DOMAIN
from .models import DomainData
if TYPE_CHECKING:
from _typeshed import DataclassInstance
@@ -23,10 +25,10 @@ def _async_redact_data(obj: DataclassInstance | None) -> dict[str, Any] | None:
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: SFRConfigEntry
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
data = entry.runtime_data
data: DomainData = hass.data[DOMAIN][entry.entry_id]
return {
"entry": {

View File

@@ -1,45 +0,0 @@
"""SFR Box base entity."""
from __future__ import annotations
from sfrbox_api.models import SystemInfo
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import SFRDataUpdateCoordinator
class SFREntity(Entity):
"""SFR Box entity."""
_attr_has_entity_name = True
def __init__(self, description: EntityDescription, system_info: SystemInfo) -> None:
"""Initialize the entity."""
self.entity_description = description
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, system_info.mac_addr)},
)
self._attr_unique_id = f"{system_info.mac_addr}_{description.key}"
class SFRCoordinatorEntity[_T](
CoordinatorEntity[SFRDataUpdateCoordinator[_T]], SFREntity
):
"""SFR Box coordinator entity."""
def __init__(
self,
coordinator: SFRDataUpdateCoordinator[_T],
description: EntityDescription,
system_info: SystemInfo,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
SFREntity.__init__(self, description, system_info)
self._attr_unique_id = (
f"{system_info.mac_addr}_{coordinator.name}_{description.key}"
)

View File

@@ -0,0 +1,19 @@
"""SFR Box models."""
from dataclasses import dataclass
from sfrbox_api.bridge import SFRBox
from sfrbox_api.models import DslInfo, FtthInfo, SystemInfo, WanInfo
from .coordinator import SFRDataUpdateCoordinator
@dataclass
class DomainData:
"""Domain data for SFR Box."""
box: SFRBox
dsl: SFRDataUpdateCoordinator[DslInfo]
ftth: SFRDataUpdateCoordinator[FtthInfo]
system: SFRDataUpdateCoordinator[SystemInfo]
wan: SFRDataUpdateCoordinator[WanInfo]

View File

@@ -12,6 +12,7 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
SIGNAL_STRENGTH_DECIBELS,
EntityCategory,
@@ -20,11 +21,14 @@ from homeassistant.const import (
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import SFRConfigEntry
from .entity import SFRCoordinatorEntity
from .const import DOMAIN
from .coordinator import SFRDataUpdateCoordinator
from .models import DomainData
@dataclass(frozen=True, kw_only=True)
@@ -216,11 +220,11 @@ def _get_temperature(value: float | None) -> float | None:
async def async_setup_entry(
hass: HomeAssistant,
entry: SFRConfigEntry,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the sensors."""
data = entry.runtime_data
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
@@ -242,10 +246,27 @@ async def async_setup_entry(
async_add_entities(entities)
class SFRBoxSensor[_T](SFRCoordinatorEntity[_T], SensorEntity):
class SFRBoxSensor[_T](CoordinatorEntity[SFRDataUpdateCoordinator[_T]], SensorEntity):
"""SFR Box sensor."""
entity_description: SFRBoxSensorEntityDescription[_T]
_attr_has_entity_name = True
def __init__(
self,
coordinator: SFRDataUpdateCoordinator[_T],
description: SFRBoxSensorEntityDescription,
system_info: SystemInfo,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = (
f"{system_info.mac_addr}_{coordinator.name}_{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, system_info.mac_addr)},
)
@property
def native_value(self) -> StateType:

View File

@@ -13,10 +13,6 @@
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "The password for accessing your SFR box's web interface, the default is the WiFi security key found on the device label",
"username": "The username for accessing your SFR box's web interface, the default is 'admin'"
}
},
"choose_auth": {

View File

@@ -41,5 +41,5 @@
"iot_class": "local_push",
"loggers": ["switchbot"],
"quality_scale": "gold",
"requirements": ["PySwitchbot==0.72.1"]
"requirements": ["PySwitchbot==0.72.0"]
}

View File

@@ -2,21 +2,24 @@
from __future__ import annotations
from collections.abc import Callable
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, cast
from aioswitcher.api import SwitcherApi
from aioswitcher.api.messages import SwitcherBaseResponse
from aioswitcher.api.remotes import SwitcherBreezeRemote
from aioswitcher.device import DeviceCategory, DeviceState, ThermostatSwing
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import SwitcherConfigEntry
from .const import API_CONTROL_BREEZE_DEVICE, SIGNAL_DEVICE_ADD
from .const import SIGNAL_DEVICE_ADD
from .coordinator import SwitcherDataUpdateCoordinator
from .entity import SwitcherEntity
from .utils import get_breeze_remote_manager
@@ -28,7 +31,10 @@ PARALLEL_UPDATES = 1
class SwitcherThermostatButtonEntityDescription(ButtonEntityDescription):
"""Class to describe a Switcher Thermostat Button entity."""
press_args: dict[str, Any]
press_fn: Callable[
[SwitcherApi, SwitcherBreezeRemote],
Coroutine[Any, Any, SwitcherBaseResponse],
]
supported: Callable[[SwitcherBreezeRemote], bool]
@@ -37,26 +43,34 @@ THERMOSTAT_BUTTONS = [
key="assume_on",
translation_key="assume_on",
entity_category=EntityCategory.CONFIG,
press_args={"state": DeviceState.ON, "update_state": True},
press_fn=lambda api, remote: api.control_breeze_device(
remote, state=DeviceState.ON, update_state=True
),
supported=lambda _: True,
),
SwitcherThermostatButtonEntityDescription(
key="assume_off",
translation_key="assume_off",
entity_category=EntityCategory.CONFIG,
press_args={"state": DeviceState.OFF, "update_state": True},
press_fn=lambda api, remote: api.control_breeze_device(
remote, state=DeviceState.OFF, update_state=True
),
supported=lambda _: True,
),
SwitcherThermostatButtonEntityDescription(
key="vertical_swing_on",
translation_key="vertical_swing_on",
press_args={"swing": ThermostatSwing.ON},
press_fn=lambda api, remote: api.control_breeze_device(
remote, swing=ThermostatSwing.ON
),
supported=lambda remote: bool(remote.separated_swing_command),
),
SwitcherThermostatButtonEntityDescription(
key="vertical_swing_off",
translation_key="vertical_swing_off",
press_args={"swing": ThermostatSwing.OFF},
press_fn=lambda api, remote: api.control_breeze_device(
remote, swing=ThermostatSwing.OFF
),
supported=lambda remote: bool(remote.separated_swing_command),
),
]
@@ -107,8 +121,23 @@ class SwitcherThermostatButtonEntity(SwitcherEntity, ButtonEntity):
async def async_press(self) -> None:
"""Press the button."""
await self._async_call_api(
API_CONTROL_BREEZE_DEVICE,
self._remote,
**self.entity_description.press_args,
)
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
) as swapi:
response = await self.entity_description.press_fn(swapi, self._remote)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
self.coordinator.last_update_success = False
self.async_write_ha_state()
raise HomeAssistantError(
f"Call api for {self.name} failed, response/error: {response or error}"
)

View File

@@ -27,18 +27,20 @@ from homeassistant.components.climate import (
)
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import SwitcherConfigEntry
from .const import API_CONTROL_BREEZE_DEVICE, SIGNAL_DEVICE_ADD
from .const import SIGNAL_DEVICE_ADD
from .coordinator import SwitcherDataUpdateCoordinator
from .entity import SwitcherEntity
from .utils import get_breeze_remote_manager
PARALLEL_UPDATES = 1
API_CONTROL_BREEZE_DEVICE = "control_breeze_device"
DEVICE_MODE_TO_HA = {
ThermostatMode.COOL: HVACMode.COOL,
ThermostatMode.HEAT: HVACMode.HEAT,
@@ -157,16 +159,21 @@ class SwitcherClimateEntity(SwitcherEntity, ClimateEntity):
"""Set new target temperature."""
data = cast(SwitcherThermostat, self.coordinator.data)
if not self._remote.modes_features[data.mode]["temperature_control"]:
raise ServiceValidationError(
"Current mode does not support setting 'Target temperature'."
raise HomeAssistantError(
"Current mode doesn't support setting Target Temperature"
)
await self._async_control_breeze_device(
target_temp=int(kwargs[ATTR_TEMPERATURE])
)
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
raise ValueError("No target temperature provided")
await self._async_control_breeze_device(target_temp=int(temperature))
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
data = cast(SwitcherThermostat, self.coordinator.data)
if not self._remote.modes_features[data.mode]["fan_levels"]:
raise HomeAssistantError("Current mode doesn't support setting Fan Mode")
await self._async_control_breeze_device(fan_level=HA_TO_DEVICE_FAN[fan_mode])
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
@@ -180,6 +187,10 @@ class SwitcherClimateEntity(SwitcherEntity, ClimateEntity):
async def async_set_swing_mode(self, swing_mode: str) -> None:
"""Set new target swing operation."""
data = cast(SwitcherThermostat, self.coordinator.data)
if not self._remote.modes_features[data.mode]["swing"]:
raise HomeAssistantError("Current mode doesn't support setting Swing Mode")
if swing_mode == SWING_VERTICAL:
await self._async_control_breeze_device(swing=ThermostatSwing.ON)
else:

View File

@@ -2,8 +2,6 @@
DOMAIN = "switcher_kis"
API_CONTROL_BREEZE_DEVICE = "control_breeze_device"
DISCOVERY_TIME_SEC = 12
SIGNAL_DEVICE_ADD = "switcher_device_add"

View File

@@ -6,7 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/switcher_kis",
"iot_class": "local_push",
"loggers": ["aioswitcher"],
"quality_scale": "silver",
"requirements": ["aioswitcher==6.0.0"],
"single_config_entry": true
}

View File

@@ -14,7 +14,7 @@ rules:
docs-actions: done
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-removal-instructions: todo
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
@@ -28,7 +28,7 @@ rules:
comment: The integration only supports a single config entry.
# Silver
action-exceptions: done
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done

View File

@@ -7,5 +7,5 @@
"documentation": "https://www.home-assistant.io/integrations/traccar",
"iot_class": "cloud_push",
"loggers": ["pytraccar"],
"requirements": ["pytraccar==3.0.0", "stringcase==1.2.0"]
"requirements": ["pytraccar==2.1.1", "stringcase==1.2.0"]
}

View File

@@ -9,7 +9,6 @@ from pytraccar import ApiClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_TOKEN,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
@@ -19,7 +18,6 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.event import async_track_time_interval
@@ -35,11 +33,6 @@ PLATFORMS: list[Platform] = [
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Traccar Server from a config entry."""
if CONF_API_TOKEN not in entry.data:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="migrate_to_api_token",
)
client_session = async_create_clientsession(
hass,
cookie_jar=CookieJar(
@@ -53,7 +46,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
client_session=client_session,
host=entry.data[CONF_HOST],
port=entry.data[CONF_PORT],
token=entry.data[CONF_API_TOKEN],
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
ssl=entry.data[CONF_SSL],
verify_ssl=entry.data[CONF_VERIFY_SSL],
),
@@ -96,15 +90,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle an options update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate old entry."""
if entry.version < 2:
# Version 2: Remove username and password, only keep API token
data = dict(entry.data)
data.pop(CONF_USERNAME, None)
data.pop(CONF_PASSWORD, None)
hass.config_entries.async_update_entry(entry, data=data, version=2)
return True

View File

@@ -16,10 +16,11 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
CONF_API_TOKEN,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import callback
@@ -60,7 +61,10 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
vol.Optional(CONF_PORT, default="8082"): TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT)
),
vol.Required(CONF_API_TOKEN): TextSelector(
vol.Required(CONF_USERNAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL)
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
vol.Optional(CONF_SSL, default=False): BooleanSelector(BooleanSelectorConfig()),
@@ -116,17 +120,16 @@ OPTIONS_FLOW = {
class TraccarServerConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Traccar Server."""
VERSION = 2
async def _get_server_info(self, user_input: dict[str, Any]) -> ServerModel:
"""Get server info."""
client = ApiClient(
client_session=async_get_clientsession(self.hass),
host=user_input[CONF_HOST],
port=user_input[CONF_PORT],
username=user_input[CONF_USERNAME],
password=user_input[CONF_PASSWORD],
ssl=user_input[CONF_SSL],
verify_ssl=user_input[CONF_VERIFY_SSL],
token=user_input[CONF_API_TOKEN],
)
return await client.get_server()
@@ -198,11 +201,19 @@ class TraccarServerConfigFlow(ConfigFlow, domain=DOMAIN):
reauth_entry,
data_updates=user_input,
)
username = (
user_input[CONF_USERNAME]
if user_input
else reauth_entry.data[CONF_USERNAME]
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_API_TOKEN): TextSelector(
vol.Required(CONF_USERNAME, default=username): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL)
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
}

View File

@@ -5,5 +5,5 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/traccar_server",
"iot_class": "local_push",
"requirements": ["pytraccar==3.0.0"]
"requirements": ["pytraccar==2.1.1"]
}

View File

@@ -12,21 +12,23 @@
"step": {
"reauth_confirm": {
"data": {
"api_token": "[%key:common::config_flow::data::api_token%]"
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"description": "The authentication credentials for {host}:{port} need to be updated."
},
"user": {
"data": {
"api_token": "[%key:common::config_flow::data::api_token%]",
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]",
"port": "[%key:common::config_flow::data::port%]",
"ssl": "[%key:common::config_flow::data::ssl%]",
"username": "[%key:common::config_flow::data::username%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"data_description": {
"api_token": "The API token generated from your account on your Traccar Server",
"host": "The hostname or IP address of your Traccar Server"
"host": "The hostname or IP address of your Traccar Server",
"username": "The username (email) you use to log in to your Traccar Server"
}
}
}
@@ -60,11 +62,6 @@
}
}
},
"exceptions": {
"migrate_to_api_token": {
"message": "To continue using Traccar Server, you need to migrate to API token based authentication."
}
},
"options": {
"step": {
"init": {

View File

@@ -222,24 +222,6 @@ BINARY_SENSORS: dict[DeviceCategory, tuple[TuyaBinarySensorEntityDescription, ..
on_value={"AQAB"},
),
),
DeviceCategory.MSP: (
TuyaBinarySensorEntityDescription(
key=f"{DPCode.FAULT}_full_fault",
dpcode=DPCode.FAULT,
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
bitmap_key="full_fault",
translation_key="bag_full",
),
TuyaBinarySensorEntityDescription(
key=f"{DPCode.FAULT}_box_out",
dpcode=DPCode.FAULT,
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
bitmap_key="box_out",
translation_key="cover_off",
),
),
DeviceCategory.PIR: (
TuyaBinarySensorEntityDescription(
key=DPCode.PIR,

View File

@@ -714,8 +714,6 @@ class DPCode(StrEnum):
ECO2 = "eco2"
EDGE_BRUSH = "edge_brush"
ELECTRICITY_LEFT = "electricity_left"
EXCRETION_TIME_DAY = "excretion_time_day"
EXCRETION_TIMES_DAY = "excretion_times_day"
FAN_BEEP = "fan_beep" # Sound
FAN_COOL = "fan_cool" # Cool wind
FAN_DIRECTION = "fan_direction" # Fan direction

View File

@@ -760,15 +760,6 @@ SENSORS: dict[DeviceCategory, tuple[TuyaSensorEntityDescription, ...]] = {
device_class=SensorDeviceClass.WEIGHT,
state_class=SensorStateClass.MEASUREMENT,
),
TuyaSensorEntityDescription(
key=DPCode.EXCRETION_TIME_DAY,
translation_key="excretion_time_day",
device_class=SensorDeviceClass.DURATION,
),
TuyaSensorEntityDescription(
key=DPCode.EXCRETION_TIMES_DAY,
translation_key="excretion_times_day",
),
),
DeviceCategory.MZJ: (
TuyaSensorEntityDescription(

View File

@@ -27,18 +27,12 @@
},
"entity": {
"binary_sensor": {
"bag_full": {
"name": "Bag full"
},
"carbon_dioxide": {
"name": "Carbon dioxide"
},
"carbon_monoxide": {
"name": "Carbon monoxide"
},
"cover_off": {
"name": "Cover off"
},
"defrost": {
"name": "Defrost"
},
@@ -626,12 +620,6 @@
"duster_cloth_life": {
"name": "Duster cloth lifetime"
},
"excretion_time_day": {
"name": "Excretion time (day)"
},
"excretion_times_day": {
"name": "Excretion times (day)"
},
"feels_like_temperature": {
"name": "Feels like"
},

View File

@@ -7,5 +7,5 @@
"documentation": "https://www.home-assistant.io/integrations/watergate",
"iot_class": "local_push",
"quality_scale": "bronze",
"requirements": ["watergate-local-api==2025.1.0"]
"requirements": ["watergate-local-api==2024.4.1"]
}

View File

@@ -17,7 +17,6 @@ CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.IMAGE,
Platform.MEDIA_PLAYER,
Platform.REMOTE,
Platform.SENSOR,

View File

@@ -9,6 +9,7 @@ from typing import Any
from xbox.webapi.api.provider.people.models import Person
from xbox.webapi.api.provider.titlehub.models import Title
from yarl import URL
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
@@ -19,12 +20,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import XboxConfigEntry
from .entity import (
XboxBaseEntity,
XboxBaseEntityDescription,
check_deprecated_entity,
profile_pic,
)
from .entity import XboxBaseEntity, XboxBaseEntityDescription, check_deprecated_entity
class XboxBinarySensor(StrEnum):
@@ -47,6 +43,23 @@ class XboxBinarySensorEntityDescription(
deprecated: bool | None = None
def profile_pic(person: Person, _: Title | None) -> str | None:
"""Return the gamer pic."""
# Xbox sometimes returns a domain that uses a wrong certificate which
# creates issues with loading the image.
# The correct domain is images-eds-ssl which can just be replaced
# to point to the correct image, with the correct domain and certificate.
# We need to also remove the 'mode=Padding' query because with it,
# it results in an error 400.
url = URL(person.display_pic_raw)
if url.host == "images-eds.xboxlive.com":
url = url.with_host("images-eds-ssl.xboxlive.com").with_scheme("https")
query = dict(url.query)
query.pop("mode", None)
return str(url.with_query(query))
def profile_attributes(person: Person, _: Title | None) -> dict[str, Any]:
"""Attributes for the profile."""
attributes: dict[str, Any] = {}

View File

@@ -9,7 +9,6 @@ from typing import Any
from xbox.webapi.api.provider.people.models import Person
from xbox.webapi.api.provider.smartglass.models import ConsoleType, SmartglassConsole
from xbox.webapi.api.provider.titlehub.models import Title
from yarl import URL
from homeassistant.components.automation import automations_with_entity
from homeassistant.components.script import scripts_with_entity
@@ -40,7 +39,6 @@ class XboxBaseEntityDescription(EntityDescription):
attributes_fn: Callable[[Person, Title | None], Mapping[str, Any] | None] | None = (
None
)
deprecated: bool | None = None
class XboxBaseEntity(CoordinatorEntity[XboxUpdateCoordinator]):
@@ -159,20 +157,3 @@ def check_deprecated_entity(
ent_reg.async_remove(entity_id)
return False
def profile_pic(person: Person, _: Title | None) -> str | None:
"""Return the gamer pic."""
# Xbox sometimes returns a domain that uses a wrong certificate which
# creates issues with loading the image.
# The correct domain is images-eds-ssl which can just be replaced
# to point to the correct image, with the correct domain and certificate.
# We need to also remove the 'mode=Padding' query because with it,
# it results in an error 400.
url = URL(person.display_pic_raw)
if url.host == "images-eds.xboxlive.com":
url = url.with_host("images-eds-ssl.xboxlive.com").with_scheme("https")
query = dict(url.query)
query.pop("mode", None)
return str(url.with_query(query))

View File

@@ -1,123 +0,0 @@
"""Image platform for the Xbox integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from xbox.webapi.api.provider.people.models import Person
from xbox.webapi.api.provider.titlehub.models import Title
from homeassistant.components.image import ImageEntity, ImageEntityDescription
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import XboxConfigEntry, XboxUpdateCoordinator
from .entity import XboxBaseEntity, XboxBaseEntityDescription, profile_pic
PARALLEL_UPDATES = 0
class XboxImage(StrEnum):
"""Xbox image."""
NOW_PLAYING = "now_playing"
GAMERPIC = "gamerpic"
AVATAR = "avatar"
@dataclass(kw_only=True, frozen=True)
class XboxImageEntityDescription(XboxBaseEntityDescription, ImageEntityDescription):
"""Xbox image description."""
image_url_fn: Callable[[Person, Title | None], str | None]
IMAGE_DESCRIPTIONS: tuple[XboxImageEntityDescription, ...] = (
XboxImageEntityDescription(
key=XboxImage.GAMERPIC,
translation_key=XboxImage.GAMERPIC,
image_url_fn=profile_pic,
),
XboxImageEntityDescription(
key=XboxImage.NOW_PLAYING,
translation_key=XboxImage.NOW_PLAYING,
image_url_fn=lambda _, title: title.display_image if title else None,
),
XboxImageEntityDescription(
key=XboxImage.AVATAR,
translation_key=XboxImage.AVATAR,
image_url_fn=(
lambda person,
_: f"https://avatar-ssl.xboxlive.com/avatar/{person.gamertag}/avatar-body.png"
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: XboxConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Xbox images."""
coordinator = config_entry.runtime_data
xuids_added: set[str] = set()
@callback
def add_entities() -> None:
"""Add image entities."""
nonlocal xuids_added
current_xuids = set(coordinator.data.presence)
if new_xuids := current_xuids - xuids_added:
async_add_entities(
[
XboxImageEntity(hass, coordinator, xuid, description)
for xuid in new_xuids
for description in IMAGE_DESCRIPTIONS
]
)
xuids_added |= new_xuids
xuids_added &= current_xuids
coordinator.async_add_listener(add_entities)
add_entities()
class XboxImageEntity(XboxBaseEntity, ImageEntity):
"""An image entity."""
entity_description: XboxImageEntityDescription
def __init__(
self,
hass: HomeAssistant,
coordinator: XboxUpdateCoordinator,
xuid: str,
entity_description: XboxImageEntityDescription,
) -> None:
"""Initialize the image entity."""
super().__init__(coordinator, xuid, entity_description)
ImageEntity.__init__(self, hass)
self._attr_image_url = self.entity_description.image_url_fn(
self.data, self.title_info
)
self._attr_image_last_updated = dt_util.utcnow()
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
url = self.entity_description.image_url_fn(self.data, self.title_info)
if url != self._attr_image_url:
self._attr_image_url = url
self._cached_image = None
self._attr_image_last_updated = dt_util.utcnow()
super()._handle_coordinator_update()

View File

@@ -44,17 +44,6 @@
}
}
},
"image": {
"avatar": {
"name": "Avatar"
},
"gamerpic": {
"name": "Gamerpic"
},
"now_playing": {
"name": "[%key:component::xbox::entity::sensor::now_playing::name%]"
}
},
"sensor": {
"follower": {
"name": "Follower",

View File

@@ -61,9 +61,6 @@ class ZHAEntity(LogMixin, RestoreEntity, Entity):
if meta.translation_key is not None:
self._attr_translation_key = meta.translation_key
if meta.translation_placeholders is not None:
self._attr_translation_placeholders = meta.translation_placeholders
@cached_property
def name(self) -> str | UndefinedType | None:
"""Return the name of the entity."""

View File

@@ -21,7 +21,7 @@
"zha",
"universal_silabs_flasher"
],
"requirements": ["zha==0.0.75"],
"requirements": ["zha==0.0.73"],
"usb": [
{
"description": "*2652*",

View File

@@ -179,7 +179,6 @@
"left": "Left",
"open": "[%key:common::action::open%]",
"right": "Right",
"rotary_knob": "Rotary knob",
"turn_off": "[%key:common::action::turn_off%]",
"turn_on": "[%key:common::action::turn_on%]"
},
@@ -207,10 +206,7 @@
"remote_button_quintuple_press": "\"{subtype}\" quintuple clicked",
"remote_button_short_press": "\"{subtype}\" pressed",
"remote_button_short_release": "\"{subtype}\" released",
"remote_button_triple_press": "\"{subtype}\" triple clicked",
"rotary_knob_continued_rotating": "Rotary knob continued rotating \"{subtype}\"",
"rotary_knob_started_rotating": "Rotary knob started rotating \"{subtype}\"",
"rotary_knob_stopped_rotating": "\"{subtype}\" stopped rotating"
"remote_button_triple_press": "\"{subtype}\" triple clicked"
}
},
"entity": {
@@ -309,9 +305,6 @@
"calibrate_valve": {
"name": "Calibrate valve"
},
"calibrate_z_axis": {
"name": "Calibrate Z axis"
},
"feed": {
"name": "Feed"
},
@@ -330,12 +323,6 @@
"reset_summation_delivered": {
"name": "Reset summation delivered"
},
"reset_summation_delivered_left": {
"name": "Reset left summation delivered"
},
"reset_summation_delivered_right": {
"name": "Reset right summation delivered"
},
"restart_device": {
"name": "Restart device"
},
@@ -477,9 +464,6 @@
"comfort_temperature_min": {
"name": "Comfort temperature min"
},
"compensation_speed": {
"name": "Compensation speed"
},
"deadzone_temperature": {
"name": "Deadzone temperature"
},
@@ -639,9 +623,6 @@
"lift_drive_up_time": {
"name": "Lift drive up time"
},
"limit_position": {
"name": "Limit position"
},
"liquid_depth_max": {
"name": "Height from sensor to liquid level"
},
@@ -738,9 +719,6 @@
"on_transition_time": {
"name": "On transition time"
},
"open_delay_time": {
"name": "Open delay time"
},
"open_window_detection_guard_period": {
"name": "Open window detection guard period"
},
@@ -768,9 +746,6 @@
"presence_timeout": {
"name": "Fade time"
},
"pulse_configuration": {
"name": "Pulse configuration"
},
"quantitative_watering": {
"name": "Quantitative watering"
},
@@ -879,24 +854,6 @@
"transmit_power": {
"name": "Transmit power"
},
"turn_off_delay": {
"name": "Turn off delay"
},
"turn_off_delay_left": {
"name": "Turn off delay left"
},
"turn_off_delay_right": {
"name": "Turn off delay right"
},
"turn_on_delay": {
"name": "Turn on delay"
},
"turn_on_delay_left": {
"name": "Turn on delay left"
},
"turn_on_delay_right": {
"name": "Turn on delay right"
},
"up_movement": {
"name": "Up movement"
},
@@ -1634,9 +1591,6 @@
"double_up_full": {
"name": "Double tap on - full"
},
"enable_pir_mode": {
"name": "Enable PIR remote"
},
"enable_siren": {
"name": "Enable siren"
},

View File

@@ -124,14 +124,6 @@ RF_REGIONS = [
"USA",
]
# USB devices to ignore in serial port selection (non-Z-Wave devices)
# Format: (manufacturer, description)
IGNORED_USB_DEVICES = {
("Nabu Casa", "SkyConnect v1.0"),
("Nabu Casa", "Home Assistant Connect ZBT-1"),
("Nabu Casa", "ZBT-2"),
}
def get_manual_schema(user_input: dict[str, Any]) -> vol.Schema:
"""Return a schema for the manual step."""
@@ -163,9 +155,6 @@ def get_usb_ports() -> dict[str, str]:
ports = list_ports.comports()
port_descriptions = {}
for port in ports:
if (port.manufacturer, port.description) in IGNORED_USB_DEVICES:
continue
vid: str | None = None
pid: str | None = None
if port.vid is not None and port.pid is not None:

View File

@@ -27,6 +27,7 @@ APPLICATION_CREDENTIALS = [
"miele",
"monzo",
"myuplink",
"neato",
"nest",
"netatmo",
"ondilo_ico",

View File

@@ -428,6 +428,7 @@ FLOWS = {
"nam",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"nest",
"netatmo",

View File

@@ -4339,6 +4339,12 @@
"integration_type": "virtual",
"supported_by": "opower"
},
"neato": {
"name": "Neato Botvac",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"nederlandse_spoorwegen": {
"name": "Nederlandse Spoorwegen (NS)",
"integration_type": "service",

10
mypy.ini generated
View File

@@ -3366,6 +3366,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.neato.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.nest.*]
check_untyped_defs = true
disallow_incomplete_defs = true

19
requirements_all.txt generated
View File

@@ -87,7 +87,7 @@ PyRMVtransport==0.3.3
PySrDaliGateway==0.13.1
# homeassistant.components.switchbot
PySwitchbot==0.72.1
PySwitchbot==0.72.0
# homeassistant.components.switchmate
PySwitchmate==0.5.1
@@ -194,7 +194,7 @@ aioairzone-cloud==0.7.2
aioairzone==1.0.2
# homeassistant.components.alexa_devices
aioamazondevices==6.5.5
aioamazondevices==6.4.6
# homeassistant.components.ambient_network
# homeassistant.components.ambient_station
@@ -465,7 +465,7 @@ airly==1.1.0
airos==0.6.0
# homeassistant.components.airthings_ble
airthings-ble==1.2.0
airthings-ble==1.1.1
# homeassistant.components.airthings
airthings-cloud==0.2.0
@@ -1151,7 +1151,7 @@ ha-iotawattpy==0.1.2
ha-philipsjs==3.2.4
# homeassistant.components.homeassistant_hardware
ha-silabs-firmware-client==0.3.0
ha-silabs-firmware-client==0.2.0
# homeassistant.components.habitica
habiticalib==0.4.6
@@ -1911,6 +1911,9 @@ pyblackbird==0.6
# homeassistant.components.bluesound
pyblu==2.0.5
# homeassistant.components.neato
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.3.4
@@ -2605,7 +2608,7 @@ pytouchlinesl==0.5.0
# homeassistant.components.traccar
# homeassistant.components.traccar_server
pytraccar==3.0.0
pytraccar==2.1.1
# homeassistant.components.tradfri
pytradfri[async]==9.0.1
@@ -3081,7 +3084,7 @@ unifi_ap==0.0.2
unifiled==0.11
# homeassistant.components.homeassistant_hardware
universal-silabs-flasher==0.0.37
universal-silabs-flasher==0.0.35
# homeassistant.components.upb
upb-lib==0.6.1
@@ -3150,7 +3153,7 @@ watchdog==6.0.0
waterfurnace==1.2.0
# homeassistant.components.watergate
watergate-local-api==2025.1.0
watergate-local-api==2024.4.1
# homeassistant.components.weatherflow_cloud
weatherflow4py==1.4.1
@@ -3259,7 +3262,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==0.0.75
zha==0.0.73
# homeassistant.components.zhong_hong
zhong-hong-hvac==1.0.13

View File

@@ -84,7 +84,7 @@ PyRMVtransport==0.3.3
PySrDaliGateway==0.13.1
# homeassistant.components.switchbot
PySwitchbot==0.72.1
PySwitchbot==0.72.0
# homeassistant.components.syncthru
PySyncThru==0.8.0
@@ -182,7 +182,7 @@ aioairzone-cloud==0.7.2
aioairzone==1.0.2
# homeassistant.components.alexa_devices
aioamazondevices==6.5.5
aioamazondevices==6.4.6
# homeassistant.components.ambient_network
# homeassistant.components.ambient_station
@@ -447,7 +447,7 @@ airly==1.1.0
airos==0.6.0
# homeassistant.components.airthings_ble
airthings-ble==1.2.0
airthings-ble==1.1.1
# homeassistant.components.airthings
airthings-cloud==0.2.0
@@ -1012,7 +1012,7 @@ ha-iotawattpy==0.1.2
ha-philipsjs==3.2.4
# homeassistant.components.homeassistant_hardware
ha-silabs-firmware-client==0.3.0
ha-silabs-firmware-client==0.2.0
# homeassistant.components.habitica
habiticalib==0.4.6
@@ -1616,6 +1616,9 @@ pyblackbird==0.6
# homeassistant.components.bluesound
pyblu==2.0.5
# homeassistant.components.neato
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.3.4
@@ -2166,7 +2169,7 @@ pytouchlinesl==0.5.0
# homeassistant.components.traccar
# homeassistant.components.traccar_server
pytraccar==3.0.0
pytraccar==2.1.1
# homeassistant.components.tradfri
pytradfri[async]==9.0.1
@@ -2552,7 +2555,7 @@ ultraheat-api==0.5.7
unifi-discovery==1.2.0
# homeassistant.components.homeassistant_hardware
universal-silabs-flasher==0.0.37
universal-silabs-flasher==0.0.35
# homeassistant.components.upb
upb-lib==0.6.1
@@ -2612,7 +2615,7 @@ wallbox==0.9.0
watchdog==6.0.0
# homeassistant.components.watergate
watergate-local-api==2025.1.0
watergate-local-api==2024.4.1
# homeassistant.components.weatherflow_cloud
weatherflow4py==1.4.1
@@ -2706,7 +2709,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==0.0.75
zha==0.0.73
# homeassistant.components.zwave_js
zwave-js-server-python==0.67.1

View File

@@ -673,6 +673,7 @@ INTEGRATIONS_WITHOUT_QUALITY_SCALE_FILE = [
"namecheapdns",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"ness_alarm",
"netatmo",
@@ -1705,6 +1706,7 @@ INTEGRATIONS_WITHOUT_SCALE = [
"namecheapdns",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"nest",
"ness_alarm",
@@ -1988,6 +1990,7 @@ INTEGRATIONS_WITHOUT_SCALE = [
"switch_as_x",
"switchbee",
"switchbot_cloud",
"switcher_kis",
"switchmate",
"syncthing",
"syncthru",

View File

@@ -170,6 +170,11 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
# pyhive-integration > unasync > setuptools
"unasync": {"setuptools"}
},
"homeassistant_hardware": {
# https://github.com/zigpy/zigpy/issues/1604
# universal-silabs-flasher > zigpy > pyserial-asyncio
"zigpy": {"pyserial-asyncio"},
},
"homewizard": {"python-homewizard-energy": {"async-timeout"}},
"imeon_inverter": {"imeon-inverter-api": {"async-timeout"}},
"influxdb": {
@@ -262,6 +267,9 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
# https://github.com/waveform80/colorzero/issues/9
# zha > zigpy-zigate > gpiozero > colorzero > setuptools
"colorzero": {"setuptools"},
# https://github.com/zigpy/zigpy/issues/1604
# zha > zigpy > pyserial-asyncio
"zigpy": {"pyserial-asyncio"},
},
}

View File

@@ -91,7 +91,7 @@ def run_single(translations, flattened_translations, integration):
json.dumps({"component": {integration: translations["component"][integration]}})
)
download.save_integrations_translations()
download.write_integration_translations()
def run():

View File

@@ -5,13 +5,15 @@ from __future__ import annotations
import json
from pathlib import Path
import re
import subprocess
from typing import Any
from .const import CLI_2_DOCKER_IMAGE, CORE_PROJECT_ID, INTEGRATIONS_DIR
from .error import ExitApp
from .util import get_lokalise_token, load_json_from_path
from .util import flatten_translations, get_lokalise_token, load_json_from_path
FILENAME_FORMAT = re.compile(r"strings\.(?P<suffix>\w+)\.json")
DOWNLOAD_DIR = Path("build/translations-download").absolute()
@@ -59,58 +61,73 @@ def save_json(filename: Path, data: list | dict) -> None:
filename.write_text(json.dumps(data, sort_keys=True, indent=4), encoding="utf-8")
def filter_translations(translations: dict[str, Any], strings: dict[str, Any]) -> None:
"""Remove translations that are not in the original strings."""
for key in list(translations.keys()):
if key not in strings:
translations.pop(key)
continue
def get_component_path(lang, component) -> Path | None:
"""Get the component translation path."""
if (Path("homeassistant") / "components" / component).is_dir():
return (
Path("homeassistant")
/ "components"
/ component
/ "translations"
/ f"{lang}.json"
)
return None
if isinstance(translations[key], dict):
if not isinstance(strings[key], dict):
translations.pop(key)
continue
filter_translations(translations[key], strings[key])
if not translations[key]:
translations.pop(key)
continue
def get_platform_path(lang, component, platform) -> Path:
"""Get the platform translation path."""
return (
Path("homeassistant")
/ "components"
/ component
/ "translations"
/ f"{platform}.{lang}.json"
)
def get_component_translations(translations):
"""Get the component level translations."""
translations = translations.copy()
translations.pop("platform", None)
return translations
def save_language_translations(lang, translations):
"""Save translations for a single language."""
"""Distribute the translations for this language."""
components = translations.get("component", {})
for component, component_translations in components.items():
# Remove legacy platform translations
component_translations.pop("platform", None)
base_translations = get_component_translations(component_translations)
if base_translations:
if (path := get_component_path(lang, component)) is None:
print(
f"Skipping {lang} for {component}, as the integration doesn't seem to exist."
)
continue
if not (
Path("homeassistant") / "components" / component / "strings.json"
).exists():
print(
f"Skipping {lang} for {component}, as the integration doesn't have a strings.json file."
)
continue
path.parent.mkdir(parents=True, exist_ok=True)
base_translations = pick_keys(component, base_translations)
save_json(path, base_translations)
if not component_translations:
if "platform" not in component_translations:
continue
component_path = Path("homeassistant") / "components" / component
if not component_path.is_dir():
print(
f"Skipping {lang} for {component}, as the integration doesn't seem to exist."
)
continue
strings_path = component_path / "strings.json"
if not strings_path.exists():
print(
f"Skipping {lang} for {component}, as the integration doesn't have a strings.json file."
)
continue
strings = load_json_from_path(strings_path)
path = component_path / "translations" / f"{lang}.json"
path.parent.mkdir(parents=True, exist_ok=True)
filter_translations(component_translations, strings)
save_json(path, component_translations)
for platform, platform_translations in component_translations[
"platform"
].items():
path = get_platform_path(lang, component, platform)
path.parent.mkdir(parents=True, exist_ok=True)
save_json(path, platform_translations)
def save_integrations_translations():
"""Save integrations translations."""
def write_integration_translations():
"""Write integration translations."""
for lang_file in DOWNLOAD_DIR.glob("*.json"):
lang = lang_file.stem
translations = load_json_from_path(lang_file)
@@ -123,6 +140,32 @@ def delete_old_translations():
fil.unlink()
def get_current_keys(component: str) -> dict[str, Any]:
"""Get the current keys for a component."""
strings_path = Path("homeassistant") / "components" / component / "strings.json"
return load_json_from_path(strings_path)
def pick_keys(component: str, translations: dict[str, Any]) -> dict[str, Any]:
"""Pick the keys that are in the current strings."""
flat_translations = flatten_translations(translations)
flat_current_keys = flatten_translations(get_current_keys(component))
flatten_result = {}
for key in flat_current_keys:
if key in flat_translations:
flatten_result[key] = flat_translations[key]
result = {}
for key, value in flatten_result.items():
parts = key.split("::")
d = result
for part in parts[:-1]:
if part not in d:
d[part] = {}
d = d[part]
d[parts[-1]] = value
return result
def run():
"""Run the script."""
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
@@ -131,6 +174,6 @@ def run():
delete_old_translations()
save_integrations_translations()
write_integration_translations()
return 0

View File

@@ -4,12 +4,14 @@
import json
import os
import pathlib
import re
import subprocess
from .const import CLI_2_DOCKER_IMAGE, CORE_PROJECT_ID, INTEGRATIONS_DIR
from .error import ExitApp
from .util import get_current_branch, get_lokalise_token, load_json_from_path
FILENAME_FORMAT = re.compile(r"strings\.(?P<suffix>\w+)\.json")
LOCAL_FILE = pathlib.Path("build/translations-upload.json").absolute()
CONTAINER_FILE = "/opt/src/build/translations-upload.json"
LANG_ISO = "en"
@@ -52,11 +54,20 @@ def run_upload_docker():
def generate_upload_data():
"""Generate the data for uploading."""
translations = load_json_from_path(INTEGRATIONS_DIR.parent / "strings.json")
translations["component"] = {}
translations["component"] = {
path.parent.name: load_json_from_path(path)
for path in INTEGRATIONS_DIR.glob(f"*{os.sep}strings.json")
}
for path in INTEGRATIONS_DIR.glob(f"*{os.sep}strings*.json"):
component = path.parent.name
match = FILENAME_FORMAT.search(path.name)
platform = match.group("suffix") if match else None
parent = translations["component"].setdefault(component, {})
if platform:
platforms = parent.setdefault("platform", {})
parent = platforms.setdefault(platform, {})
parent.update(load_json_from_path(path))
return translations

View File

@@ -1,13 +1,6 @@
"""Alexa Devices tests const."""
from datetime import UTC, datetime
from aioamazondevices.api import AmazonDevice, AmazonDeviceSensor, AmazonSchedule
from aioamazondevices.const import (
NOTIFICATION_ALARM,
NOTIFICATION_REMINDER,
NOTIFICATION_TIMER,
)
from aioamazondevices.api import AmazonDevice, AmazonDeviceSensor
TEST_CODE = "023123"
TEST_PASSWORD = "fake_password"
@@ -46,26 +39,6 @@ TEST_DEVICE_1 = AmazonDevice(
scale="CELSIUS",
),
},
notifications={
NOTIFICATION_ALARM: AmazonSchedule(
type=NOTIFICATION_ALARM,
status="ON",
label="Morning Alarm",
next_occurrence=datetime(2023, 10, 1, 7, 0, 0, tzinfo=UTC),
),
NOTIFICATION_REMINDER: AmazonSchedule(
type=NOTIFICATION_REMINDER,
status="ON",
label="Take out the trash",
next_occurrence=None,
),
NOTIFICATION_TIMER: AmazonSchedule(
type=NOTIFICATION_TIMER,
status="OFF",
label="",
next_occurrence=None,
),
},
)
TEST_DEVICE_2_SN = "echo_test_2_serial_number"
@@ -93,5 +66,4 @@ TEST_DEVICE_2 = AmazonDevice(
scale="CELSIUS",
)
},
notifications={},
)

View File

@@ -1,151 +1,4 @@
# serializer version: 1
# name: test_all_entities[sensor.echo_test_next_alarm-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.echo_test_next_alarm',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Next alarm',
'platform': 'alexa_devices',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'alarm',
'unique_id': 'echo_test_serial_number-Alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.echo_test_next_alarm-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Echo Test Next alarm',
}),
'context': <ANY>,
'entity_id': 'sensor.echo_test_next_alarm',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-10-01T07:00:00+00:00',
})
# ---
# name: test_all_entities[sensor.echo_test_next_reminder-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.echo_test_next_reminder',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Next reminder',
'platform': 'alexa_devices',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'reminder',
'unique_id': 'echo_test_serial_number-Reminder',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.echo_test_next_reminder-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Echo Test Next reminder',
}),
'context': <ANY>,
'entity_id': 'sensor.echo_test_next_reminder',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_all_entities[sensor.echo_test_next_timer-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.echo_test_next_timer',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Next timer',
'platform': 'alexa_devices',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'timer',
'unique_id': 'echo_test_serial_number-Timer',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[sensor.echo_test_next_timer-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Echo Test Next timer',
}),
'context': <ANY>,
'entity_id': 'sensor.echo_test_next_timer',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_all_entities[sensor.echo_test_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -17,26 +17,6 @@
'endpoint_id': 'G1234567890123456789012345678A',
'entity_id': '11111111-2222-3333-4444-555555555555',
'household_device': False,
'notifications': dict({
'Alarm': dict({
'label': 'Morning Alarm',
'next_occurrence': datetime.datetime(2023, 10, 1, 7, 0, tzinfo=datetime.timezone.utc),
'status': 'ON',
'type': 'Alarm',
}),
'Reminder': dict({
'label': 'Take out the trash',
'next_occurrence': None,
'status': 'ON',
'type': 'Reminder',
}),
'Timer': dict({
'label': '',
'next_occurrence': None,
'status': 'OFF',
'type': 'Timer',
}),
}),
'online': True,
'sensors': dict({
'dnd': dict({
@@ -83,26 +63,6 @@
'endpoint_id': 'G1234567890123456789012345678A',
'entity_id': '11111111-2222-3333-4444-555555555555',
'household_device': False,
'notifications': dict({
'Alarm': dict({
'label': 'Morning Alarm',
'next_occurrence': datetime.datetime(2023, 10, 1, 7, 0, tzinfo=datetime.timezone.utc),
'status': 'ON',
'type': 'Alarm',
}),
'Reminder': dict({
'label': 'Take out the trash',
'next_occurrence': None,
'status': 'ON',
'type': 'Reminder',
}),
'Timer': dict({
'label': '',
'next_occurrence': None,
'status': 'OFF',
'type': 'Timer',
}),
}),
'online': True,
'sensors': dict({
'dnd': dict({

View File

@@ -577,6 +577,7 @@
"ourgroceries": 469,
"vicare": 1495,
"thermopro": 639,
"neato": 935,
"roon": 405,
"renault": 1287,
"bthome": 4166,

Some files were not shown because too many files have changed in this diff Show More