latest changes

This commit is contained in:
2025-11-20 20:18:29 +01:00
parent 69252eb9c1
commit 87275eb7c3
21 changed files with 65 additions and 4888 deletions
@@ -1,119 +0,0 @@
"""electrolux status integration."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_LANGUAGE,
CONF_PASSWORD,
CONF_USERNAME,
CONF_COUNTRY_CODE,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_RENEW_INTERVAL,
DEFAULT_LANGUAGE,
DEFAULT_COUNTRY_CODE,
DEFAULT_WEBSOCKET_RENEWAL_DELAY,
DOMAIN,
PLATFORMS,
languages,
)
from .coordinator import ElectroluxCoordinator
from .util import get_electrolux_session
_LOGGER: logging.Logger = logging.getLogger(__package__)
# noinspection PyUnusedLocal
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up this integration using YAML is not supported."""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up this integration using UI."""
if hass.data.get(DOMAIN) is None:
hass.data.setdefault(DOMAIN, {})
renew_interval = DEFAULT_WEBSOCKET_RENEWAL_DELAY
if entry.options.get(CONF_RENEW_INTERVAL):
renew_interval = entry.options[CONF_RENEW_INTERVAL]
username = entry.data.get(CONF_USERNAME)
password = entry.data.get(CONF_PASSWORD)
country_code = entry.data.get(CONF_COUNTRY_CODE, DEFAULT_COUNTRY_CODE)
language = languages.get(entry.data.get(CONF_LANGUAGE, DEFAULT_LANGUAGE), "eng")
session = async_get_clientsession(hass)
client = get_electrolux_session(username, password, country_code, session, language)
coordinator = ElectroluxCoordinator(
hass,
client=client,
renew_interval=renew_interval,
username=username,
)
await coordinator.get_stored_token()
if not await coordinator.async_login():
raise ConfigEntryAuthFailed("Electrolux wrong credentials")
# Bug ?
if coordinator.config_entry is None:
coordinator.config_entry = entry
hass.data[DOMAIN][entry.entry_id] = coordinator
# Initialize entities
_LOGGER.debug("async_setup_entry setup_entities")
await coordinator.setup_entities()
_LOGGER.debug("async_setup_entry listen_websocket")
coordinator.listen_websocket()
# _LOGGER.debug("async_setup_entry launch_websocket_renewal_task")
# await coordinator.launch_websocket_renewal_task()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, coordinator.api.close)
)
entry.async_on_unload(entry.add_update_listener(update_listener))
_LOGGER.debug("async_setup_entry async_config_entry_first_refresh")
# Fill in the values for first time
await coordinator.async_config_entry_first_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
_LOGGER.debug("async_setup_entry extend PLATFORMS")
coordinator.platforms.extend(PLATFORMS)
# Call async_setup_entry in entity files
_LOGGER.debug("async_setup_entry async_forward_entry_setups")
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
_LOGGER.debug("async_setup_entry OVER")
return True
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Update listener."""
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Handle removal of an entry."""
coordinator: ElectroluxCoordinator = hass.data[DOMAIN][entry.entry_id]
await coordinator.close_websocket()
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload config entry."""
_LOGGER.debug("Electrolux async_reload_entry %s", entry)
await async_unload_entry(hass, entry)
await async_setup_entry(hass, entry)
-701
View File
@@ -1,701 +0,0 @@
"""API for Electrolux Status."""
import copy
import logging
import re
from typing import Any
from pyelectroluxocp.apiModels import ApplianceInfoResponse, ApplienceStatusResponse
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.button import ButtonDeviceClass
from homeassistant.components.number import NumberDeviceClass
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.components.switch import SwitchDeviceClass
from homeassistant.const import Platform, UnitOfTemperature
from .binary_sensor import ElectroluxBinarySensor
from .button import ElectroluxButton
from .catalog_core import CATALOG_BASE, CATALOG_MODEL
from .const import (
BINARY_SENSOR,
BUTTON,
ATTRIBUTES_BLACKLIST,
NUMBER,
PLATFORMS,
RENAME_RULES,
SELECT,
SENSOR,
STATIC_ATTRIBUTES,
SWITCH, ATTRIBUTES_WHITELIST,
)
from .entity import ElectroluxEntity
from .model import ElectroluxDevice
from .number import ElectroluxNumber
from .select import ElectroluxSelect
from .sensor import ElectroluxSensor
from .switch import ElectroluxSwitch
_LOGGER: logging.Logger = logging.getLogger(__package__)
HEADERS = {"Content-type": "application/json; charset=UTF-8"}
def deep_merge_dicts(dict1, dict2):
"""
Recursively merge two dictionaries.
"""
result = dict1.copy()
for key, value in dict2.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = deep_merge_dicts(result[key], value)
else:
result[key] = value
return result
class ElectroluxLibraryEntity:
"""Electrolux Library Entity."""
def __init__(
self,
name,
status: str,
state: ApplienceStatusResponse,
appliance_info: ApplianceInfoResponse,
capabilities: dict[str, Any],
) -> None:
"""Initaliaze the entity."""
self.name = name
self.status = status
self.state = state
self.appliance_info = appliance_info
self.capabilities = capabilities
@property
def reported_state(self) -> dict[str, Any]:
"""Return the reported state of the appliance."""
return self.state.get("properties", {}).get("reported")
def get_name(self):
"""Get entity name."""
return self.name
def get_value(self, attr_name) -> Any:
"""Return value by attribute."""
if "/" in attr_name:
source, attr = attr_name.split("/")
return self.reported_state.get(source, {}).get(attr, None)
return self.reported_state.get(attr_name, None)
def get_sensor_name(self, attr_name: str) -> str:
"""Get the name of the sensor."""
sensor = attr_name
for truncate_rule in RENAME_RULES:
sensor = re.sub(truncate_rule, "", sensor)
sensor = sensor[0].upper() + sensor[1:]
sensor = sensor.replace("_", " ")
sensor = sensor.replace("/", " ")
group = ""
words = []
for i, char in enumerate(sensor):
if group == "":
group = char
else:
if char == " " and len(group) > 0:
words.append(group)
group = ""
continue
if (
(char.isupper() or char.isdigit())
and (sensor[i - 1].isupper() or sensor[i - 1].isdigit())
and (
(i == len(sensor) - 1)
or (sensor[i + 1].isupper() or sensor[i + 1].isdigit())
)
):
group += char
elif (char.isupper() or char.isdigit()) and sensor[i - 1].islower():
if re.match("^[A-Z0-9]+$", group):
words.append(group)
else:
words.append(group.lower())
group = char
else:
group += char
if len(group) > 0:
if re.match("^[A-Z0-9]+$", group):
words.append(group)
else:
words.append(group.lower())
return " ".join(words).lower().capitalize()
# def get_sensor_name_old(self, attr_name: str, container: str | None = None):
# """Convert sensor format.
# ex: "fCMiscellaneousState/detergentExtradosage" to "Detergent extradosage".
# """
# attr_name = attr_name.rpartition("/")[-1] or attr_name
# attr_name = attr_name[0].upper() + attr_name[1:]
# attr_name = " ".join(re.findall("[A-Z][^A-Z]*", attr_name))
# return attr_name.capitalize()
def get_entity_name(self, attr_name: str) -> str:
"""Extract Entity Name.
ex: Convert format "fCMiscellaneousState/EWX1493A_detergentExtradosage" to "XdetergentExtradosage"
"""
for truncate_rule in RENAME_RULES:
attr_name = re.sub(truncate_rule, "", attr_name)
return attr_name.rpartition("/")[-1] or attr_name
def get_entity_attr(self, attr_name: str) -> str:
"""Extract Entity attr in raw format.
ex: Convert format "fCMiscellaneousState/EWX1493A_detergentExtradosage" to "EWX1493A_detergentExtradosage"
"""
return attr_name.rpartition("/")[-1] or attr_name
def get_category(self, attr_name: str) -> str:
"""Extract category.
ex: "fCMiscellaneousState/detergentExtradosage" to "fCMiscellaneousState".
or "" if none
"""
return attr_name.rpartition("/")[0]
def get_capability(self, attr_name: str) -> dict[str, Any] | None:
"""Retrieve the capability from self.capabilities using the attribute name.
May contain slashes for nested keys.
"""
if self.capabilities.get(attr_name, None):
return self.capabilities.get(attr_name)
keys = attr_name.split("/")
result = self.capabilities
for key in keys:
result = result.get(key)
if result is None:
return None
return result
def get_entity_unit(self, attr_name: str):
"""Get entity unit type."""
capability_def: dict[str, Any] | None = self.get_capability(attr_name)
if not capability_def:
return None
# Type : string, int, number, boolean (other values ignored)
type_units = capability_def.get("type", None)
if not type_units:
return None
if type_units == "temperature":
return UnitOfTemperature.CELSIUS
return None
def get_entity_device_class(self, attr_name: str):
"""Get entity device class."""
capability_def: dict[str, Any] | None = self.get_capability(attr_name)
if not capability_def:
return None
# Type : string, int, number, boolean (other values ignored)
type_class = capability_def.get("type", None)
if not type_class:
return None
if type_class == "temperature":
if capability_def.get("access", None) == "readwrite":
return NumberDeviceClass.TEMPERATURE
return SensorDeviceClass.TEMPERATURE
return None
def get_entity_type(self, attr_name: str) -> Platform | None:
"""Get entity type."""
capability_def: dict[str, Any] | None = self.get_capability(attr_name)
if not capability_def:
return None
# Type : string, int, number, boolean (other values ignored)
type_object = capability_def.get("type", None)
if not type_object:
return None
# Access : read, readwrite (other values ignored)
access = capability_def.get("access", None)
if not access:
return None
# Exception (Electrolux bug)
if (
type_object == "boolean"
and access == "readwrite"
and capability_def.get("values", None) is not None
):
return SWITCH
# List of values ? if values is defined and has at least 1 entry
values: dict[str, Any] | None = capability_def.get("values", None)
if (
values
and access == "readwrite"
and isinstance(values, dict)
and len(values) > 0
):
if (
type_object not in ["number", "temperature"]
or capability_def.get("min", None) is None
):
return SELECT
match type_object:
case "boolean":
if access == "read":
return BINARY_SENSOR
if access == "readwrite":
return SWITCH
case "temperature":
if access == "read":
return SENSOR
if access == "readwrite":
return NUMBER
case "alert":
return SENSOR
case _:
if (
self.get_entity_name(attr_name) == "executeCommand"
and access == "read"
): # FIX for https://github.com/albaintor/homeassistant_electrolux_status/issues/74
return BUTTON
if access == "write":
return BUTTON
if access == "constant":
return SENSOR
if access == "read" and type_object in [
"number",
"int",
"boolean",
"string",
]:
return SENSOR
if type_object in ("int", "number"):
return NUMBER
_LOGGER.debug(
"Electrolux unable to determine type for %s. Type: %s Access: %s",
attr_name,
type_object,
access,
)
return None
def sources_list(self) -> list[str] | None:
"""List the capability types."""
if self.capabilities is None:
_LOGGER.warning("Electrolux capabilities list is empty")
return None
# dont load these entities by as they are not useful
# we do load some of these directly via STATIC_ATTRIBUTES as
# one or another are useful, but not all child values are
def keep_source(source: str) -> bool:
for ignored_pattern in ATTRIBUTES_BLACKLIST:
if re.match(ignored_pattern, source):
for whitelist_pattern in ATTRIBUTES_WHITELIST:
if re.match(whitelist_pattern, source):
return True
_LOGGER.debug("Exclude source %s from list", source)
return False
return True
sources = [
key
for key in list(self.capabilities.keys())
if keep_source(key)
]
for key, value in self.capabilities.items():
if not keep_source(key):
continue
if isinstance(value, dict):
for sub_key, sub_value in value.items():
if (
isinstance(sub_value, dict)
and "access" in sub_value
and "type" in sub_value
):
sources.append(f"{key}/{sub_key}")
elif "access" in value and "type" in value:
sources.append(key)
return sources
# def sources_list_old(self):
# _LOGGER.warning(self.capabilities)
# return [
# key
# for key in list(self.capabilities.keys())
# if not key.startswith("applianceCareAndMaintenance")
# ]
class Appliance:
"""Define the Appliance Class."""
brand: str
device: str
entities: list[ElectroluxEntity]
coordinator: Any
def __init__(
self,
coordinator: Any,
name: str,
pnc_id: str,
brand: str,
model: str,
state: ApplienceStatusResponse,
) -> None:
"""Initiate the appliance."""
self.own_capabilties = False
self.data = None
self.coordinator = coordinator
self.model = model
self.pnc_id = pnc_id
self.name = name
self.brand = brand
self.state: ApplienceStatusResponse = state
@property
def reported_state(self) -> dict[str, Any]:
"""Return the reported state of the appliance."""
return self.state.get("properties", {}).get("reported", {})
@property
def appliance_type(self) -> dict[str, Any]:
"""Return the reported type of the appliance.
CR: Refridgerator
WM: Washing Machine
"""
return self.reported_state.get("applianceInfo", {}).get(
"applianceType"
) or self.state.get("applianceData", {}).get("modelName")
@property
def catalog(self) -> dict[str, ElectroluxDevice]:
"""Return the defined catalog for the appliance."""
# TODO: Use appliance_type as opposed to model?
if self.model in CATALOG_MODEL:
_LOGGER.debug("Extending catalog for %s", self.model)
# Make a deep copy of the base catalog to preserve it
new_catalog = copy.deepcopy(CATALOG_BASE)
# Get the specific model's extended catalog
model_catalog = CATALOG_MODEL[self.model]
# Update the existing catalog with the extended information for this model
for key, device in model_catalog.items():
new_catalog[key] = device
return new_catalog
return CATALOG_BASE
def update_missing_entities(self) -> None:
"""Add missing entities when no capabilities returned by the API.
This is done dynamically but only when the reported state contains the attributes.
"""
if not self.own_capabilties or not self.reported_state:
return
for key, catalog_item in self.catalog.items():
category = self.data.get_category(key)
if (
category
and self.reported_state.get(category, None)
and self.reported_state.get(category, None).get(key)
) or (not category and self.reported_state.get(key, None)):
found: bool = False
for entity in self.entities:
if entity.entity_attr == key and entity.entity_source == category:
found = True
keys = key.split("/")
capabilities = self.data.capabilities
for key in keys[:-1]:
capabilities = capabilities.setdefault(key, {})
capabilities[keys[-1]] = catalog_item.capability_info
break
if not found:
_LOGGER.debug(
"Electrolux discovered new entity from extracted data. Key: %s",
key,
)
if entity := self.get_entity(key):
self.entities.extend(entity)
def get_state(self, attr_name: str) -> dict[str, Any] | None:
"""Retrieve the start from self.reported_state using the attribute name.
May contain slashes for nested keys.
"""
keys = attr_name.split("/")
result = self.reported_state
for key in keys:
result = result.get(key)
if result is None:
return None
return result
def get_entity(self, capability: str) -> list[ElectroluxEntity] | None:
"""Return the entity."""
entity_type = self.data.get_entity_type(capability)
entity_name = self.data.get_entity_name(capability)
entity_attr = self.data.get_entity_attr(capability)
category = self.data.get_category(capability)
capability_info = self.data.get_capability(capability)
device_class = self.data.get_entity_device_class(capability)
entity_category = None
entity_icon = None
unit = self.data.get_entity_unit(capability)
display_name = self.data.get_sensor_name(capability)
# get the item definition from the catalog
catalog_item = self.catalog.get(capability, None)
if catalog_item:
if capability_info is None:
capability_info = catalog_item.capability_info
elif (
"values" not in capability_info
and "values" in catalog_item.capability_info
):
capability_info["values"] = catalog_item.capability_info["values"]
device_class = catalog_item.device_class
unit = catalog_item.unit
entity_category = catalog_item.entity_category
entity_icon = catalog_item.entity_icon
# override the api determined type by the catalog entity_type
if isinstance(device_class, BinarySensorDeviceClass):
entity_type = BINARY_SENSOR
if isinstance(device_class, ButtonDeviceClass):
entity_type = BUTTON
if isinstance(device_class, NumberDeviceClass):
entity_type = NUMBER
if isinstance(device_class, SensorDeviceClass):
entity_type = SENSOR
if isinstance(device_class, SwitchDeviceClass):
entity_type = SWITCH
# override the api determined type by the catalog entity_platform
if catalog_item and isinstance(catalog_item.entity_platform, Platform):
entity_type = catalog_item.entity_platform
_LOGGER.debug(
"Electrolux get_entity. entity_type: %s entity_name: %s entity_attr: %s entity_source: %s capability: %s device_class: %s unit: %s, catalog: %s",
entity_type,
entity_name,
entity_attr,
category,
capability_info,
device_class,
unit,
catalog_item,
)
def electrolux_entity_factory(
name: str,
entity_type: Platform | None,
entity_name: str,
entity_attr: str,
entity_source: str,
capability: str,
unit: str,
entity_category: str,
device_class: str,
icon: str,
catalog_entry: ElectroluxDevice | None,
commands: Any | None = None,
):
entity_classes = {
BINARY_SENSOR: ElectroluxBinarySensor,
BUTTON: ElectroluxButton,
NUMBER: ElectroluxNumber,
SELECT: ElectroluxSelect,
SENSOR: ElectroluxSensor,
SWITCH: ElectroluxSwitch,
}
entity_class = entity_classes.get(entity_type)
if entity_class is None:
_LOGGER.debug("Unknown entity type %s for %s", entity_type, name)
raise ValueError(f"Unknown entity type: {entity_type}")
entity_params = {
"coordinator": self.coordinator,
"config_entry": self.coordinator.config_entry,
"pnc_id": self.pnc_id,
"name": name,
"entity_type": entity_type,
"entity_name": entity_name,
"entity_attr": entity_attr,
"entity_source": entity_source,
"capability": capability,
"unit": unit,
"entity_category": entity_category,
"device_class": device_class,
"icon": icon,
"catalog_entry": catalog_entry,
}
if commands is None:
return [entity_class(**entity_params)]
entities: list[
ElectroluxBinarySensor
| ElectroluxNumber
| ElectroluxSensor
| ElectroluxSwitch
| ElectroluxButton
| ElectroluxSelect
] = []
# Replace entity name and icons for multi-entities attribute (one value = one entity)
for command in commands:
entity = {**entity_params, "val_to_send": command}
if catalog_item:
if catalog_item.entity_value_named:
entity["name"] = command
if (
catalog_item.entity_icons_value_map
and catalog_item.entity_icons_value_map.get(command, None)
):
entity["icon"] = catalog_item.entity_icons_value_map.get(
command
)
# Instanciate the new entity and append it
entities.append(entity_class(**entity))
return entities
if entity_type in PLATFORMS:
commands = (
capability_info.get("values", {}) if entity_type == BUTTON else None
)
return electrolux_entity_factory(
name=display_name,
entity_type=entity_type,
entity_name=entity_name,
entity_attr=entity_attr,
entity_source=category,
capability=capability_info,
unit=unit,
entity_category=entity_category,
device_class=device_class,
icon=entity_icon,
catalog_entry=catalog_item,
commands=commands,
)
return []
def setup(self, data: ElectroluxLibraryEntity):
"""Configure the entity."""
self.data: ElectroluxLibraryEntity = data
self.entities: list[ElectroluxEntity] = []
entities: list[ElectroluxEntity] = []
# Extraction of the appliance capabilities & mapping to the known entities of the component
# [ "applianceState", "autoDosing",..., "userSelections/analogTemperature",...]
capabilities_names = self.data.sources_list()
if capabilities_names is None and self.state:
# No capabilities returned (unstable API)
# We could rebuild them from catalog but this creates entities that are
# not required by each device type (fridge, dryer, vacumn etc are all different)
_LOGGER.warning("Electrolux API returned no capability definition")
# Add static attribute
# these are attributes that are not in the capability entry
# but are returned by the api independantly
for static_attribute in STATIC_ATTRIBUTES:
_LOGGER.debug("Electrolux static_attribute %s", static_attribute)
# attr not found in state, next attr
if self.get_state(static_attribute) is None:
continue
if catalog_item := self.catalog.get(static_attribute, None):
if (entity := self.get_entity(static_attribute)) is None:
# catalog definition and automatic checks fail to determine type
_LOGGER.debug(
"Electrolux static_attribute undefined %s", static_attribute
)
continue
# add to the capability dict
keys = static_attribute.split("/")
capabilities = self.data.capabilities
for key in keys[:-1]:
capabilities = capabilities.setdefault(key, {})
capabilities[keys[-1]] = catalog_item.capability_info
_LOGGER.debug("Electrolux adding static_attribute %s", static_attribute)
entities.extend(entity)
# For each capability src
if capabilities_names:
for capability in capabilities_names:
if entity := self.get_entity(capability):
entities.extend(entity)
else:
_LOGGER.debug("Could not create entity for capability %s", capability)
# Setup each found entity
self.entities = entities
for entity in entities:
entity.setup(data)
def update_reported_data(self, reported_data: dict[str, Any]):
"""Update the reported data."""
_LOGGER.debug("Electrolux update reported data %s", reported_data)
try:
self.reported_state.update(deep_merge_dicts(self.reported_state, reported_data))
_LOGGER.debug("Electrolux updated reported data %s", self.state)
self.update_missing_entities()
for entity in self.entities:
entity.update(self.state)
except Exception as ex: # noqa: BLE001
_LOGGER.debug(
"Electrolux status could not update reported data with %s. %s",
reported_data,
ex,
)
def update(self, appliance_status: ApplienceStatusResponse):
"""Update appliance status."""
self.state = appliance_status
self.update_missing_entities()
for entity in self.entities:
entity.update(self.state)
class Appliances:
"""Appliance class definition."""
def __init__(self, appliances: dict[str, Appliance]) -> None:
"""Initialize the class."""
self.appliances = appliances
def get_appliance(self, pnc_id) -> Appliance:
"""Return the appliance."""
return self.appliances.get(pnc_id, None)
def get_appliances(self) -> dict[str, Appliance]:
"""Return all appliances."""
return self.appliances
def get_appliance_ids(self) -> list[str]:
"""Return all appliance ids."""
return list(self.appliances)
@@ -1,66 +0,0 @@
"""Binary sensor platform for Electrolux Status."""
import logging
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import BINARY_SENSOR, DOMAIN
from .entity import ElectroluxEntity
from .util import string_to_boolean
_LOGGER: logging.Logger = logging.getLogger(__package__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Configure binary sensor platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
if appliances := coordinator.data.get("appliances", None):
for appliance_id, appliance in appliances.appliances.items():
entities = [
entity
for entity in appliance.entities
if entity.entity_type == BINARY_SENSOR
]
_LOGGER.debug(
"Electrolux add %d BINARY_SENSOR entities to registry for appliance %s",
len(entities),
appliance_id,
)
async_add_entities(entities)
class ElectroluxBinarySensor(ElectroluxEntity, BinarySensorEntity):
"""Electrolux Status binary_sensor class."""
@property
def entity_domain(self):
"""Enitity domain for the entry. Used for consistent entity_id."""
return BINARY_SENSOR
@property
def invert(self) -> bool:
"""Determine if the value returned for the entity needs to be reversed."""
if self.catalog_entry:
return self.catalog_entry.state_invert
return False
@property
def is_on(self) -> bool:
"""Return true if the binary_sensor is on."""
value = self.extract_value()
if isinstance(value, str):
value = string_to_boolean(value, True)
if value is None:
if self.catalog_entry and self.catalog_entry.state_mapping:
mapping = self.catalog_entry.state_mapping
value = self.get_state_attr(mapping)
if value is not None:
self._cached_value = value
return not self._cached_value if self.invert else self._cached_value
@@ -1,140 +0,0 @@
"""Button platform for Electrolux Status."""
import logging
from typing import Any
from pyelectroluxocp.oneAppApi import OneAppApi
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 AddEntitiesCallback
from .const import BUTTON, DOMAIN, icon_mapping
from .entity import ElectroluxEntity
from .model import ElectroluxDevice
_LOGGER: logging.Logger = logging.getLogger(__package__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Configure button platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
if appliances := coordinator.data.get("appliances", None):
for appliance_id, appliance in appliances.appliances.items():
entities = [
entity for entity in appliance.entities if entity.entity_type == BUTTON
]
_LOGGER.debug(
"Electrolux add %d BUTTON entities to registry for appliance %s",
len(entities),
appliance_id,
)
async_add_entities(entities)
class ElectroluxButton(ElectroluxEntity, ButtonEntity):
"""Electrolux Status button class."""
def __init__(
self,
coordinator: Any,
name: str,
config_entry,
pnc_id: str,
entity_type: str,
entity_name,
entity_attr,
entity_source,
capability: dict[str, Any],
unit: str,
device_class: str,
entity_category: EntityCategory,
icon: str,
catalog_entry: ElectroluxDevice | None,
val_to_send: str,
) -> None:
"""Initialize the Button Entity."""
super().__init__(
coordinator=coordinator,
capability=capability,
name=name,
config_entry=config_entry,
pnc_id=pnc_id,
entity_type=entity_type,
entity_name=entity_name,
entity_attr=entity_attr,
entity_source=entity_source,
unit=None,
device_class=device_class,
entity_category=entity_category,
icon=icon,
catalog_entry=catalog_entry,
)
self.val_to_send = val_to_send
@property
def entity_domain(self):
"""Enitity domain for the entry. Used for consistent entity_id."""
return BUTTON
@property
def unique_id(self) -> str:
"""Return a unique ID to use for this entity."""
return f"{self.config_entry.entry_id}-{self.val_to_send}-{self.entity_attr}-{self.entity_source}-{self.pnc_id}"
@property
def name(self) -> str:
"""Return the name of the sensor."""
name = self._name
if self.catalog_entry and self.catalog_entry.friendly_name:
name = (
f"{self.get_appliance.name} {self.catalog_entry.friendly_name.lower()}"
)
# Get the last word from the 'name' variable
# and compare to the command we are sending duplicate names
# "air filter state reset reset" for instance
last_word = name.split()[-1]
if last_word.lower() == str(self.val_to_send).lower():
return name
return f"{name} {self.val_to_send}"
@property
def icon(self) -> str | None:
"""Return the icon of the entity."""
return self._icon or icon_mapping.get(
self.val_to_send, "mdi:gesture-tap-button"
)
async def send_command(self) -> bool:
"""Send a command to the device."""
client: OneAppApi = self.api
value = self.val_to_send
if self.entity_source:
if self.entity_source == "userSelections":
command = {
self.entity_source: {
"programUID": self.appliance_status["properties"]['reported']["userSelections"]['programUID'],
self.entity_attr: value
},
}
else:
command = {self.entity_source: {self.entity_attr: value}}
else:
command = {self.entity_attr: value}
_LOGGER.debug("Electrolux send command %s", command)
result = await client.execute_appliance_command(self.pnc_id, command)
_LOGGER.debug("Electrolux send command result %s", result)
return True
async def async_press(self) -> None:
"""Execute a button press."""
await self.send_command()
# await self.hass.async_add_executor_job(self.send_command)
# if self.entity_attr == "ExecuteCommand":
# await self.hass.async_add_executor_job(self.coordinator.api.setHacl, self.get_appliance.pnc_id, "0x0403", self.val_to_send, self.entity_source)
@@ -1,237 +0,0 @@
"""Adds config flow for Electrolux Status."""
from collections.abc import Mapping
import logging
from typing import Any
import voluptuous as vol
from homeassistant.config_entries import (
CONN_CLASS_CLOUD_PUSH,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, CONF_COUNTRY_CODE
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
selector,
)
from .const import (
CONF_LANGUAGE,
CONF_NOTIFICATION_DEFAULT,
CONF_NOTIFICATION_DIAG,
CONF_NOTIFICATION_WARNING,
DEFAULT_LANGUAGE,
DEFAULT_COUNTRY_CODE,
DOMAIN,
languages,
)
from .util import get_electrolux_session
_LOGGER = logging.getLogger(__name__)
class ElectroluxStatusFlowHandler(ConfigFlow, domain=DOMAIN):
"""Config flow for Electrolux Status."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_CLOUD_PUSH
def __init__(self) -> None:
"""Initialize."""
self._errors = {}
async def async_step_user(self, user_input=None) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
self._errors = {}
if user_input is not None:
# check if the specified account is configured already
# to prevent them from being added twice
for entry in self._async_current_entries():
if user_input[CONF_USERNAME] == entry.data.get("username", None):
return self.async_abort(reason="already_configured_account")
valid = await self._test_credentials(
user_input[CONF_USERNAME], user_input[CONF_PASSWORD], user_input[CONF_COUNTRY_CODE]
)
if valid:
return self.async_create_entry(
title=user_input[CONF_USERNAME], data=user_input
)
self._errors["base"] = "invalid_auth"
return await self._show_config_form(user_input)
return await self._show_config_form(user_input)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle configuration by re-auth."""
return await self.async_step_reauth_validate(entry_data)
async def async_step_reauth_validate(self, user_input=None) -> ConfigFlowResult:
"""Handle reauth and validation."""
self._errors = {}
if user_input is not None:
valid = await self._test_credentials(
user_input[CONF_USERNAME], user_input[CONF_PASSWORD], user_input[CONF_COUNTRY_CODE]
)
if valid:
return self.async_create_entry(
title=user_input[CONF_USERNAME], data=user_input
)
self._errors["base"] = "invalid_auth"
return await self._show_config_form(user_input)
return await self._show_config_form(user_input)
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Present the configuration options dialog."""
return ElectroluxStatusOptionsFlowHandler(config_entry)
async def _show_config_form(self, user_input): # pylint: disable=unused-argument
"""Show the configuration form to edit location data."""
data_schema = {
vol.Required(CONF_USERNAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL, autocomplete="username")
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD, autocomplete="current-password"
)
),
vol.Required(CONF_COUNTRY_CODE, default=DEFAULT_COUNTRY_CODE): TextSelector(
TextSelectorConfig(
type=TextSelectorType.TEXT, autocomplete="country-code"
)
),
}
if self.show_advanced_options:
data_schema.update(
{
vol.Optional(CONF_LANGUAGE, default=DEFAULT_LANGUAGE): selector(
{
"select": {
"options": list(languages.keys()),
"mode": "dropdown",
}
}
),
vol.Optional(CONF_NOTIFICATION_DEFAULT, default=True): cv.boolean,
vol.Optional(CONF_NOTIFICATION_WARNING, default=False): cv.boolean,
vol.Optional(CONF_NOTIFICATION_DIAG, default=False): cv.boolean,
}
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(data_schema),
errors=self._errors,
)
async def _test_credentials(self, username, password, country_code):
"""Return true if credentials is valid."""
try:
client = get_electrolux_session(
username, password, country_code, async_get_clientsession(self.hass)
)
await client.get_appliances_list()
except Exception as inst: # pylint: disable=broad-except # noqa: BLE001
_LOGGER.error("Authentication to electrolux failed: %s", inst)
return False
return True
class ElectroluxStatusOptionsFlowHandler(OptionsFlow):
"""Config flow options handler for Electrolux Status."""
def __init__(self, config_entry) -> None:
"""Initialize HACS options flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None) -> ConfigFlowResult:
"""Manage the options."""
return await self.async_step_user()
async def async_step_user(self, user_input=None) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
if user_input is not None:
self.options.update(user_input)
return await self._update_options()
selected_language = self.config_entry.data.get(CONF_LANGUAGE, DEFAULT_LANGUAGE)
current_password = self.config_entry.data.get(CONF_PASSWORD, None)
current_country_code = self.config_entry.data.get(CONF_COUNTRY_CODE, None)
notify_alert = self.config_entry.data.get(CONF_NOTIFICATION_DEFAULT, True)
notify_warning = self.config_entry.data.get(CONF_NOTIFICATION_WARNING, False)
notify_diagnostic = self.config_entry.data.get(CONF_NOTIFICATION_DIAG, False)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_PASSWORD, default=current_password): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD,
autocomplete="current-password",
)
),
vol.Required(CONF_COUNTRY_CODE, default=current_country_code): TextSelector(
TextSelectorConfig(
type=TextSelectorType.TEXT, autocomplete="country-code"
)
),
vol.Required(CONF_LANGUAGE, default=selected_language): selector(
{
"select": {
"options": list(languages.keys()),
"mode": "dropdown",
}
}
),
vol.Optional(
CONF_NOTIFICATION_DEFAULT, default=notify_alert
): cv.boolean,
vol.Optional(
CONF_NOTIFICATION_WARNING, default=notify_warning
): cv.boolean,
vol.Optional(
CONF_NOTIFICATION_DIAG, default=notify_diagnostic
): cv.boolean,
# vol.Optional(
# CONF_RENEW_INTERVAL,
# default=self.config_entry.options.get(
# CONF_RENEW_INTERVAL, DEFAULT_WEBSOCKET_RENEWAL_DELAY
# ),
# ): cv.positive_int,
}
),
)
async def _update_options(self):
"""Update config entry options."""
_LOGGER.debug("Electrolux updating configuration")
data = {
**self.config_entry.data,
CONF_PASSWORD: self.options[CONF_PASSWORD],
CONF_COUNTRY_CODE: self.options[CONF_COUNTRY_CODE],
CONF_LANGUAGE: self.options[CONF_LANGUAGE],
CONF_NOTIFICATION_DEFAULT: self.options[CONF_NOTIFICATION_DEFAULT],
CONF_NOTIFICATION_WARNING: self.options[CONF_NOTIFICATION_WARNING],
CONF_NOTIFICATION_DIAG: self.options[CONF_NOTIFICATION_DIAG],
}
self.hass.config_entries.async_update_entry(self.config_entry, data=data)
return self.async_create_entry(
title=self.config_entry.data.get(CONF_USERNAME),
data=data,
)
@@ -1,105 +0,0 @@
"""The electrolux Status constants."""
import re
from homeassistant.const import Platform
# Base component constants
NAME = "Electrolux status"
DOMAIN = "electrolux_status"
DOMAIN_DATA = f"{DOMAIN}_data"
COMPONENTS_DIRECTORY = "custom_components"
LOOKUP_DIRECTORY = "appliance_definitions"
LOOKUP_DIRECTORY_PATH = f"{COMPONENTS_DIRECTORY}/{DOMAIN}/{LOOKUP_DIRECTORY}/"
# Platforms
BINARY_SENSOR = Platform.BINARY_SENSOR
BUTTON = Platform.BUTTON
NUMBER = Platform.NUMBER
SELECT = Platform.SELECT
SENSOR = Platform.SENSOR
SWITCH = Platform.SWITCH
PLATFORMS = [BINARY_SENSOR, BUTTON, NUMBER, SELECT, SENSOR, SWITCH]
# Configuration and options
CONF_LANGUAGE = "language"
CONF_RENEW_INTERVAL = "renew_interval"
CONF_NOTIFICATION_DEFAULT = "notifications"
CONF_NOTIFICATION_DIAG = "notifications_diagnostic"
CONF_NOTIFICATION_WARNING = "notifications_warning"
# Defaults
DEFAULT_LANGUAGE = "English"
DEFAULT_COUNTRY_CODE = "us"
DEFAULT_WEBSOCKET_RENEWAL_DELAY = 43200 # 12 hours
# these are attributes that appear in the state file but not in the capabilities.
# defining them here and in the catalog will allow these devices to be added dynamically
STATIC_ATTRIBUTES = [
"connectivityState",
"networkInterface/linkQualityIndicator",
"applianceMode",
]
# Icon mappings for default executeCommands
icon_mapping = {
"OFF": "mdi:power-off",
"ON": "mdi:power-on",
"START": "mdi:play",
"STOPRESET": "mdi:stop",
"PAUSE": "mdi:pause",
"RESUME": "mdi:play-pause",
}
# List of supported Mobile App languages
# refer to https://emea-production.api.electrolux.net/masterdata-service/api/v1/languages
languages = {
"български": "bul",
"český": "ces",
"Dansk": "dan",
"Deutsch": "deu",
"ελληνικός": "ell",
"English": "eng",
"eesti": "est",
"Soome": "fin",
"Français": "fra",
"Hrvatski": "hrv",
"magyar": "hun",
"Italiano": "ita",
"lettone": "lav",
"lituano": "lit",
"Luxembourgish": "ltz",
"nederlands": "nld",
"Norsk": "nor",
"Polski": "pol",
"Português": "por",
"Română": "ron",
"rusesc": "rus",
"slovenský": "slk",
"slovinský": "slv",
"Español": "spa",
"Svenska": "swe",
"Türk": "tur",
"Ukrayna": "ukr",
}
# List of attributes to ignore and that won't be added as entities (regex format)
ATTRIBUTES_BLACKLIST: list[str] = ["^fCMiscellaneous.+",
"fcOptisenseLoadWeight.*",
"applianceCareAndMaintenance.*",
"applianceMainBoardSwVersion",
"coolingValveState",
"networkInterface",
"temperatureRepresentation",
]
ATTRIBUTES_WHITELIST: list[str] = [".*waterUsage",
".*tankAReserve",
".*tankBReserve"]
# Rules to simplify the naming of entities
RENAME_RULES: list[str] = [r"^userSelections\/[^_]+_", r"^userSelections\/",
r"^fCMiscellaneousState\/[^_]+_", r"^fCMiscellaneousState\/"]
# List of entity names that need to be updated to 0 manually when they are close to 0
TIME_ENTITIES_TO_UPDATE = ["timeToEnd"]
@@ -1,249 +0,0 @@
"""Entity platform for Electrolux Status."""
import logging
from typing import Any, cast
from pyelectroluxocp import OneAppApi
from pyelectroluxocp.apiModels import ApplienceStatusResponse
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .model import ElectroluxDevice
_LOGGER: logging.Logger = logging.getLogger(__package__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Configure entity platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
if appliances := coordinator.data.get("appliances", None):
for appliance_id, appliance in appliances.appliances.items():
entities = [
entity
for entity in appliance.entities
if entity.entity_type == "entity"
]
_LOGGER.debug(
"Electrolux add %d entities to registry for appliance %s",
len(entities),
appliance_id,
)
async_add_entities(entities)
class ElectroluxEntity(CoordinatorEntity):
"""Class for Electorolux devices."""
_attr_has_entity_name = True
appliance_status: ApplienceStatusResponse
def __init__(
self,
coordinator: Any,
name: str,
config_entry,
pnc_id: str,
entity_type: Platform,
entity_name,
entity_attr,
entity_source,
capability: dict[str, Any],
unit: str,
device_class: str,
entity_category: EntityCategory,
icon: str,
catalog_entry: ElectroluxDevice | None = None,
) -> None:
"""Initaliaze the entity."""
super().__init__(coordinator)
self.root_attribute = ["properties", "reported"]
self.data = None
self.coordinator = coordinator
self._cached_value = None
self._name = name
self._icon = icon
self._device_class = device_class
self._entity_category = entity_category
self._catalog_entry = catalog_entry
self.api: OneAppApi = coordinator.api
self.entity_name = entity_name
self.entity_attr = entity_attr
self.entity_type = entity_type
self.entity_source = entity_source
self.config_entry = config_entry
self.pnc_id = pnc_id
self.unit = unit
self.capability = capability
self.entity_id = f"{self.entity_domain}.{self.get_appliance.brand}_{self.get_appliance.name}_{self.entity_source}_{self.entity_attr}"
if catalog_entry:
self.entity_registry_enabled_default = (
catalog_entry.entity_registry_enabled_default
)
_LOGGER.debug("Electrolux new entity %s for appliance %s", name, pnc_id)
def setup(self, data):
"""Initialiaze setup."""
self.data = data
@property
def entity_domain(self) -> str:
"""Enitity domain for the entry."""
return "sensor"
@property
def unique_id(self) -> str:
"""Return a unique ID to use for this entity."""
return f"{self.config_entry.entry_id}-{self.entity_attr}-{self.entity_source or 'root'}-{self.pnc_id}"
# Disabled this as this removes the value from display : there is no readonly property for entities
# @property
# def available(self) -> bool:
# if (self._entity_category == EntityCategory.DIAGNOSTIC
# or self.entity_attr in ALWAYS_ENABLED_ATTRIBUTES):
# return True
# connection_state = self.get_connection_state()
# if connection_state and connection_state != "disconnected":
# return True
# return False
@property
def should_poll(self) -> bool:
"""Confirm if device should be polled."""
return False
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
# _LOGGER.debug("Electrolux entity got data %s", self.coordinator.data)
if self.coordinator.data is None:
return
appliances = self.coordinator.data.get("appliances", None)
self.appliance_status = appliances.get_appliance(self.pnc_id).state
self.async_write_ha_state()
def get_connection_state(self) -> str | None:
"""Return connection state."""
if self.appliance_status:
return self.appliance_status.get("connectionState", None)
return None
def get_state_attr(self, path: str) -> str | None:
"""Return value of other appliance attributes.
Used for the evaluation of state_mapping one property to another.
"""
if "/" in path:
if self.reported_state.get(path, None):
return self.reported_state.get(path)
source, attr = path.split("/")
return self.reported_state.get(source, {}).get(attr, None)
return self.reported_state.get(path, None)
@property
def reported_state(self):
"""Return reported state of the appliance."""
return self.appliance_status.get("properties", {}).get("reported", {})
@property
def name(self) -> str:
"""Return the name of the sensor."""
if self.catalog_entry and self.catalog_entry.friendly_name:
return self.catalog_entry.friendly_name.capitalize()
return self._name
@property
def icon(self) -> str | None:
"""Return the icon of the entity."""
return self._icon
# @property
# def get_entity(self) -> ApplianceEntity:
# return self.get_appliance.get_entity(self.entity_type, self.entity_attr, self.entity_source, None)
@property
def get_appliance(self):
"""Return the appliance device."""
return self.coordinator.data["appliances"].get_appliance(self.pnc_id)
@property
def device_info(self):
"""Return identifiers of the device."""
return {
"identifiers": {(DOMAIN, self.pnc_id)},
"name": self.get_appliance.name,
"model": self.get_appliance.model,
"manufacturer": self.get_appliance.brand,
}
@property
def entity_category(self) -> EntityCategory | None:
"""Return entity category."""
return self._entity_category
@property
def device_class(self):
"""Return the device class of the sensor."""
return self._device_class
def extract_value(self):
"""Return the appliance attributes of the entity."""
root_attribute = self.root_attribute
attribute = self.entity_attr
if self.appliance_status:
root = cast(any, self.appliance_status)
# Format returned by push is slightly different from format returned by API : fields are at root level
# Let's check if we can find the fields at root first
if (
self.entity_source and root.get(self.entity_source, None) is not None
) or root.get(attribute, None):
root_attribute = None
if root_attribute:
for item in root_attribute:
if root:
root = root.get(item, None)
if root:
if self.entity_source:
category: dict[str, Any] | None = root.get(self.entity_source, None)
if category:
return category.get(attribute)
else:
return root.get(attribute, None)
return None
def update(self, appliance_status: ApplienceStatusResponse):
"""Update the appliance status."""
self.appliance_status = appliance_status
# if self.hass:
# self.async_write_ha_state()
@property
def json_path(self) -> str | None:
"""Return the path to the entry."""
if self.entity_source:
return f"{self.entity_source}/{self.entity_attr}"
return self.entity_attr
@property
def catalog_entry(self) -> ElectroluxDevice | None:
"""Return matched catalog entry."""
return self._catalog_entry
# @property
# def extra_state_attributes(self) -> dict[str, Any]:
# """Return the state attributes of the sensor."""
# return {
# "Path": self.json_path,
# "entity_type": str(self.entity_type),
# "entity_category": str(self.entity_category),
# "device_class": str(self.device_class),
# "capability": str(self.capability),
# }
@@ -1,19 +0,0 @@
{
"domain": "electrolux_status",
"name": "Electrolux Status",
"codeowners": ["@albaintor","@kingy444"],
"config_flow": true,
"documentation": "https://github.com/albaintor/homeassistant_electrolux_status",
"homekit": {},
"integration_type": "hub",
"iot_class": "cloud_push",
"issue_tracker": "https://github.com/albaintor/homeassistant_electrolux_status/issues",
"loggers": ["pyelectroluxocp"],
"requirements": [
"pyelectroluxocp==0.1.3",
"aiofiles==24.1.0"
],
"ssdp": [],
"version": "2.2.0",
"zeroconf": []
}
@@ -1,128 +0,0 @@
"""Switch platform for Electrolux Status."""
import contextlib
import logging
from typing import Any
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTemperature, UnitOfTime, UnitOfVolume
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, SENSOR
from .entity import ElectroluxEntity
from .util import create_notification
_LOGGER: logging.Logger = logging.getLogger(__package__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Configure sensor platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
if appliances := coordinator.data.get("appliances", None):
for appliance_id, appliance in appliances.appliances.items():
entities = [
entity for entity in appliance.entities if entity.entity_type == SENSOR
]
_LOGGER.debug(
"Electrolux add %d SENSOR entities to registry for appliance %s",
len(entities),
appliance_id,
)
async_add_entities(entities)
class ElectroluxSensor(ElectroluxEntity, SensorEntity):
"""Electrolux Status Sensor class."""
@property
def entity_domain(self):
"""Enitity domain for the entry. Used for consistent entity_id."""
return SENSOR
@property
def suggested_display_precision(self) -> int | None:
"""Get the display precision."""
if self.unit == UnitOfTemperature.CELSIUS:
return 2
if self.unit == UnitOfTemperature.FAHRENHEIT:
return 2
if self.unit == UnitOfVolume.LITERS:
return 0
if self.unit == UnitOfTime.SECONDS:
return 0
return None
@property
def native_value(self) -> str | int | float:
"""Return the state of the sensor."""
value = self.extract_value()
if self.capability.get("access") == "constant":
value = self.capability.get("default")
elif self.entity_attr == "alerts":
value = len(value) if value is not None else 0
elif value is not None and isinstance(self.unit, UnitOfTime):
# Electrolux bug - prevent negative/disabled timers
value = max(value, 0)
if self.catalog_entry and self.catalog_entry.value_mapping:
# Electrolux presents as string but returns an int
# the mapping entry allows us to correctly display this to the frontend
mapping = self.catalog_entry.value_mapping
_LOGGER.debug("Mapping %s: %s to %s", self.json_path, value, mapping)
if value in mapping:
value = mapping.get(value, value)
if isinstance(value, str):
if "_" in value:
value = value.replace("_", " ")
value = value.title()
if value is not None:
self._cached_value = value
else:
value = self._cached_value
return value
@property
def native_unit_of_measurement(self) -> str | None:
"""Return unit of measurement."""
# store the value of the sensor in the native format
return self.unit
@property
def suggested_unit_of_measurement(self) -> str | None:
"""Return suggested unit of measurement."""
# change the display unit in the HA frontend
if self.unit == UnitOfTime.SECONDS:
return UnitOfTime.MINUTES
with contextlib.suppress(ValueError):
if self._is_valid_suggested_unit(self.unit):
return self.unit
return None
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the sensor."""
if self.entity_attr == "alerts":
alert_types = self.capability.get("values", {})
# default is nullable - set a value for display to user
alert_types = {key: "OFF" for key in alert_types}
if current_alerts := self.extract_value():
for alert in current_alerts:
name = alert.get("code", "Unknown")
severity = alert.get("severity", "Alert")
status = alert.get("acknowledgeStatus", "")
alert_types[name] = f"{severity}-{status}"
create_notification(
self.hass,
self.config_entry,
alert_name=name,
alert_severity=severity,
alert_status=status,
title=self.name,
)
return alert_types
return {}
@@ -1,41 +0,0 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"already_configured_account": "Account is already configured",
"reconfigure_successful": "Re-configuration was successful",
"single_instance_allowed": "Already configured. Only a single configuration possible."
},
"error": {
"invalid_auth": "Invalid authentication"
},
"step": {
"user": {
"data": {
"language": "Language",
"notifications": "Raise notifications for ALERT level notices",
"notifications_diagnostic": "Raise notifications for DIAGNOSTIC level notices",
"notifications_warning": "Raise notifications for WARNING level notices",
"password": "Password",
"username": "Username"
},
"description": "If you need help with the configuration visit: https://github.com/albaintor/homeassistant_electrolux_status",
"title": "Electrolux Status"
}
}
},
"options": {
"step": {
"user": {
"data": {
"language": "Language",
"notifications": "Raise notifications for ALERT level notices",
"notifications_diagnostic": "Raise notifications for DIAGNOSTIC level notices",
"notifications_warning": "Raise notifications for WARNING level notices",
"password": "Password",
"renew_interval": "Renewal interval of websocket (seconds)"
}
}
}
}
}
@@ -1,41 +0,0 @@
{
"config": {
"abort": {
"already_configured": "Il dispositivo è già configurato",
"already_configured_account": "L'account è già configurato",
"reconfigure_successful": "La riconfigurazione ha avuto successo",
"single_instance_allowed": "Già configurato.Solo una singola configurazione possibile."
},
"error": {
"invalid_auth": "Autenticazione non valida"
},
"step": {
"user": {
"data": {
"language": "Lingua",
"notifications": "Aumenta le notifiche per gli avvisi di livello di allerta",
"notifications_diagnostic": "Aumenta le notifiche per gli avvisi a livello diagnostico",
"notifications_warning": "Aumenta le notifiche per gli avvisi di livello di avvertimento",
"password": "Password",
"username": "Nome utente"
},
"description": "Se hai bisogno di aiuto con la visita di configurazione: https://github.com/albaintor/homeeasistant_electrolux_status",
"title": "Stato Electrolux"
}
}
},
"options": {
"step": {
"user": {
"data": {
"language": "Lingua",
"notifications": "Aumenta le notifiche per gli avvisi di livello di allerta",
"notifications_diagnostic": "Aumenta le notifiche per gli avvisi a livello diagnostico",
"notifications_warning": "Aumenta le notifiche per gli avvisi di livello di avvertimento",
"password": "Password",
"renew_interval": "Intervallo di rinnovo di WebSocket (secondi)"
}
}
}
}
}
@@ -1,41 +0,0 @@
{
"config": {
"abort": {
"already_configured": "Urządzenie jest już skonfigurowane",
"already_configured_account": "Konto jest już skonfigurowane",
"reconfigure_successful": "Ponowna konfiguracja zakończyła się powodzeniem",
"single_instance_allowed": "Już skonfigurowano. Możliwa tylko jedna konfiguracja."
},
"error": {
"invalid_auth": "Nieprawidłowe uwierzytelnienie"
},
"step": {
"user": {
"data": {
"language": "Język",
"notifications": "Pokazuj alerty",
"notifications_diagnostic": "Pokazuj powoadomienia diagnostyczne",
"notifications_warning": "Pokazuj powiadomienia ostrzegawcze",
"password": "Hasło",
"username": "Nazwa użytkownika"
},
"description": "Jeśli potrzebujesz pomocy w konfiguracji, odwiedź: https://github.com/albaintor/homeassistant_electrolux_status",
"title": "Electrolux Status"
}
}
},
"options": {
"step": {
"user": {
"data": {
"language": "Język",
"notifications": "Pokazuj alerty",
"notifications_diagnostic": "Pokazuj powoadomienia diagnostyczne",
"notifications_warning": "Pokazuj powiadomienia ostrzegawcze",
"password": "Hasło",
"renew_interval": "Okres odnawiania połączenia WebSocket (sekundy)"
}
}
}
}
}
@@ -1,41 +0,0 @@
{
"config": {
"abort": {
"already_configured": "Zariadenie je už nakonfigurované",
"already_configured_account": "Účet je už nakonfigurovaný",
"reconfigure_successful": "Opätovná konfigurácia bola úspešná",
"single_instance_allowed": "Už nakonfigurované.Iba jedna možná konfigurácia."
},
"error": {
"invalid_auth": "Neplatné overovanie"
},
"step": {
"user": {
"data": {
"language": "Jazyk",
"notifications": "Zvýšiť oznámenia o upozorneniach na úrovni výstrahy",
"notifications_diagnostic": "Zvýšte upozornenia pre oznámenia o diagnostickej úrovni",
"notifications_warning": "Zvýšte oznámenia o upozorneniach na úrovni varovania",
"password": "Heslo",
"username": "Užívateľské meno"
},
"description": "Ak potrebujete pomoc s konfiguráciou, navštívte: https://github.com/albaintor/homeassistant_electrolux_status",
"title": "Status"
}
}
},
"options": {
"step": {
"user": {
"data": {
"language": "Jazyk",
"notifications": "Zvýšiť oznámenia o upozorneniach na úrovni výstrahy",
"notifications_diagnostic": "Zvýšte upozornenia pre oznámenia o diagnostickej úrovni",
"notifications_warning": "Zvýšte oznámenia o upozorneniach na úrovni varovania",
"password": "Heslo",
"renew_interval": "Interval obnovy WebSocket (sekundy)"
}
}
}
}
}