Compare commits

..

3 Commits

Author SHA1 Message Date
Daniel Hjelseth Høyer
d2a55dec55 Add Tibber binary sensors
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-06 18:29:57 +01:00
Daniel Hjelseth Høyer
cddc4bdf8f Add Tibber binary sensors
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-06 18:16:48 +01:00
Daniel Hjelseth Høyer
adc201bb4e Add Tibber binary sensors
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-06 14:34:46 +01:00
37 changed files with 389 additions and 747 deletions

View File

@@ -7,7 +7,7 @@
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["pybravia"],
"requirements": ["pybravia==0.4.1"],
"requirements": ["pybravia==0.3.4"],
"ssdp": [
{
"manufacturer": "Sony Corporation",

View File

@@ -23,5 +23,5 @@
"winter_mode": {}
},
"quality_scale": "internal",
"requirements": ["home-assistant-frontend==20251229.1"]
"requirements": ["home-assistant-frontend==20251229.0"]
}

View File

@@ -1,51 +0,0 @@
"""The Homevolt integration."""
from __future__ import annotations
from homevolt import Homevolt, HomevoltConnectionError
from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, HomevoltConfigEntry
from .coordinator import HomevoltDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup_entry(hass: HomeAssistant, entry: HomevoltConfigEntry) -> bool:
"""Set up Homevolt from a config entry."""
host: str = entry.data[CONF_HOST]
password: str | None = entry.data.get(CONF_PASSWORD)
websession = async_get_clientsession(hass)
client = Homevolt(host, password, websession=websession)
try:
await client.update_info()
except HomevoltConnectionError as err:
raise ConfigEntryNotReady(
f"Unable to connect to Homevolt battery: {err}"
) from err
coordinator = HomevoltDataUpdateCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: HomevoltConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
if entry.runtime_data:
await entry.runtime_data.client.close_connection()
return unload_ok

View File

@@ -1,63 +0,0 @@
"""Config flow for the Homevolt integration."""
from __future__ import annotations
import logging
from typing import Any
from homevolt import Homevolt, HomevoltAuthenticationError, HomevoltConnectionError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_PASSWORD): str,
}
)
class HomevoltConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Homevolt."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
password = user_input.get(CONF_PASSWORD)
websession = async_get_clientsession(self.hass)
try:
await Homevolt(host, password, websession=websession).update_info()
except HomevoltAuthenticationError:
errors["base"] = "invalid_auth"
except HomevoltConnectionError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001
errors["base"] = "unknown"
else:
await self.async_set_unique_id(host)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="Homevolt Local",
data={
CONF_HOST: host,
CONF_PASSWORD: password,
},
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@@ -1,16 +0,0 @@
"""Constants for the Homevolt integration."""
from __future__ import annotations
from datetime import timedelta
from typing import TYPE_CHECKING
from homeassistant.config_entries import ConfigEntry
if TYPE_CHECKING:
from .coordinator import HomevoltDataUpdateCoordinator
DOMAIN = "homevolt"
MANUFACTURER = "Homevolt"
SCAN_INTERVAL = timedelta(seconds=15)
type HomevoltConfigEntry = ConfigEntry["HomevoltDataUpdateCoordinator"]

View File

@@ -1,51 +0,0 @@
"""Data update coordinator for Homevolt integration."""
from __future__ import annotations
import logging
from homevolt import (
Device,
Homevolt,
HomevoltAuthenticationError,
HomevoltConnectionError,
HomevoltError,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, SCAN_INTERVAL, HomevoltConfigEntry
_LOGGER = logging.getLogger(__name__)
class HomevoltDataUpdateCoordinator(DataUpdateCoordinator[Device]):
"""Class to manage fetching Homevolt data."""
def __init__(
self,
hass: HomeAssistant,
entry: HomevoltConfigEntry,
client: Homevolt,
) -> None:
"""Initialize the Homevolt coordinator."""
self.client = client
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
config_entry=entry,
)
async def _async_update_data(self) -> Device:
"""Fetch data from the Homevolt API."""
try:
await self.client.update_info()
return self.client.get_device()
except HomevoltAuthenticationError as err:
raise ConfigEntryAuthFailed from err
except (HomevoltConnectionError, HomevoltError) as err:
raise UpdateFailed(f"Error communicating with device: {err}") from err

View File

@@ -1,15 +0,0 @@
{
"domain": "homevolt",
"name": "Homevolt",
"codeowners": ["@danielhiversen"],
"config_flow": true,
"dependencies": [],
"documentation": "https://www.home-assistant.io/integrations/homevolt",
"homekit": {},
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["homevolt==0.2.3"],
"ssdp": [],
"zeroconf": []
}

View File

@@ -1,72 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register custom actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: Local_polling without events
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: Integration does not register custom actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: Integration does not have an options flow.
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates:
status: exempt
comment: Coordinator handles updates, no explicit parallel updates needed.
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -1,167 +0,0 @@
"""Support for Homevolt sensors."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from homevolt.models import SensorType
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfFrequency,
UnitOfPower,
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 .const import DOMAIN, MANUFACTURER, HomevoltConfigEntry
from .coordinator import HomevoltDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@dataclass
class Description:
"""Sensor metadata description."""
device_class: SensorDeviceClass | None
state_class: SensorStateClass | None
native_unit_of_measurement: str | None
SENSOR_META: dict[SensorType, Description] = {
SensorType.COUNT: Description(
None,
SensorStateClass.MEASUREMENT,
"N",
),
SensorType.CURRENT: Description(
SensorDeviceClass.CURRENT,
SensorStateClass.MEASUREMENT,
UnitOfElectricCurrent.AMPERE,
),
SensorType.ENERGY_INCREASING: Description(
SensorDeviceClass.ENERGY,
SensorStateClass.TOTAL_INCREASING,
UnitOfEnergy.KILO_WATT_HOUR,
),
SensorType.ENERGY_TOTAL: Description(
SensorDeviceClass.ENERGY,
SensorStateClass.TOTAL,
UnitOfEnergy.WATT_HOUR,
),
SensorType.FREQUENCY: Description(
SensorDeviceClass.FREQUENCY,
SensorStateClass.MEASUREMENT,
UnitOfFrequency.HERTZ,
),
SensorType.PERCENTAGE: Description(
SensorDeviceClass.BATTERY,
SensorStateClass.MEASUREMENT,
PERCENTAGE,
),
SensorType.POWER: Description(
SensorDeviceClass.POWER,
SensorStateClass.MEASUREMENT,
UnitOfPower.WATT,
),
SensorType.SCHEDULE_TYPE: Description(
None,
None,
None,
),
SensorType.SIGNAL_STRENGTH: Description(
SensorDeviceClass.SIGNAL_STRENGTH,
SensorStateClass.MEASUREMENT,
SIGNAL_STRENGTH_DECIBELS,
),
SensorType.TEMPERATURE: Description(
SensorDeviceClass.TEMPERATURE,
SensorStateClass.MEASUREMENT,
UnitOfTemperature.CELSIUS,
),
SensorType.TEXT: Description(
None,
None,
None,
),
SensorType.VOLTAGE: Description(
SensorDeviceClass.VOLTAGE,
SensorStateClass.MEASUREMENT,
UnitOfElectricPotential.VOLT,
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: HomevoltConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Homevolt sensor."""
coordinator = entry.runtime_data
entities = []
for sensor_name, sensor in coordinator.data.sensors.items():
if sensor.type not in SENSOR_META:
continue
sensor_meta = SENSOR_META[sensor.type]
entities.append(
HomevoltSensor(
SensorEntityDescription(
key=sensor_name,
name=sensor_name,
device_class=sensor_meta.device_class,
state_class=sensor_meta.state_class,
native_unit_of_measurement=sensor_meta.native_unit_of_measurement,
),
coordinator,
)
)
async_add_entities(entities)
class HomevoltSensor(CoordinatorEntity[HomevoltDataUpdateCoordinator], SensorEntity):
"""Representation of a Homevolt sensor."""
_attr_has_entity_name = True
def __init__(
self,
description: SensorEntityDescription,
coordinator: HomevoltDataUpdateCoordinator,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
device_id = coordinator.data.device_id
self._attr_unique_id = f"{device_id}_{description.key}"
sensor = coordinator.data.sensors[description.key]
sensor_device_id = sensor.device_identifier
device_metadata = coordinator.data.device_metadata.get(sensor_device_id)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{device_id}_{sensor_device_id}")},
configuration_url=coordinator.client.hostname,
manufacturer=MANUFACTURER,
model=device_metadata.model,
name=device_metadata.name,
)
@property
def native_value(self) -> StateType:
"""Return the native value of the sensor."""
return self.coordinator.data.sensors[self.entity_description.key].value

View File

@@ -1,26 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"host": "The IP address or hostname of your Homevolt battery on your local network.",
"password": "The local password configured for your Homevolt battery, if required."
},
"description": "Connect Home Assistant to your Homevolt battery over the local network.",
"title": "Homevolt Local"
}
}
}
}

View File

@@ -7,6 +7,6 @@
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["pysaunum"],
"quality_scale": "gold",
"quality_scale": "silver",
"requirements": ["pysaunum==0.1.0"]
}

View File

@@ -49,7 +49,7 @@ rules:
status: exempt
comment: Device cannot be discovered and the Modbus TCP API does not provide MAC address or other unique network identifiers needed to update connection information.
docs-data-update: done
docs-examples: done
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done

View File

@@ -46,7 +46,7 @@ SENSOR_TYPES = [
key="lifetime_energy",
json_key="lifeTimeData",
translation_key="lifetime_energy",
state_class=SensorStateClass.TOTAL_INCREASING,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
),
@@ -55,7 +55,6 @@ SENSOR_TYPES = [
json_key="lastYearData",
translation_key="energy_this_year",
entity_registry_enabled_default=False,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
),
@@ -64,7 +63,6 @@ SENSOR_TYPES = [
json_key="lastMonthData",
translation_key="energy_this_month",
entity_registry_enabled_default=False,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
),
@@ -73,7 +71,6 @@ SENSOR_TYPES = [
json_key="lastDayData",
translation_key="energy_today",
entity_registry_enabled_default=False,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
),
@@ -126,32 +123,24 @@ SENSOR_TYPES = [
json_key="LOAD",
translation_key="power_consumption",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
),
SolarEdgeSensorEntityDescription(
key="solar_power",
json_key="PV",
translation_key="solar_power",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
),
SolarEdgeSensorEntityDescription(
key="grid_power",
json_key="GRID",
translation_key="grid_power",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
),
SolarEdgeSensorEntityDescription(
key="storage_power",
json_key="STORAGE",
translation_key="storage_power",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER,
),
SolarEdgeSensorEntityDescription(
key="purchased_energy",
@@ -205,7 +194,6 @@ SENSOR_TYPES = [
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
),
]

View File

@@ -33,7 +33,7 @@ from .const import (
from .coordinator import TibberDataAPICoordinator
from .services import async_setup_services
PLATFORMS = [Platform.NOTIFY, Platform.SENSOR]
PLATFORMS = [Platform.BINARY_SENSOR, Platform.NOTIFY, Platform.SENSOR]
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)

View File

@@ -0,0 +1,138 @@
"""Support for Tibber binary sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging
from tibber.data_api import TibberDevice
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
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 .const import DOMAIN, TibberConfigEntry
from .coordinator import TibberDataAPICoordinator
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True)
class TibberBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes Tibber binary sensor entity."""
is_on_fn: Callable[[str | None], bool | None]
def _connector_status_is_on(value: str | None) -> bool | None:
"""Map connector status value to binary sensor state."""
if value == "connected":
return True
if value == "disconnected":
return False
return None
def _charging_status_is_on(value: str | None) -> bool | None:
"""Map charging status value to binary sensor state."""
if value == "charging":
return True
if value == "idle":
return False
return None
def _device_status_is_on(value: str | None) -> bool | None:
"""Map device status value to binary sensor state."""
if value == "on":
return True
if value == "off":
return False
return None
DATA_API_BINARY_SENSORS: tuple[TibberBinarySensorEntityDescription, ...] = (
TibberBinarySensorEntityDescription(
key="connector.status",
device_class=BinarySensorDeviceClass.PLUG,
is_on_fn=_connector_status_is_on,
),
TibberBinarySensorEntityDescription(
key="charging.status",
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
is_on_fn=_charging_status_is_on,
),
TibberBinarySensorEntityDescription(
key="onOff",
device_class=BinarySensorDeviceClass.POWER,
is_on_fn=_device_status_is_on,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: TibberConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Tibber binary sensors."""
coordinator = entry.runtime_data.data_api_coordinator
assert coordinator is not None
entities: list[TibberDataAPIBinarySensor] = []
api_binary_sensors = {sensor.key: sensor for sensor in DATA_API_BINARY_SENSORS}
for device in coordinator.data.values():
for sensor in device.sensors:
description: TibberBinarySensorEntityDescription | None = (
api_binary_sensors.get(sensor.id)
)
if description is None:
continue
entities.append(TibberDataAPIBinarySensor(coordinator, device, description))
async_add_entities(entities)
class TibberDataAPIBinarySensor(
CoordinatorEntity[TibberDataAPICoordinator], BinarySensorEntity
):
"""Representation of a Tibber Data API binary sensor."""
_attr_has_entity_name = True
entity_description: TibberBinarySensorEntityDescription
def __init__(
self,
coordinator: TibberDataAPICoordinator,
device: TibberDevice,
entity_description: TibberBinarySensorEntityDescription,
) -> None:
"""Initialize the binary sensor."""
super().__init__(coordinator)
self._device_id: str = device.id
self.entity_description = entity_description
self._attr_unique_id = f"{device.external_id}_{self.entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.external_id)},
name=device.name,
manufacturer=device.brand,
model=device.model,
)
@property
def is_on(self) -> bool | None:
"""Return the state of the binary sensor."""
sensors = self.coordinator.sensors_by_device.get(self._device_id, {})
sensor = sensors[self.entity_description.key]
value: str | None = str(sensor.value) if sensor.value is not None else None
return self.entity_description.is_on_fn(value)

View File

@@ -430,9 +430,6 @@ def _setup_data_api_sensors(
for sensor in device.sensors:
description: SensorEntityDescription | None = api_sensors.get(sensor.id)
if description is None:
_LOGGER.debug(
"Sensor %s not found in DATA_API_SENSORS, skipping", sensor
)
continue
entities.append(TibberDataAPISensor(coordinator, device, description))
async_add_entities(entities)

View File

@@ -183,12 +183,13 @@ class TuyaAlarmEntity(TuyaEntity, AlarmControlPanelEntity):
self._state_wrapper = state_wrapper
# Determine supported modes
if "arm_home" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
if "arm_away" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
if "trigger" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.TRIGGER
if action_wrapper.options:
if "arm_home" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
if "arm_away" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
if "trigger" in action_wrapper.options:
self._attr_supported_features |= AlarmControlPanelEntityFeature.TRIGGER
@property
def alarm_state(self) -> AlarmControlPanelState | None:

View File

@@ -368,7 +368,7 @@ class TuyaClimateEntity(TuyaEntity, ClimateEntity):
# Determine HVAC modes
self._attr_hvac_modes: list[HVACMode] = []
self._hvac_to_tuya = {}
if hvac_mode_wrapper:
if hvac_mode_wrapper and hvac_mode_wrapper.options is not None:
self._attr_hvac_modes = [HVACMode.OFF]
unknown_hvac_modes: list[str] = []
for tuya_mode in hvac_mode_wrapper.options:

View File

@@ -351,7 +351,7 @@ class TuyaCoverEntity(TuyaEntity, CoverEntity):
self._set_position = set_position
self._tilt_position = tilt_position
if instruction_wrapper:
if instruction_wrapper and instruction_wrapper.options:
if "open" in instruction_wrapper.options:
self._attr_supported_features |= CoverEntityFeature.OPEN
if "close" in instruction_wrapper.options:
@@ -424,7 +424,11 @@ class TuyaCoverEntity(TuyaEntity, CoverEntity):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
if self._instruction_wrapper and "stop" in self._instruction_wrapper.options:
if (
self._instruction_wrapper
and (options := self._instruction_wrapper.options)
and "stop" in options
):
await self._async_send_wrapper_updates(self._instruction_wrapper, "stop")
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:

View File

@@ -21,7 +21,6 @@ from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
from .entity import TuyaEntity
from .models import (
DeviceWrapper,
DPCodeEnumWrapper,
DPCodeRawWrapper,
DPCodeStringWrapper,
@@ -29,58 +28,75 @@ from .models import (
)
class _EventEnumWrapper(DPCodeEnumWrapper):
"""Wrapper for event enum DP codes."""
class _DPCodeEventWrapper(DPCodeTypeInformationWrapper):
"""Base class for Tuya event wrappers."""
def read_device_status(self, device: CustomerDevice) -> tuple[str, None] | None:
"""Return the event details."""
if (raw_value := super().read_device_status(device)) is None:
return None
return (raw_value, None)
class _AlarmMessageWrapper(DPCodeStringWrapper):
"""Wrapper for a STRING message on DPCode.ALARM_MESSAGE."""
options: list[str]
def __init__(self, dpcode: str, type_information: Any) -> None:
"""Init _AlarmMessageWrapper."""
"""Init _DPCodeEventWrapper."""
super().__init__(dpcode, type_information)
self.options = ["triggered"]
def read_device_status(
self, device: CustomerDevice
) -> tuple[str, dict[str, Any]] | None:
"""Return the event attributes for the alarm message."""
if (raw_value := super().read_device_status(device)) is None:
def get_event_type(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> str | None:
"""Return the event type."""
if (
updated_status_properties is None
or self.dpcode not in updated_status_properties
):
return None
return ("triggered", {"message": b64decode(raw_value).decode("utf-8")})
return "triggered"
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes."""
return None
class _DoorbellPicWrapper(DPCodeRawWrapper):
class _EventEnumWrapper(DPCodeEnumWrapper, _DPCodeEventWrapper):
"""Wrapper for event enum DP codes."""
def get_event_type(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> str | None:
"""Return the triggered event type."""
if (
updated_status_properties is None
or self.dpcode not in updated_status_properties
):
return None
return self.read_device_status(device)
class _AlarmMessageWrapper(DPCodeStringWrapper, _DPCodeEventWrapper):
"""Wrapper for a STRING message on DPCode.ALARM_MESSAGE."""
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes for the alarm message."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
return {"message": b64decode(raw_value).decode("utf-8")}
class _DoorbellPicWrapper(DPCodeRawWrapper, _DPCodeEventWrapper):
"""Wrapper for a RAW message on DPCode.DOORBELL_PIC.
It is expected that the RAW data is base64/utf8 encoded URL of the picture.
"""
def __init__(self, dpcode: str, type_information: Any) -> None:
"""Init _DoorbellPicWrapper."""
super().__init__(dpcode, type_information)
self.options = ["triggered"]
def read_device_status(
self, device: CustomerDevice
) -> tuple[str, dict[str, Any]] | None:
def get_event_attributes(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Return the event attributes for the doorbell picture."""
if (status := super().read_device_status(device)) is None:
return None
return ("triggered", {"message": status.decode("utf-8")})
return {"message": status.decode("utf-8")}
@dataclass(frozen=True)
class TuyaEventEntityDescription(EventEntityDescription):
"""Describe a Tuya Event entity."""
wrapper_class: type[DPCodeTypeInformationWrapper] = _EventEnumWrapper
wrapper_class: type[_DPCodeEventWrapper] = _EventEnumWrapper
# All descriptions can be found here. Mostly the Enum data types in the
@@ -206,7 +222,7 @@ class TuyaEventEntity(TuyaEntity, EventEntity):
device: CustomerDevice,
device_manager: Manager,
description: EventEntityDescription,
dpcode_wrapper: DeviceWrapper[tuple[str, dict[str, Any] | None]],
dpcode_wrapper: _DPCodeEventWrapper,
) -> None:
"""Init Tuya event entity."""
super().__init__(device, device_manager)
@@ -220,11 +236,15 @@ class TuyaEventEntity(TuyaEntity, EventEntity):
updated_status_properties: list[str] | None,
dp_timestamps: dict | None = None,
) -> None:
if self._dpcode_wrapper.skip_update(
self.device, updated_status_properties
) or not (event_data := self._dpcode_wrapper.read_device_status(self.device)):
if (
event_type := self._dpcode_wrapper.get_event_type(
self.device, updated_status_properties
)
) is None:
return
event_type, event_attributes = event_data
self._trigger_event(event_type, event_attributes)
self._trigger_event(
event_type,
self._dpcode_wrapper.get_event_attributes(self.device),
)
self.async_write_ha_state()

View File

@@ -198,9 +198,7 @@ class TuyaFanEntity(TuyaEntity, FanEntity):
if speed_wrapper:
self._attr_supported_features |= FanEntityFeature.SET_SPEED
# if speed is from an enum, set speed count from options
# else keep entity default 100
if hasattr(speed_wrapper, "options"):
if speed_wrapper.options is not None:
self._attr_speed_count = len(speed_wrapper.options)
if oscillate_wrapper:

View File

@@ -706,6 +706,7 @@ class TuyaLightEntity(TuyaEntity, LightEntity):
elif (
color_supported(color_modes)
and color_mode_wrapper is not None
and color_mode_wrapper.options
and WorkMode.WHITE in color_mode_wrapper.options
):
color_modes.add(ColorMode.WHITE)

View File

@@ -22,23 +22,13 @@ class DeviceWrapper[T]:
"""Base device wrapper."""
native_unit: str | None = None
options: list[str] | None = None
suggested_unit: str | None = None
max_value: float
min_value: float
value_step: float
options: list[str]
def skip_update(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> bool:
"""Determine if the wrapper should skip an update.
The default is to always skip, unless overridden in subclasses.
"""
return True
def read_device_status(self, device: CustomerDevice) -> T | None:
"""Read device status and convert to a Home Assistant value."""
raise NotImplementedError
@@ -61,19 +51,6 @@ class DPCodeWrapper(DeviceWrapper):
"""Init DPCodeWrapper."""
self.dpcode = dpcode
def skip_update(
self, device: CustomerDevice, updated_status_properties: list[str] | None
) -> bool:
"""Determine if the wrapper should skip an update.
By default, skip if updated_status_properties is given and
does not include this dpcode.
"""
return (
updated_status_properties is None
or self.dpcode not in updated_status_properties
)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value.
@@ -161,6 +138,7 @@ class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
"""Simple wrapper for EnumTypeInformation values."""
_DPTYPE = EnumTypeInformation
options: list[str]
def __init__(self, dpcode: str, type_information: EnumTypeInformation) -> None:
"""Init DPCodeEnumWrapper."""

View File

@@ -2,6 +2,8 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.select import SelectEntity, SelectEntityDescription
@@ -400,6 +402,8 @@ class TuyaSelectEntity(TuyaEntity, SelectEntity):
self.entity_description = description
self._attr_unique_id = f"{super().unique_id}{description.key}"
self._dpcode_wrapper = dpcode_wrapper
if TYPE_CHECKING:
assert dpcode_wrapper.options
self._attr_options = dpcode_wrapper.options
@property

View File

@@ -212,7 +212,7 @@ class TuyaVacuumEntity(TuyaEntity, StateVacuumEntity):
self._attr_fan_speed_list = []
self._attr_supported_features = VacuumEntityFeature.SEND_COMMAND
if action_wrapper:
if action_wrapper and action_wrapper.options:
if "pause" in action_wrapper.options:
self._attr_supported_features |= VacuumEntityFeature.PAUSE
if "return_to_base" in action_wrapper.options:
@@ -227,7 +227,7 @@ class TuyaVacuumEntity(TuyaEntity, StateVacuumEntity):
if activity_wrapper:
self._attr_supported_features |= VacuumEntityFeature.STATE
if fan_speed_wrapper:
if fan_speed_wrapper and fan_speed_wrapper.options:
self._attr_fan_speed_list = fan_speed_wrapper.options
self._attr_supported_features |= VacuumEntityFeature.FAN_SPEED

View File

@@ -23,7 +23,7 @@
"universal_silabs_flasher",
"serialx"
],
"requirements": ["zha==0.0.83", "serialx==0.5.0"],
"requirements": ["zha==0.0.82", "serialx==0.5.0"],
"usb": [
{
"description": "*2652*",

View File

@@ -293,7 +293,6 @@ FLOWS = {
"homekit",
"homekit_controller",
"homematicip_cloud",
"homevolt",
"homewizard",
"homeworks",
"honeywell",

View File

@@ -2807,12 +2807,6 @@
"config_flow": true,
"iot_class": "local_push"
},
"homevolt": {
"name": "Homevolt",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling"
},
"homematic": {
"name": "Homematic",
"integrations": {

View File

@@ -537,7 +537,7 @@ def _validate_range[_T: dict[str, Any]](
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA = vol.Schema(
{
vol.Required("active_choice"): vol.In(["number", "entity"]),
vol.Required("chosen_selector"): vol.In(["number", "entity"]),
vol.Optional("entity"): cv.entity_id,
vol.Optional("number"): vol.Coerce(float),
}
@@ -548,7 +548,7 @@ def _validate_number_or_entity(value: dict | float | str) -> float | str:
"""Validate number or entity selector result."""
if isinstance(value, dict):
_NUMBER_OR_ENTITY_CHOOSE_SCHEMA(value)
return value[value["active_choice"]] # type: ignore[no-any-return]
return value[value["chosen_selector"]] # type: ignore[no-any-return]
return value

View File

@@ -39,7 +39,7 @@ habluetooth==5.8.0
hass-nabucasa==1.7.0
hassil==3.5.0
home-assistant-bluetooth==1.13.1
home-assistant-frontend==20251229.1
home-assistant-frontend==20251229.0
home-assistant-intents==2026.1.1
httpx==0.28.1
ifaddr==0.2.0

9
requirements_all.txt generated
View File

@@ -1213,7 +1213,7 @@ hole==0.9.0
holidays==0.84
# homeassistant.components.frontend
home-assistant-frontend==20251229.1
home-assistant-frontend==20251229.0
# homeassistant.components.conversation
home-assistant-intents==2026.1.1
@@ -1224,9 +1224,6 @@ homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
# homeassistant.components.homevolt
homevolt==0.2.3
# homeassistant.components.horizon
horimote==0.4.1
@@ -1934,7 +1931,7 @@ pyblu==2.0.5
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.4.1
pybravia==0.3.4
# homeassistant.components.nissan_leaf
pycarwings2==2.14
@@ -3280,7 +3277,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==0.0.83
zha==0.0.82
# homeassistant.components.zhong_hong
zhong-hong-hvac==1.0.13

View File

@@ -1071,7 +1071,7 @@ hole==0.9.0
holidays==0.84
# homeassistant.components.frontend
home-assistant-frontend==20251229.1
home-assistant-frontend==20251229.0
# homeassistant.components.conversation
home-assistant-intents==2026.1.1
@@ -1082,9 +1082,6 @@ homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
# homeassistant.components.homevolt
homevolt==0.2.3
# homeassistant.components.remember_the_milk
httplib2==0.20.4
@@ -1656,7 +1653,7 @@ pyblu==2.0.5
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.4.1
pybravia==0.3.4
# homeassistant.components.cloudflare
pycfdns==3.0.0
@@ -2744,7 +2741,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==0.0.83
zha==0.0.82
# homeassistant.components.zwave_js
zwave-js-server-python==0.67.1

View File

@@ -1 +0,0 @@
"""Tests for the Homevolt integration."""

View File

@@ -1,15 +0,0 @@
"""Common fixtures for the Homevolt tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.homevolt.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry

View File

@@ -1,147 +0,0 @@
"""Tests for the Homevolt config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
from homevolt import HomevoltAuthenticationError, HomevoltConnectionError
import pytest
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_full_flow_success(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test a complete successful user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.100",
CONF_PASSWORD: "test-password",
}
with patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Homevolt Local"
assert result["data"] == user_input
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(HomevoltAuthenticationError, "invalid_auth"),
(HomevoltConnectionError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_step_user_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
exception: Exception,
expected_error: str,
) -> None:
"""Test error cases for the user step with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.100",
CONF_PASSWORD: "test-password",
}
with patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
) as mock_update_info:
mock_update_info.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
# Clear the error and complete the flow successfully
with patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Homevolt Local"
assert result["data"] == user_input
assert len(mock_setup_entry.mock_calls) == 1
async def test_duplicate_entry(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test that a duplicate host aborts the flow."""
existing_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100", CONF_PASSWORD: "test-password"},
unique_id="192.168.1.100",
)
existing_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.100",
CONF_PASSWORD: "test-password",
}
with patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -0,0 +1,150 @@
"""Tests for the Tibber binary sensors."""
from __future__ import annotations
from unittest.mock import AsyncMock
import tibber
from homeassistant.components.recorder import Recorder
from homeassistant.components.tibber.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry
def create_tibber_device_with_binary_sensors(
device_id: str = "device-id",
external_id: str = "external-id",
name: str = "Test Device",
brand: str = "Tibber",
model: str = "Gen1",
connector_status: str | None = "connected",
charging_status: str | None = "charging",
device_status: str | None = "on",
home_id: str = "home-id",
) -> tibber.data_api.TibberDevice:
"""Create a fake Tibber Data API device with binary sensor capabilities."""
device_data = {
"id": device_id,
"externalId": external_id,
"info": {
"name": name,
"brand": brand,
"model": model,
},
"capabilities": [
{
"id": "connector.status",
"value": connector_status,
"description": "Connector status",
"unit": "",
},
{
"id": "charging.status",
"value": charging_status,
"description": "Charging status",
"unit": "",
},
{
"id": "onOff",
"value": device_status,
"description": "Device status",
"unit": "",
},
],
}
return tibber.data_api.TibberDevice(device_data, home_id=home_id)
async def test_binary_sensors_are_created(
recorder_mock: Recorder,
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
setup_credentials: None,
entity_registry: er.EntityRegistry,
) -> None:
"""Ensure binary sensors are created from Data API devices."""
device = create_tibber_device_with_binary_sensors()
data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
connector_unique_id = "external-id_connector.status"
connector_entity_id = entity_registry.async_get_entity_id(
"binary_sensor", DOMAIN, connector_unique_id
)
assert connector_entity_id is not None
state = hass.states.get(connector_entity_id)
assert state is not None
assert state.state == "on"
charging_unique_id = "external-id_charging.status"
charging_entity_id = entity_registry.async_get_entity_id(
"binary_sensor", DOMAIN, charging_unique_id
)
assert charging_entity_id is not None
state = hass.states.get(charging_entity_id)
assert state is not None
assert state.state == "on"
device_unique_id = "external-id_onOff"
device_entity_id = entity_registry.async_get_entity_id(
"binary_sensor", DOMAIN, device_unique_id
)
assert device_entity_id is not None
state = hass.states.get(device_entity_id)
assert state is not None
assert state.state == "on"
async def test_device_status_on(
recorder_mock: Recorder,
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
setup_credentials: None,
entity_registry: er.EntityRegistry,
) -> None:
"""Test device status on state."""
device = create_tibber_device_with_binary_sensors(device_status="on")
data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
unique_id = "external-id_onOff"
entity_id = entity_registry.async_get_entity_id("binary_sensor", DOMAIN, unique_id)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "on"
async def test_device_status_off(
recorder_mock: Recorder,
hass: HomeAssistant,
config_entry: MockConfigEntry,
data_api_client_mock: AsyncMock,
setup_credentials: None,
entity_registry: er.EntityRegistry,
) -> None:
"""Test device status off state."""
device = create_tibber_device_with_binary_sensors(device_status="off")
data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
unique_id = "external-id_onOff"
entity_id = entity_registry.async_get_entity_id("binary_sensor", DOMAIN, unique_id)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "off"

View File

@@ -1207,19 +1207,19 @@ async def test_subscribe_triggers_no_triggers(
),
# Test verbose choose selector options
(
{CONF_ABOVE: {"active_choice": "entity", "entity": "sensor.test"}},
{CONF_ABOVE: {"chosen_selector": "entity", "entity": "sensor.test"}},
does_not_raise(),
),
(
{CONF_ABOVE: {"active_choice": "number", "number": 10}},
{CONF_ABOVE: {"chosen_selector": "number", "number": 10}},
does_not_raise(),
),
(
{CONF_BELOW: {"active_choice": "entity", "entity": "sensor.test"}},
{CONF_BELOW: {"chosen_selector": "entity", "entity": "sensor.test"}},
does_not_raise(),
),
(
{CONF_BELOW: {"active_choice": "number", "number": 90}},
{CONF_BELOW: {"chosen_selector": "number", "number": 90}},
does_not_raise(),
),
# Test invalid configurations
@@ -1235,7 +1235,7 @@ async def test_subscribe_triggers_no_triggers(
),
(
# Invalid choose selector option
{CONF_BELOW: {"active_choice": "cat", "cat": 90}},
{CONF_BELOW: {"chosen_selector": "cat", "cat": 90}},
pytest.raises(vol.Invalid),
),
],