mirror of
https://github.com/home-assistant/core.git
synced 2026-09-24 15:31:52 -05:00
Add Bitvis Power Hub integration (#165457)
This commit is contained in:
MandusBorjesson
@@ -118,6 +118,7 @@ homeassistant.components.bayesian.*
|
||||
homeassistant.components.besen.*
|
||||
homeassistant.components.binary_sensor.*
|
||||
homeassistant.components.bitcoin.*
|
||||
homeassistant.components.bitvis.*
|
||||
homeassistant.components.blockchain.*
|
||||
homeassistant.components.blue_current.*
|
||||
homeassistant.components.blueprint.*
|
||||
|
||||
Generated
+2
@@ -233,6 +233,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/besen/ @moryoav
|
||||
/homeassistant/components/binary_sensor/ @home-assistant/core
|
||||
/tests/components/binary_sensor/ @home-assistant/core
|
||||
/homeassistant/components/bitvis/ @MandusBorjesson @real-tintin @simontegelid
|
||||
/tests/components/bitvis/ @MandusBorjesson @real-tintin @simontegelid
|
||||
/homeassistant/components/bizkaibus/ @UgaitzEtxebarria
|
||||
/homeassistant/components/blebox/ @bbx-a @swistakm @bkobus-bbx
|
||||
/tests/components/blebox/ @bbx-a @swistakm @bkobus-bbx
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
"""The Bitvis Power Hub integration."""
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from homeassistant.const import CONF_PORT, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import DATA_LISTENER_REGISTRY, DOMAIN
|
||||
from .coordinator import (
|
||||
BitvisConfigEntry,
|
||||
BitvisDataUpdateCoordinator,
|
||||
async_get_listener_registry,
|
||||
)
|
||||
|
||||
_PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Bitvis Power Hub integration."""
|
||||
async_get_listener_registry(hass)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: BitvisConfigEntry) -> bool:
|
||||
"""Set up Bitvis Power Hub from a config entry."""
|
||||
async_get_listener_registry(hass)
|
||||
if TYPE_CHECKING:
|
||||
assert entry.unique_id is not None
|
||||
coordinator = BitvisDataUpdateCoordinator(
|
||||
hass,
|
||||
entry,
|
||||
entry.data[CONF_PORT],
|
||||
entry.unique_id,
|
||||
)
|
||||
|
||||
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: BitvisConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
|
||||
if unload_ok:
|
||||
await entry.runtime_data.async_stop()
|
||||
if not hass.config_entries.async_loaded_entries(DOMAIN):
|
||||
hass.data.pop(DATA_LISTENER_REGISTRY, None)
|
||||
return unload_ok
|
||||
@@ -0,0 +1,235 @@
|
||||
"""Config flow for the Bitvis Power Hub integration."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Self, override
|
||||
|
||||
from bitvis_protobuf.listener import FilterIp
|
||||
from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample
|
||||
from bitvis_protobuf.utils import (
|
||||
InvalidMacAddressError,
|
||||
async_resolve_host,
|
||||
async_verify_udp_port_bindable,
|
||||
normalize_host,
|
||||
)
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.data_entry_flow import AbortFlow
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.device_registry import format_mac
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
|
||||
from .const import DEFAULT_NAME, DEFAULT_PORT, DISCOVERY_TIMEOUT, DOMAIN
|
||||
from .coordinator import async_get_listener_registry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _get_friendly_name(name: str | None) -> str:
|
||||
"""Return a user-friendly name derived from the zeroconf name."""
|
||||
if not name:
|
||||
return DEFAULT_NAME
|
||||
instance = name.split(".", 1)[0]
|
||||
return instance or DEFAULT_NAME
|
||||
|
||||
|
||||
async def _async_test_port(hass: HomeAssistant, port: int) -> None:
|
||||
"""Verify the UDP port can be bound."""
|
||||
|
||||
if async_get_listener_registry(hass).has_listener(port):
|
||||
return
|
||||
|
||||
await async_verify_udp_port_bindable(port)
|
||||
|
||||
|
||||
async def _async_discover_mac_address(hass: HomeAssistant, host: str, port: int) -> str:
|
||||
"""Wait for a UDP message from the device and return its MAC address."""
|
||||
resolved_ips = await async_resolve_host(host)
|
||||
listener_registry = async_get_listener_registry(hass)
|
||||
listener = await listener_registry.async_get_or_create(port)
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
future: asyncio.Future[str] = loop.create_future()
|
||||
|
||||
@callback
|
||||
def _on_payload(
|
||||
payload: PayloadSample | PayloadDiagnostic, _addr: tuple[str, int]
|
||||
) -> None:
|
||||
if not future.done():
|
||||
future.set_result(payload.mac_address)
|
||||
|
||||
@callback
|
||||
def _on_error(err: Exception, addr: tuple[str, int]) -> None:
|
||||
if (
|
||||
addr[0] in resolved_ips
|
||||
and not future.done()
|
||||
and isinstance(err, InvalidMacAddressError)
|
||||
):
|
||||
future.set_exception(err)
|
||||
|
||||
filters: list[FilterIp] = []
|
||||
listener.register_error_callback(_on_error)
|
||||
try:
|
||||
for ip in resolved_ips:
|
||||
filt = FilterIp(ip)
|
||||
try:
|
||||
listener.register(filt, _on_payload)
|
||||
except RuntimeError as err:
|
||||
raise AbortFlow("already_in_progress") from err
|
||||
filters.append(filt)
|
||||
|
||||
return await asyncio.wait_for(future, timeout=DISCOVERY_TIMEOUT)
|
||||
finally:
|
||||
listener.unregister_error_callback(_on_error)
|
||||
for filt in filters:
|
||||
listener.unregister(filt)
|
||||
await listener_registry.async_remove_if_unused(port)
|
||||
|
||||
|
||||
class BitvisConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Bitvis Power Hub."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the config flow."""
|
||||
self._discovery_info: ZeroconfServiceInfo | None = None
|
||||
self._validated_host: str | None = None
|
||||
self._host: str | None = None
|
||||
|
||||
@override
|
||||
def is_matching(self, other_flow: Self) -> bool:
|
||||
"""Return True if other_flow is matching this flow."""
|
||||
return self._host is not None and self._host == other_flow._host
|
||||
|
||||
async def _async_validate_host(self, host: str) -> str:
|
||||
"""Verify port availability and discover the device MAC address."""
|
||||
await _async_test_port(self.hass, DEFAULT_PORT)
|
||||
return await _async_discover_mac_address(self.hass, host, DEFAULT_PORT)
|
||||
|
||||
async def _async_create_entry_from_host(
|
||||
self, host: str, title: str
|
||||
) -> ConfigFlowResult:
|
||||
"""Validate connectivity, discover MAC address, and create the entry."""
|
||||
self._host = host
|
||||
if self.hass.config_entries.flow.async_has_matching_flow(self):
|
||||
return self.async_abort(reason="already_in_progress")
|
||||
|
||||
try:
|
||||
mac_address = await self._async_validate_host(host)
|
||||
except TimeoutError:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_USER_DATA_SCHEMA, {CONF_HOST: host}
|
||||
),
|
||||
errors={"base": "timeout_connect"},
|
||||
)
|
||||
except InvalidMacAddressError:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_USER_DATA_SCHEMA, {CONF_HOST: host}
|
||||
),
|
||||
errors={"base": "invalid_mac"},
|
||||
)
|
||||
except OSError:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_USER_DATA_SCHEMA, {CONF_HOST: host}
|
||||
),
|
||||
errors={"base": "cannot_connect"},
|
||||
)
|
||||
|
||||
await self.async_set_unique_id(format_mac(mac_address))
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=title,
|
||||
data={
|
||||
CONF_HOST: host,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
},
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step."""
|
||||
if user_input is not None:
|
||||
host = normalize_host(user_input[CONF_HOST])
|
||||
return await self._async_create_entry_from_host(host, DEFAULT_NAME)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=STEP_USER_DATA_SCHEMA,
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_step_zeroconf(
|
||||
self, discovery_info: ZeroconfServiceInfo
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle zeroconf discovery."""
|
||||
_LOGGER.debug("Discovered Bitvis Power Hub via Zeroconf: %s", discovery_info)
|
||||
|
||||
host = discovery_info.host
|
||||
self._host = host
|
||||
|
||||
if self.hass.config_entries.flow.async_has_matching_flow(self):
|
||||
return self.async_abort(reason="already_in_progress")
|
||||
|
||||
try:
|
||||
mac_address = await self._async_validate_host(host)
|
||||
except TimeoutError:
|
||||
return self.async_abort(reason="timeout_connect")
|
||||
except InvalidMacAddressError:
|
||||
return self.async_abort(reason="invalid_mac")
|
||||
except OSError:
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
await self.async_set_unique_id(format_mac(mac_address))
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
|
||||
|
||||
self._discovery_info = discovery_info
|
||||
self._validated_host = host
|
||||
|
||||
self.context["title_placeholders"] = {
|
||||
"name": _get_friendly_name(discovery_info.name),
|
||||
"host": host,
|
||||
}
|
||||
|
||||
return await self.async_step_zeroconf_confirm()
|
||||
|
||||
async def async_step_zeroconf_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Confirm discovery."""
|
||||
assert self._discovery_info is not None
|
||||
|
||||
if user_input is not None:
|
||||
assert self._validated_host is not None
|
||||
|
||||
return self.async_create_entry(
|
||||
title=_get_friendly_name(self._discovery_info.name),
|
||||
data={
|
||||
CONF_HOST: self._validated_host,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
},
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="zeroconf_confirm",
|
||||
description_placeholders={
|
||||
"name": _get_friendly_name(self._discovery_info.name),
|
||||
"host": self._discovery_info.host,
|
||||
},
|
||||
)
|
||||
@@ -0,0 +1,18 @@
|
||||
"""Constants for the Bitvis Power Hub integration."""
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .coordinator import BitvisListenerRegistry
|
||||
|
||||
DOMAIN = "bitvis"
|
||||
MANUFACTURER = "Bitvis"
|
||||
MODEL_NAME = "Power Hub"
|
||||
|
||||
DEFAULT_NAME = "Bitvis Power Hub"
|
||||
DEFAULT_PORT = 58220
|
||||
DISCOVERY_TIMEOUT = 30
|
||||
|
||||
DATA_LISTENER_REGISTRY: HassKey[BitvisListenerRegistry] = HassKey(DOMAIN)
|
||||
@@ -0,0 +1,197 @@
|
||||
"""Data coordinator for Bitvis Power Hub."""
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
from typing import override
|
||||
|
||||
from bitvis_protobuf.listener import FilterMac, SharedListener
|
||||
from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryError
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from homeassistant.util import dt as dt_util
|
||||
from homeassistant.util.variance import ignore_variance
|
||||
|
||||
from .const import DATA_LISTENER_REGISTRY, DOMAIN, MODEL_NAME
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
type BitvisConfigEntry = ConfigEntry[BitvisDataUpdateCoordinator]
|
||||
|
||||
|
||||
def _uptime_to_boot_time(uptime_s: int) -> datetime:
|
||||
"""Convert uptime in seconds to an absolute boot datetime."""
|
||||
return dt_util.utcnow().replace(microsecond=0) - timedelta(seconds=uptime_s)
|
||||
|
||||
|
||||
@dataclass(kw_only=True)
|
||||
class BitvisData:
|
||||
"""Data structure for Bitvis measurements."""
|
||||
|
||||
sample: PayloadSample | None = None
|
||||
diagnostic: PayloadDiagnostic | None = None
|
||||
boot_time: datetime | None = None
|
||||
|
||||
|
||||
class BitvisListenerRegistry:
|
||||
"""Registry that manages one shared UDP listener per port.
|
||||
|
||||
Stored at hass.data[DATA_LISTENER_REGISTRY] so all coordinators can
|
||||
look it up without duplicating state-management logic.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize registry storage."""
|
||||
self._listeners: dict[int, SharedListener] = {}
|
||||
self._locks: dict[int, asyncio.Lock] = {}
|
||||
|
||||
async def async_get_or_create(self, port: int) -> SharedListener:
|
||||
"""Return the listener for *port*, creating and starting it if needed."""
|
||||
port_lock = self._locks.setdefault(port, asyncio.Lock())
|
||||
async with port_lock:
|
||||
if port not in self._listeners:
|
||||
listener = SharedListener()
|
||||
await listener.start(port)
|
||||
self._listeners[port] = listener
|
||||
return self._listeners[port]
|
||||
|
||||
async def async_remove_if_unused(self, port: int) -> None:
|
||||
"""Stop and remove the listener for *port* when no coordinators remain."""
|
||||
port_lock = self._locks.setdefault(port, asyncio.Lock())
|
||||
async with port_lock:
|
||||
listener = self._listeners.get(port)
|
||||
if listener is None or not listener.is_empty:
|
||||
return
|
||||
await listener.stop()
|
||||
del self._listeners[port]
|
||||
|
||||
def get(self, port: int) -> SharedListener | None:
|
||||
"""Return an existing listener for *port*, or None."""
|
||||
return self._listeners.get(port)
|
||||
|
||||
def has_listener(self, port: int) -> bool:
|
||||
"""Return True if a listener is already active on *port*."""
|
||||
return port in self._listeners
|
||||
|
||||
|
||||
def async_get_listener_registry(hass: HomeAssistant) -> BitvisListenerRegistry:
|
||||
"""Return (creating if needed) the Bitvis listener registry for this HA instance."""
|
||||
if DATA_LISTENER_REGISTRY not in hass.data:
|
||||
hass.data[DATA_LISTENER_REGISTRY] = BitvisListenerRegistry()
|
||||
return hass.data[DATA_LISTENER_REGISTRY]
|
||||
|
||||
|
||||
class BitvisDataUpdateCoordinator(DataUpdateCoordinator[BitvisData]):
|
||||
"""Coordinator to manage data updates from UDP packets."""
|
||||
|
||||
config_entry: BitvisConfigEntry
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
config_entry: BitvisConfigEntry,
|
||||
port: int,
|
||||
mac_address: str,
|
||||
) -> None:
|
||||
"""Initialize the coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
config_entry=config_entry,
|
||||
)
|
||||
self.port = port
|
||||
self.mac_address = mac_address
|
||||
self._filter = FilterMac(mac_address)
|
||||
self._registered = False
|
||||
self._stable_boot_time = ignore_variance(
|
||||
_uptime_to_boot_time, timedelta(minutes=5)
|
||||
)
|
||||
self.data = BitvisData()
|
||||
|
||||
@override
|
||||
async def _async_setup(self) -> None:
|
||||
"""Set up the coordinator by registering with the shared UDP listener."""
|
||||
try:
|
||||
listener = await self.hass.data[DATA_LISTENER_REGISTRY].async_get_or_create(
|
||||
self.port
|
||||
)
|
||||
listener.register(self._filter, self._handle_payload)
|
||||
self._registered = True
|
||||
except OSError as err:
|
||||
raise UpdateFailed(
|
||||
f"Failed to start UDP listener on port {self.port}"
|
||||
) from err
|
||||
except RuntimeError as err:
|
||||
raise ConfigEntryError(
|
||||
f"Failed to register MAC filter for {self.mac_address} "
|
||||
f"on port {self.port}"
|
||||
) from err
|
||||
|
||||
async def async_stop(self) -> None:
|
||||
"""Unregister from the shared listener, stopping it when no longer needed."""
|
||||
if not self._registered:
|
||||
return
|
||||
|
||||
listener_registry = self.hass.data[DATA_LISTENER_REGISTRY]
|
||||
if listener := listener_registry.get(self.port):
|
||||
listener.unregister(self._filter)
|
||||
await listener_registry.async_remove_if_unused(self.port)
|
||||
|
||||
self._registered = False
|
||||
_LOGGER.debug(
|
||||
"Unregistered coordinator from shared UDP listener for port %s", self.port
|
||||
)
|
||||
|
||||
@callback
|
||||
def _handle_payload(
|
||||
self,
|
||||
payload: PayloadSample | PayloadDiagnostic,
|
||||
addr: tuple[str, int],
|
||||
) -> None:
|
||||
"""Handle a parsed payload dispatched by the shared listener."""
|
||||
_LOGGER.debug("Received payload from %s", addr)
|
||||
if isinstance(payload, PayloadSample):
|
||||
self._handle_sample(payload)
|
||||
else:
|
||||
self._handle_diagnostic(payload)
|
||||
|
||||
@callback
|
||||
def _handle_sample(self, payload: PayloadSample) -> None:
|
||||
"""Update sample data and notify listeners."""
|
||||
self.data.sample = payload
|
||||
self.async_set_updated_data(self.data)
|
||||
|
||||
@callback
|
||||
def _handle_diagnostic(self, payload: PayloadDiagnostic) -> None:
|
||||
"""Update diagnostic data and notify listeners."""
|
||||
self.data.diagnostic = payload
|
||||
diagnostic = payload.diagnostic
|
||||
self.data.boot_time = self._stable_boot_time(diagnostic.uptime_s)
|
||||
|
||||
if diagnostic.HasField("device_info"):
|
||||
device_reg = dr.async_get(self.hass)
|
||||
if device := device_reg.async_get_device_by_identifier(
|
||||
(DOMAIN, self.mac_address), self.config_entry.entry_id
|
||||
):
|
||||
device_info = diagnostic.device_info
|
||||
model = device_info.model_name or MODEL_NAME
|
||||
sw_version = device_info.sw_version or None
|
||||
if device.model != model or device.sw_version != sw_version:
|
||||
device_reg.async_update_device(
|
||||
device.id,
|
||||
model=model,
|
||||
sw_version=sw_version,
|
||||
)
|
||||
|
||||
self.async_set_updated_data(self.data)
|
||||
|
||||
@override
|
||||
async def _async_update_data(self) -> BitvisData:
|
||||
"""Return current data (updates are push-based via UDP datagrams)."""
|
||||
return self.data
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"domain": "bitvis",
|
||||
"name": "Bitvis Power Hub",
|
||||
"codeowners": ["@MandusBorjesson", "@real-tintin", "@simontegelid"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/bitvis",
|
||||
"integration_type": "device",
|
||||
"iot_class": "local_push",
|
||||
"quality_scale": "bronze",
|
||||
"requirements": ["bitvis-protobuf==2.0.4"],
|
||||
"zeroconf": ["_powerhub._udp.local."]
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: "The integration does not have any actions."
|
||||
appropriate-polling:
|
||||
status: exempt
|
||||
comment: "The integration does not poll."
|
||||
brands: done
|
||||
common-modules: done
|
||||
config-flow-test-coverage: done
|
||||
config-flow: done
|
||||
dependency-transparency: done
|
||||
docs-actions:
|
||||
status: exempt
|
||||
comment: "The integration does not have actions."
|
||||
docs-conditions:
|
||||
status: exempt
|
||||
comment: "The integration does not have conditions."
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
docs-triggers:
|
||||
status: exempt
|
||||
comment: "The integration does not have triggers."
|
||||
entity-event-setup: done
|
||||
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: "The integration does not have any service actions."
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters:
|
||||
status: exempt
|
||||
comment: "The integration has no options flow."
|
||||
docs-installation-parameters: done
|
||||
entity-unavailable: todo
|
||||
integration-owner: done
|
||||
log-when-unavailable: todo
|
||||
parallel-updates: done
|
||||
reauthentication-flow:
|
||||
status: exempt
|
||||
comment: "The integration does not use any credentials. It connects to a local device using only host and port via unauthenticated UDP push."
|
||||
test-coverage: done
|
||||
|
||||
# Gold
|
||||
devices: done
|
||||
diagnostics: todo
|
||||
discovery-update-info: done
|
||||
discovery: done
|
||||
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:
|
||||
status: exempt
|
||||
comment: "The integration connects to a single device."
|
||||
entity-category: done
|
||||
entity-device-class: done
|
||||
entity-disabled-by-default: done
|
||||
entity-translations: done
|
||||
exception-translations: todo
|
||||
icon-translations: todo
|
||||
reconfiguration-flow: todo
|
||||
repair-issues: todo
|
||||
stale-devices: todo
|
||||
|
||||
# Platinum
|
||||
async-dependency: done
|
||||
inject-websession:
|
||||
status: exempt
|
||||
comment: "The integration does not use HTTP."
|
||||
strict-typing: done
|
||||
@@ -0,0 +1,501 @@
|
||||
"""Sensor platform for Bitvis Power Hub."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING, cast, override
|
||||
|
||||
from bitvis_protobuf.han_port_pb2 import HanPortSample
|
||||
from bitvis_protobuf.powerhub_pb2 import Diagnostic
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
|
||||
EntityCategory,
|
||||
UnitOfElectricCurrent,
|
||||
UnitOfElectricPotential,
|
||||
UnitOfEnergy,
|
||||
UnitOfPower,
|
||||
UnitOfReactiveEnergy,
|
||||
UnitOfReactivePower,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import BitvisConfigEntry
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
from .coordinator import BitvisDataUpdateCoordinator
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
def _optional(field: str) -> Callable[[HanPortSample], float | None]:
|
||||
"""Return a getter that yields None when a protobuf field is unset."""
|
||||
|
||||
def _get(data: HanPortSample) -> float | None:
|
||||
if data.HasField(field):
|
||||
return cast(float, getattr(data, field))
|
||||
return None
|
||||
|
||||
return _get
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class BitvisSensorEntityDescription(SensorEntityDescription):
|
||||
"""Describes Bitvis sensor entity."""
|
||||
|
||||
value_fn: Callable[[HanPortSample], float | None]
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class BitvisDiagnosticSensorEntityDescription(SensorEntityDescription):
|
||||
"""Describes Bitvis diagnostic sensor entity."""
|
||||
|
||||
value_fn: Callable[[Diagnostic], float | int | str | datetime | None]
|
||||
|
||||
|
||||
SENSOR_DESCRIPTIONS: tuple[BitvisSensorEntityDescription, ...] = (
|
||||
# Phase voltages
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_voltage_l1",
|
||||
translation_key="phase_voltage",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=1,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("phase_voltage_l1_v"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_voltage_l2",
|
||||
translation_key="phase_voltage",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=1,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("phase_voltage_l2_v"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_voltage_l3",
|
||||
translation_key="phase_voltage",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=1,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("phase_voltage_l3_v"),
|
||||
),
|
||||
# Phase currents
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_current_l1",
|
||||
translation_key="phase_current",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=2,
|
||||
value_fn=_optional("phase_current_l1_a"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_current_l2",
|
||||
translation_key="phase_current",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=2,
|
||||
value_fn=_optional("phase_current_l2_a"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="phase_current_l3",
|
||||
translation_key="phase_current",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.CURRENT,
|
||||
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=2,
|
||||
value_fn=_optional("phase_current_l3_a"),
|
||||
),
|
||||
# Total active power
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_delivered_to_client",
|
||||
translation_key="power_active_import",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
value_fn=_optional("power_active_delivered_to_client_kw"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_delivered_by_client",
|
||||
translation_key="power_active_export",
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
value_fn=_optional("power_active_delivered_by_client_kw"),
|
||||
),
|
||||
# Total reactive power
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_delivered_to_client",
|
||||
translation_key="power_reactive_import",
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_delivered_to_client_kvar"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_delivered_by_client",
|
||||
translation_key="power_reactive_export",
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_delivered_by_client_kvar"),
|
||||
),
|
||||
# Per-phase active power (to client)
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l1_delivered_to_client",
|
||||
translation_key="power_active_phase_import",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l1_delivered_to_client_kw"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l2_delivered_to_client",
|
||||
translation_key="power_active_phase_import",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l2_delivered_to_client_kw"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l3_delivered_to_client",
|
||||
translation_key="power_active_phase_import",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l3_delivered_to_client_kw"),
|
||||
),
|
||||
# Per-phase active power (by client)
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l1_delivered_by_client",
|
||||
translation_key="power_active_phase_export",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l1_delivered_by_client_kw"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l2_delivered_by_client",
|
||||
translation_key="power_active_phase_export",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l2_delivered_by_client_kw"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_active_l3_delivered_by_client",
|
||||
translation_key="power_active_phase_export",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.POWER,
|
||||
native_unit_of_measurement=UnitOfPower.KILO_WATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_active_l3_delivered_by_client_kw"),
|
||||
),
|
||||
# Per-phase reactive power (to client)
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l1_delivered_to_client",
|
||||
translation_key="power_reactive_phase_import",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l1_delivered_to_client_kvar"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l2_delivered_to_client",
|
||||
translation_key="power_reactive_phase_import",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l2_delivered_to_client_kvar"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l3_delivered_to_client",
|
||||
translation_key="power_reactive_phase_import",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l3_delivered_to_client_kvar"),
|
||||
),
|
||||
# Per-phase reactive power (by client)
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l1_delivered_by_client",
|
||||
translation_key="power_reactive_phase_export",
|
||||
translation_placeholders={"phase": "L1"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l1_delivered_by_client_kvar"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l2_delivered_by_client",
|
||||
translation_key="power_reactive_phase_export",
|
||||
translation_placeholders={"phase": "L2"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l2_delivered_by_client_kvar"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="power_reactive_l3_delivered_by_client",
|
||||
translation_key="power_reactive_phase_export",
|
||||
translation_placeholders={"phase": "L3"},
|
||||
device_class=SensorDeviceClass.REACTIVE_POWER,
|
||||
native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=3,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("power_reactive_l3_delivered_by_client_kvar"),
|
||||
),
|
||||
# Energy - active
|
||||
BitvisSensorEntityDescription(
|
||||
key="energy_active_delivered_to_client",
|
||||
translation_key="energy_active_import",
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=2,
|
||||
value_fn=_optional("energy_active_delivered_to_client_kwh"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="energy_active_delivered_by_client",
|
||||
translation_key="energy_active_export",
|
||||
device_class=SensorDeviceClass.ENERGY,
|
||||
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=2,
|
||||
value_fn=_optional("energy_active_delivered_by_client_kwh"),
|
||||
),
|
||||
# Energy - reactive
|
||||
BitvisSensorEntityDescription(
|
||||
key="energy_reactive_delivered_to_client",
|
||||
translation_key="energy_reactive_import",
|
||||
device_class=SensorDeviceClass.REACTIVE_ENERGY,
|
||||
native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=2,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("energy_reactive_delivered_to_client_kvarh"),
|
||||
),
|
||||
BitvisSensorEntityDescription(
|
||||
key="energy_reactive_delivered_by_client",
|
||||
translation_key="energy_reactive_export",
|
||||
device_class=SensorDeviceClass.REACTIVE_ENERGY,
|
||||
native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=2,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=_optional("energy_reactive_delivered_by_client_kvarh"),
|
||||
),
|
||||
)
|
||||
|
||||
UPTIME_DESCRIPTION = SensorEntityDescription(
|
||||
key="uptime",
|
||||
device_class=SensorDeviceClass.UPTIME,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
)
|
||||
|
||||
DIAGNOSTIC_SENSOR_DESCRIPTIONS: tuple[BitvisDiagnosticSensorEntityDescription, ...] = (
|
||||
BitvisDiagnosticSensorEntityDescription(
|
||||
key="wifi_rssi",
|
||||
translation_key="wifi_rssi",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=lambda data: data.wifi_rssi_dbm,
|
||||
),
|
||||
BitvisDiagnosticSensorEntityDescription(
|
||||
key="han_msg_successfully_parsed",
|
||||
translation_key="han_msg_successfully_parsed",
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=lambda data: data.han_msg_successfully_parsed,
|
||||
),
|
||||
BitvisDiagnosticSensorEntityDescription(
|
||||
key="han_msg_buffer_overflow",
|
||||
translation_key="han_msg_buffer_overflow",
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
value_fn=lambda data: data.han_msg_buffer_overflow,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: BitvisConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Bitvis sensor platform."""
|
||||
coordinator = entry.runtime_data
|
||||
known_keys: set[str] = set()
|
||||
|
||||
async_add_entities(
|
||||
[
|
||||
BitvisUptimeSensorEntity(coordinator, UPTIME_DESCRIPTION),
|
||||
*(
|
||||
BitvisDiagnosticSensorEntity(coordinator, description)
|
||||
for description in DIAGNOSTIC_SENSOR_DESCRIPTIONS
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
@callback
|
||||
def _check_entities() -> None:
|
||||
if (payload := coordinator.data.sample) is None:
|
||||
return
|
||||
entities = [
|
||||
BitvisSensorEntity(coordinator, description)
|
||||
for description in SENSOR_DESCRIPTIONS
|
||||
if description.key not in known_keys
|
||||
and description.value_fn(payload.sample) is not None
|
||||
]
|
||||
if entities:
|
||||
known_keys.update(entity.entity_description.key for entity in entities)
|
||||
async_add_entities(entities)
|
||||
|
||||
_check_entities()
|
||||
entry.async_on_unload(coordinator.async_add_listener(_check_entities))
|
||||
|
||||
|
||||
class BitvisBaseSensorEntity(
|
||||
CoordinatorEntity[BitvisDataUpdateCoordinator], SensorEntity
|
||||
):
|
||||
"""Base class for Bitvis sensor entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: BitvisDataUpdateCoordinator,
|
||||
description: SensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
mac_address = coordinator.mac_address
|
||||
self._attr_unique_id = f"{mac_address}_{description.key}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, mac_address)},
|
||||
connections={(CONNECTION_NETWORK_MAC, mac_address)},
|
||||
manufacturer=MANUFACTURER,
|
||||
)
|
||||
|
||||
|
||||
class BitvisSensorEntity(BitvisBaseSensorEntity):
|
||||
"""Representation of a Bitvis sensor."""
|
||||
|
||||
entity_description: BitvisSensorEntityDescription
|
||||
|
||||
@property
|
||||
@override
|
||||
def native_value(self) -> float | None:
|
||||
"""Return the state of the sensor."""
|
||||
payload = self.coordinator.data.sample
|
||||
if TYPE_CHECKING:
|
||||
assert payload is not None
|
||||
return self.entity_description.value_fn(payload.sample)
|
||||
|
||||
@property
|
||||
@override
|
||||
def available(self) -> bool:
|
||||
"""Return if entity is available."""
|
||||
return super().available and self.coordinator.data.sample is not None
|
||||
|
||||
|
||||
class BitvisDiagnosticSensorEntity(BitvisBaseSensorEntity):
|
||||
"""Representation of a Bitvis diagnostic sensor."""
|
||||
|
||||
entity_description: BitvisDiagnosticSensorEntityDescription
|
||||
|
||||
@property
|
||||
@override
|
||||
def native_value(self) -> float | int | str | datetime | None:
|
||||
"""Return the state of the sensor."""
|
||||
payload = self.coordinator.data.diagnostic
|
||||
if TYPE_CHECKING:
|
||||
assert payload is not None
|
||||
return self.entity_description.value_fn(payload.diagnostic)
|
||||
|
||||
@property
|
||||
@override
|
||||
def available(self) -> bool:
|
||||
"""Return if entity is available."""
|
||||
return super().available and self.coordinator.data.diagnostic is not None
|
||||
|
||||
|
||||
class BitvisUptimeSensorEntity(BitvisBaseSensorEntity):
|
||||
"""Sensor entity for device uptime (boot time)."""
|
||||
|
||||
@property
|
||||
@override
|
||||
def native_value(self) -> datetime | None:
|
||||
"""Return the stable boot time computed by the coordinator."""
|
||||
return self.coordinator.data.boot_time
|
||||
|
||||
@property
|
||||
@override
|
||||
def available(self) -> bool:
|
||||
"""Return if entity is available."""
|
||||
return super().available and self.coordinator.data.boot_time is not None
|
||||
@@ -0,0 +1,87 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_mac": "The device did not provide a valid MAC address. Update the Power Hub to the latest firmware and try again.",
|
||||
"timeout_connect": "[%key:common::config_flow::error::timeout_connect%]"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_mac": "The device did not provide a valid MAC address. Update the Power Hub to the latest firmware and try again.",
|
||||
"timeout_connect": "[%key:common::config_flow::error::timeout_connect%]"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "Hostname or IP address of your Bitvis Power Hub device."
|
||||
},
|
||||
"description": "Enter the hostname or IP address of your Bitvis Power Hub. Setup waits for a UDP packet from the device.",
|
||||
"title": "Set up Bitvis Power Hub"
|
||||
},
|
||||
"zeroconf_confirm": {
|
||||
"description": "Do you want to add the discovered Bitvis Power Hub ({name}) at {host} to Home Assistant?",
|
||||
"title": "Discovered Bitvis Power Hub"
|
||||
}
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"energy_active_export": {
|
||||
"name": "Active energy export"
|
||||
},
|
||||
"energy_active_import": {
|
||||
"name": "Active energy import"
|
||||
},
|
||||
"energy_reactive_export": {
|
||||
"name": "Reactive energy export"
|
||||
},
|
||||
"energy_reactive_import": {
|
||||
"name": "Reactive energy import"
|
||||
},
|
||||
"han_msg_buffer_overflow": {
|
||||
"name": "HAN buffer overflows"
|
||||
},
|
||||
"han_msg_successfully_parsed": {
|
||||
"name": "HAN messages successfully parsed"
|
||||
},
|
||||
"phase_current": {
|
||||
"name": "Current {phase}"
|
||||
},
|
||||
"phase_voltage": {
|
||||
"name": "Voltage {phase}"
|
||||
},
|
||||
"power_active_export": {
|
||||
"name": "Active power export"
|
||||
},
|
||||
"power_active_import": {
|
||||
"name": "Active power import"
|
||||
},
|
||||
"power_active_phase_export": {
|
||||
"name": "Active power export {phase}"
|
||||
},
|
||||
"power_active_phase_import": {
|
||||
"name": "Active power import {phase}"
|
||||
},
|
||||
"power_reactive_export": {
|
||||
"name": "Reactive power export"
|
||||
},
|
||||
"power_reactive_import": {
|
||||
"name": "Reactive power import"
|
||||
},
|
||||
"power_reactive_phase_export": {
|
||||
"name": "Reactive power export {phase}"
|
||||
},
|
||||
"power_reactive_phase_import": {
|
||||
"name": "Reactive power import {phase}"
|
||||
},
|
||||
"wifi_rssi": {
|
||||
"name": "Wi-Fi signal strength"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+1
@@ -102,6 +102,7 @@ FLOWS = {
|
||||
"bang_olufsen",
|
||||
"bayesian",
|
||||
"besen",
|
||||
"bitvis",
|
||||
"blebox",
|
||||
"blink",
|
||||
"blue_current",
|
||||
|
||||
@@ -756,6 +756,12 @@
|
||||
"config_flow": false,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"bitvis": {
|
||||
"name": "Bitvis Power Hub",
|
||||
"integration_type": "device",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"bizkaibus": {
|
||||
"name": "Bizkaibus",
|
||||
"integration_type": "hub",
|
||||
|
||||
Generated
+5
@@ -888,6 +888,11 @@ ZEROCONF = {
|
||||
"domain": "plugwise",
|
||||
},
|
||||
],
|
||||
"_powerhub._udp.local.": [
|
||||
{
|
||||
"domain": "bitvis",
|
||||
},
|
||||
],
|
||||
"_powerview._tcp.local.": [
|
||||
{
|
||||
"domain": "hunterdouglas_powerview",
|
||||
|
||||
@@ -937,6 +937,16 @@ disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.bitvis.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.blockchain.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
||||
Generated
+3
@@ -669,6 +669,9 @@ beautifulsoup4==4.13.3
|
||||
# homeassistant.components.besen
|
||||
besen==0.4.0
|
||||
|
||||
# homeassistant.components.bitvis
|
||||
bitvis-protobuf==2.0.4
|
||||
|
||||
# homeassistant.components.bizkaibus
|
||||
bizkaibus==0.1.1
|
||||
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Tests for the Bitvis Power Hub integration."""
|
||||
|
||||
from collections.abc import Callable
|
||||
|
||||
from bitvis_protobuf.listener import FilterMac
|
||||
from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample
|
||||
import pytest
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Set up the integration."""
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
def find_listener_callback(
|
||||
listener: object,
|
||||
mac_address: str,
|
||||
) -> Callable[[PayloadSample | PayloadDiagnostic, tuple[str, int]], None]:
|
||||
"""Find the listener callback registered for a MAC address."""
|
||||
register = listener.register
|
||||
for call in register.call_args_list:
|
||||
filt = call[0][0]
|
||||
if (
|
||||
isinstance(filt, FilterMac)
|
||||
and filt.mac_address.lower() == mac_address.lower()
|
||||
):
|
||||
return call[0][1]
|
||||
pytest.fail(f"Callback for MAC {mac_address} not found")
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Common fixtures for Bitvis Power Hub tests."""
|
||||
|
||||
from collections.abc import Callable, Generator, Iterator
|
||||
from contextlib import ExitStack, contextmanager
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from bitvis_protobuf.listener import Filter, FilterIp
|
||||
from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample, parse_payload
|
||||
from bitvis_protobuf.powerhub_pb2 import Payload
|
||||
from bitvis_protobuf.utils import InvalidMacAddressError
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.bitvis.const import DEFAULT_NAME, DEFAULT_PORT, DOMAIN
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
TEST_DEVICE_MAC = "aa:bb:cc:dd:ee:ff"
|
||||
SECOND_DEVICE_MAC = "11:22:33:44:55:66"
|
||||
|
||||
type ListenerCallback = Callable[
|
||||
[PayloadSample | PayloadDiagnostic, tuple[str, int]], None
|
||||
]
|
||||
type ErrorCallback = Callable[[Exception, tuple[str, int]], None]
|
||||
|
||||
|
||||
class FakeListener:
|
||||
"""In-memory SharedListener stand-in for config-flow and coordinator tests."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize callback storage and async start/stop mocks."""
|
||||
self._callbacks: dict[Filter, ListenerCallback] = {}
|
||||
self._error_callbacks: list[ErrorCallback] = []
|
||||
self.start = AsyncMock()
|
||||
self.stop = AsyncMock()
|
||||
self.register = MagicMock(side_effect=self._register)
|
||||
self.unregister = MagicMock(side_effect=self._unregister)
|
||||
self.register_error_callback = MagicMock(
|
||||
side_effect=self._error_callbacks.append
|
||||
)
|
||||
self.unregister_error_callback = MagicMock(side_effect=self._unregister_error)
|
||||
self.dispatch = MagicMock(side_effect=self._dispatch)
|
||||
|
||||
@property
|
||||
def is_empty(self) -> bool:
|
||||
"""Return True when no payload callbacks are registered."""
|
||||
return not self._callbacks
|
||||
|
||||
def _register(self, filt: Filter, callback: ListenerCallback) -> None:
|
||||
if filt in self._callbacks:
|
||||
raise RuntimeError(f"Filter already registered: {filt}")
|
||||
self._callbacks[filt] = callback
|
||||
|
||||
def _unregister(self, filt: Filter) -> None:
|
||||
self._callbacks.pop(filt, None)
|
||||
|
||||
def _unregister_error(self, callback: ErrorCallback) -> None:
|
||||
if callback in self._error_callbacks:
|
||||
self._error_callbacks.remove(callback)
|
||||
|
||||
def _dispatch(self, data: bytes, addr: tuple[str, int]) -> None:
|
||||
try:
|
||||
payload = parse_payload(data)
|
||||
except InvalidMacAddressError as err:
|
||||
for callback in self._error_callbacks:
|
||||
callback(err, addr)
|
||||
return
|
||||
if payload is None:
|
||||
return
|
||||
self.deliver(payload, addr)
|
||||
|
||||
def deliver(
|
||||
self,
|
||||
payload: PayloadSample | PayloadDiagnostic,
|
||||
addr: tuple[str, int],
|
||||
) -> None:
|
||||
"""Deliver an already-parsed payload to matching callbacks."""
|
||||
host = addr[0]
|
||||
for filt, callback in self._callbacks.items():
|
||||
if filt.match(payload, host):
|
||||
callback(payload, addr)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def patch_config_flow_connectivity(
|
||||
resolved_host: str,
|
||||
*,
|
||||
mac_address: str = TEST_DEVICE_MAC,
|
||||
deliver_mac: bool = True,
|
||||
invalid_mac: bool = False,
|
||||
port_bind_side_effect: BaseException | None = None,
|
||||
discovery_timeout: bool = False,
|
||||
register_side_effect: BaseException | None = None,
|
||||
shared_listener: FakeListener | None = None,
|
||||
) -> Iterator[AsyncMock]:
|
||||
"""Patch library connectivity helpers used by the config flow."""
|
||||
listener = shared_listener or FakeListener()
|
||||
|
||||
if register_side_effect is not None:
|
||||
listener.register.side_effect = register_side_effect
|
||||
else:
|
||||
original_register = listener.register.side_effect
|
||||
|
||||
def _on_register(filt: Filter, callback: ListenerCallback) -> None:
|
||||
original_register(filt, callback)
|
||||
if not isinstance(filt, FilterIp):
|
||||
return
|
||||
if invalid_mac:
|
||||
payload = Payload()
|
||||
payload.sample.SetInParent()
|
||||
listener.dispatch(payload.SerializeToString(), (resolved_host, 1234))
|
||||
elif deliver_mac and not discovery_timeout:
|
||||
callback(
|
||||
PayloadSample(mac_address=mac_address, sample=MagicMock()),
|
||||
(resolved_host, 1234),
|
||||
)
|
||||
|
||||
listener.register.side_effect = _on_register
|
||||
|
||||
with ExitStack() as stack:
|
||||
mock_verify = stack.enter_context(
|
||||
patch(
|
||||
"homeassistant.components.bitvis.config_flow.async_verify_udp_port_bindable",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=port_bind_side_effect,
|
||||
)
|
||||
)
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"homeassistant.components.bitvis.config_flow.async_resolve_host",
|
||||
new_callable=AsyncMock,
|
||||
return_value={resolved_host},
|
||||
)
|
||||
)
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"homeassistant.components.bitvis.coordinator.SharedListener",
|
||||
return_value=listener,
|
||||
)
|
||||
)
|
||||
if discovery_timeout:
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"homeassistant.components.bitvis.config_flow.DISCOVERY_TIMEOUT",
|
||||
0,
|
||||
)
|
||||
)
|
||||
yield mock_verify
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Return the default mocked config entry."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_HOST: "192.168.1.100", CONF_PORT: DEFAULT_PORT},
|
||||
unique_id=TEST_DEVICE_MAC,
|
||||
title=DEFAULT_NAME,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_zeroconf_config_entry() -> MockConfigEntry:
|
||||
"""Return a mocked config entry for zeroconf discovery host."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_HOST: "192.168.1.200", CONF_PORT: DEFAULT_PORT},
|
||||
unique_id=TEST_DEVICE_MAC,
|
||||
title=DEFAULT_NAME,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ipv6_config_entry() -> MockConfigEntry:
|
||||
"""Return a mocked config entry with an IPv6 host."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_HOST: "2001:db8::10", CONF_PORT: DEFAULT_PORT},
|
||||
unique_id=SECOND_DEVICE_MAC,
|
||||
title=DEFAULT_NAME,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_second_config_entry() -> MockConfigEntry:
|
||||
"""Return a second mocked config entry on the same UDP port."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_HOST: "192.168.1.101", CONF_PORT: DEFAULT_PORT},
|
||||
unique_id=SECOND_DEVICE_MAC,
|
||||
title=DEFAULT_NAME,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_shared_listener() -> FakeListener:
|
||||
"""Return a fake bitvis_protobuf SharedListener."""
|
||||
return FakeListener()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patch_shared_listener(
|
||||
mock_shared_listener: FakeListener,
|
||||
) -> Generator[FakeListener]:
|
||||
"""Patch SharedListener to return a mocked instance."""
|
||||
with patch(
|
||||
"homeassistant.components.bitvis.coordinator.SharedListener",
|
||||
return_value=mock_shared_listener,
|
||||
):
|
||||
yield mock_shared_listener
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.bitvis.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def init_integration(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> MockConfigEntry:
|
||||
"""Set up the integration with a mocked UDP listener."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
return mock_config_entry
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,613 @@
|
||||
"""Tests for the Bitvis Power Hub config flow."""
|
||||
|
||||
import asyncio
|
||||
from ipaddress import ip_address
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from bitvis_protobuf.parse import PayloadSample
|
||||
from bitvis_protobuf.powerhub_pb2 import Payload
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.bitvis.const import DEFAULT_NAME, DEFAULT_PORT, DOMAIN
|
||||
from homeassistant.components.bitvis.coordinator import async_get_listener_registry
|
||||
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
|
||||
from .conftest import (
|
||||
SECOND_DEVICE_MAC,
|
||||
TEST_DEVICE_MAC,
|
||||
FakeListener,
|
||||
patch_config_flow_connectivity,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
|
||||
|
||||
ZEROCONF_HOST = "192.168.1.200"
|
||||
USER_HOST = "192.168.1.100"
|
||||
|
||||
|
||||
def _zeroconf_discovery(
|
||||
host: str = ZEROCONF_HOST,
|
||||
name: str = "Bitvis Power Hub._powerhub._udp.local.",
|
||||
port: int | None = DEFAULT_PORT,
|
||||
) -> ZeroconfServiceInfo:
|
||||
return ZeroconfServiceInfo(
|
||||
ip_address=ip_address(host),
|
||||
ip_addresses=[ip_address(host)],
|
||||
hostname="powerhub.local.",
|
||||
name=name,
|
||||
port=port,
|
||||
properties={},
|
||||
type="_powerhub._udp.local.",
|
||||
)
|
||||
|
||||
|
||||
ZEROCONF_DISCOVERY = _zeroconf_discovery()
|
||||
UNRELATED_HOST = "10.9.9.9"
|
||||
|
||||
|
||||
def _invalid_mac_datagram() -> bytes:
|
||||
payload = Payload()
|
||||
payload.sample.SetInParent()
|
||||
return payload.SerializeToString()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("input_host", "resolved_ip", "expected_host"),
|
||||
[
|
||||
pytest.param(USER_HOST, USER_HOST, USER_HOST, id="ipv4"),
|
||||
pytest.param("2001:db8::10", "2001:db8::10", "2001:db8::10", id="ipv6"),
|
||||
pytest.param(
|
||||
"my-powerhub.local", "10.0.0.5", "my-powerhub.local", id="hostname"
|
||||
),
|
||||
pytest.param(
|
||||
"[2001:db8::10]", "2001:db8::10", "2001:db8::10", id="bracketed-ipv6"
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_user_form_create_entry(
|
||||
hass: HomeAssistant,
|
||||
input_host: str,
|
||||
resolved_ip: str,
|
||||
expected_host: str,
|
||||
) -> None:
|
||||
"""Test creating an entry via 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"
|
||||
|
||||
with patch_config_flow_connectivity(resolved_ip):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: input_host,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["data"] == {
|
||||
CONF_HOST: expected_host,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
}
|
||||
assert result["result"].unique_id == TEST_DEVICE_MAC
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("connectivity_kwargs", "error_key"),
|
||||
[
|
||||
pytest.param(
|
||||
{"port_bind_side_effect": OSError("UDP port is unavailable")},
|
||||
"cannot_connect",
|
||||
id="cannot-connect",
|
||||
),
|
||||
pytest.param({"invalid_mac": True}, "invalid_mac", id="invalid-mac"),
|
||||
pytest.param(
|
||||
{"deliver_mac": False, "discovery_timeout": True},
|
||||
"timeout_connect",
|
||||
id="timeout",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_user_form_error_and_recovery(
|
||||
hass: HomeAssistant,
|
||||
connectivity_kwargs: dict[str, object],
|
||||
error_key: str,
|
||||
) -> None:
|
||||
"""Test user form error then successful recovery."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch_config_flow_connectivity(USER_HOST, **connectivity_kwargs):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": error_key}
|
||||
|
||||
with patch_config_flow_connectivity(USER_HOST):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == DEFAULT_NAME
|
||||
assert result["data"] == {
|
||||
CONF_HOST: USER_HOST,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
}
|
||||
assert result["result"].unique_id == TEST_DEVICE_MAC
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"input_host",
|
||||
[
|
||||
pytest.param(USER_HOST, id="same-host"),
|
||||
pytest.param("192.168.1.101", id="different-host"),
|
||||
],
|
||||
)
|
||||
async def test_user_form_duplicate_mac(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, input_host: str
|
||||
) -> None:
|
||||
"""Test duplicate detection is based on MAC address, not host."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch_config_flow_connectivity(input_host):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: input_host,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_user_form_reused_ip_new_device(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test a new device can be added at an IP already stored on another entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch_config_flow_connectivity(USER_HOST, mac_address=SECOND_DEVICE_MAC):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
CONF_HOST: USER_HOST,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
}
|
||||
assert result["result"].unique_id == SECOND_DEVICE_MAC
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("connectivity_kwargs", "reason"),
|
||||
[
|
||||
pytest.param(
|
||||
{"port_bind_side_effect": OSError("UDP port is unavailable")},
|
||||
"cannot_connect",
|
||||
id="cannot-connect",
|
||||
),
|
||||
pytest.param({"invalid_mac": True}, "invalid_mac", id="invalid-mac"),
|
||||
pytest.param(
|
||||
{"deliver_mac": False, "discovery_timeout": True},
|
||||
"timeout_connect",
|
||||
id="timeout",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_zeroconf_abort(
|
||||
hass: HomeAssistant,
|
||||
connectivity_kwargs: dict[str, object],
|
||||
reason: str,
|
||||
) -> None:
|
||||
"""Test zeroconf discovery abort reasons."""
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST, **connectivity_kwargs):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == reason
|
||||
|
||||
|
||||
async def test_zeroconf_duplicate(
|
||||
hass: HomeAssistant, mock_zeroconf_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test that a duplicate zeroconf discovery is aborted by MAC address."""
|
||||
mock_zeroconf_config_entry.add_to_hass(hass)
|
||||
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_zeroconf_reused_ip_new_device(
|
||||
hass: HomeAssistant, mock_zeroconf_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test zeroconf can add a new device at an IP already stored on another entry."""
|
||||
mock_zeroconf_config_entry.add_to_hass(hass)
|
||||
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST, mac_address=SECOND_DEVICE_MAC):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "zeroconf_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
CONF_HOST: ZEROCONF_HOST,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
}
|
||||
assert result["result"].unique_id == SECOND_DEVICE_MAC
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("name", "port", "expected_title"),
|
||||
[
|
||||
pytest.param(
|
||||
"Bitvis Power Hub._powerhub._udp.local.",
|
||||
DEFAULT_PORT,
|
||||
"Bitvis Power Hub",
|
||||
id="happy-path",
|
||||
),
|
||||
pytest.param(
|
||||
"Bitvis Power Hub._powerhub._udp.local.",
|
||||
None,
|
||||
"Bitvis Power Hub",
|
||||
id="none-port",
|
||||
),
|
||||
pytest.param(
|
||||
"My Custom Hub._powerhub._udp.local.",
|
||||
DEFAULT_PORT,
|
||||
"My Custom Hub",
|
||||
id="friendly-name",
|
||||
),
|
||||
pytest.param("", DEFAULT_PORT, DEFAULT_NAME, id="empty-name"),
|
||||
pytest.param(
|
||||
"._powerhub._udp.local.", DEFAULT_PORT, DEFAULT_NAME, id="dot-prefixed"
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_zeroconf_create_entry(
|
||||
hass: HomeAssistant,
|
||||
name: str,
|
||||
port: int | None,
|
||||
expected_title: str,
|
||||
) -> None:
|
||||
"""Test zeroconf confirm creates an entry with title, data, and unique_id."""
|
||||
discovery = _zeroconf_discovery(name=name, port=port)
|
||||
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=discovery,
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "zeroconf_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == expected_title
|
||||
assert result["data"] == {
|
||||
CONF_HOST: ZEROCONF_HOST,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
}
|
||||
assert result["result"].unique_id == TEST_DEVICE_MAC
|
||||
|
||||
|
||||
async def test_zeroconf_updates_host_on_new_ip(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test rediscovery on a new IP updates the stored host and aborts."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert mock_config_entry.data[CONF_HOST] != ZEROCONF_HOST
|
||||
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
assert mock_config_entry.data[CONF_HOST] == ZEROCONF_HOST
|
||||
|
||||
|
||||
async def test_aborted_flow_removes_listener(
|
||||
hass: HomeAssistant,
|
||||
mock_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test listener is stopped after an aborted config flow."""
|
||||
with patch_config_flow_connectivity(
|
||||
USER_HOST,
|
||||
deliver_mac=False,
|
||||
discovery_timeout=True,
|
||||
shared_listener=mock_shared_listener,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "timeout_connect"}
|
||||
mock_shared_listener.stop.assert_awaited_once()
|
||||
assert not async_get_listener_registry(hass).has_listener(DEFAULT_PORT)
|
||||
|
||||
|
||||
async def test_invalid_mac_from_other_host_is_ignored(
|
||||
hass: HomeAssistant, mock_shared_listener: FakeListener
|
||||
) -> None:
|
||||
"""Test an invalid-MAC datagram from another host does not fail the flow."""
|
||||
with patch_config_flow_connectivity(
|
||||
USER_HOST, deliver_mac=False, shared_listener=mock_shared_listener
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
configure_task = asyncio.create_task(
|
||||
hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_shared_listener.dispatch(_invalid_mac_datagram(), (UNRELATED_HOST, 1234))
|
||||
mock_shared_listener.deliver(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=MagicMock()),
|
||||
(USER_HOST, 1234),
|
||||
)
|
||||
result = await configure_task
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == TEST_DEVICE_MAC
|
||||
|
||||
|
||||
async def test_invalid_mac_does_not_fail_other_flow(
|
||||
hass: HomeAssistant, mock_shared_listener: FakeListener
|
||||
) -> None:
|
||||
"""Test an invalid-MAC datagram only fails the flow waiting for that host."""
|
||||
|
||||
async def resolve_host(host: str) -> set[str]:
|
||||
return {host}
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.bitvis.config_flow.async_verify_udp_port_bindable",
|
||||
new_callable=AsyncMock,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.bitvis.config_flow.async_resolve_host",
|
||||
side_effect=resolve_host,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.bitvis.coordinator.SharedListener",
|
||||
return_value=mock_shared_listener,
|
||||
),
|
||||
):
|
||||
first_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
first_task = asyncio.create_task(
|
||||
hass.config_entries.flow.async_configure(
|
||||
first_result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
second_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
second_task = asyncio.create_task(
|
||||
hass.config_entries.flow.async_configure(
|
||||
second_result["flow_id"],
|
||||
{
|
||||
CONF_HOST: ZEROCONF_HOST,
|
||||
},
|
||||
)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_shared_listener.dispatch(_invalid_mac_datagram(), (ZEROCONF_HOST, 1234))
|
||||
mock_shared_listener.deliver(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=MagicMock()),
|
||||
(USER_HOST, 1234),
|
||||
)
|
||||
first_result = await first_task
|
||||
second_result = await second_task
|
||||
|
||||
assert first_result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert first_result["result"].unique_id == TEST_DEVICE_MAC
|
||||
assert second_result["type"] is FlowResultType.FORM
|
||||
assert second_result["errors"] == {"base": "invalid_mac"}
|
||||
|
||||
|
||||
async def test_concurrent_flow_same_host_aborts(hass: HomeAssistant) -> None:
|
||||
"""Test concurrent flows for the same host abort with already_in_progress."""
|
||||
with patch_config_flow_connectivity(USER_HOST, deliver_mac=False):
|
||||
first_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
first_task = asyncio.create_task(
|
||||
hass.config_entries.flow.async_configure(
|
||||
first_result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
second_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
second_result = await hass.config_entries.flow.async_configure(
|
||||
second_result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
first_task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await first_task
|
||||
|
||||
assert second_result["type"] is FlowResultType.ABORT
|
||||
assert second_result["reason"] == "already_in_progress"
|
||||
|
||||
|
||||
async def test_discovery_register_runtime_error_aborts(hass: HomeAssistant) -> None:
|
||||
"""Test discovery aborts when filter registration raises RuntimeError."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch_config_flow_connectivity(
|
||||
USER_HOST,
|
||||
deliver_mac=False,
|
||||
register_side_effect=RuntimeError("Filter already registered"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: USER_HOST,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_in_progress"
|
||||
|
||||
|
||||
async def test_zeroconf_concurrent_flow_same_host_aborts(hass: HomeAssistant) -> None:
|
||||
"""Test concurrent zeroconf flows for the same host abort."""
|
||||
with patch_config_flow_connectivity(ZEROCONF_HOST, deliver_mac=False):
|
||||
first_task = asyncio.create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
second_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_ZEROCONF},
|
||||
data=ZEROCONF_DISCOVERY,
|
||||
)
|
||||
|
||||
first_task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await first_task
|
||||
|
||||
assert second_result["type"] is FlowResultType.ABORT
|
||||
assert second_result["reason"] == "already_in_progress"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("listener_already_running", "expected_awaits"),
|
||||
[
|
||||
pytest.param(True, 0, id="listener-exists"),
|
||||
pytest.param(False, 1, id="no-listener"),
|
||||
],
|
||||
)
|
||||
async def test_user_form_port_bind_check(
|
||||
hass: HomeAssistant,
|
||||
mock_shared_listener: FakeListener,
|
||||
listener_already_running: bool,
|
||||
expected_awaits: int,
|
||||
) -> None:
|
||||
"""Test user flow skips the port bind check only when a listener exists."""
|
||||
if listener_already_running:
|
||||
with patch(
|
||||
"homeassistant.components.bitvis.coordinator.SharedListener",
|
||||
return_value=mock_shared_listener,
|
||||
):
|
||||
await async_get_listener_registry(hass).async_get_or_create(DEFAULT_PORT)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
kwargs: dict[str, object] = {
|
||||
"mac_address": SECOND_DEVICE_MAC
|
||||
if listener_already_running
|
||||
else TEST_DEVICE_MAC
|
||||
}
|
||||
if listener_already_running:
|
||||
kwargs["shared_listener"] = mock_shared_listener
|
||||
|
||||
with patch_config_flow_connectivity("192.168.1.101", **kwargs) as mock_verify:
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: "192.168.1.101",
|
||||
},
|
||||
)
|
||||
|
||||
assert mock_verify.await_count == expected_awaits
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Tests for the Bitvis Power Hub coordinator."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import FakeListener
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("patch_shared_listener")
|
||||
|
||||
|
||||
async def test_setup_oserror_results_in_setup_retry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that OSError from SharedListener.start results in SETUP_RETRY."""
|
||||
mock_shared_listener.start = AsyncMock(side_effect=OSError("port in use"))
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_runtime_error_results_in_setup_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that RuntimeError from SharedListener.register results in SETUP_ERROR."""
|
||||
mock_shared_listener.register.side_effect = RuntimeError("duplicate filter")
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
|
||||
mock_shared_listener.unregister.assert_not_called()
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Tests for the Bitvis Power Hub integration."""
|
||||
|
||||
from bitvis_protobuf import powerhub_pb2
|
||||
from bitvis_protobuf.parse import PayloadSample
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.bitvis.const import DATA_LISTENER_REGISTRY, DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import find_listener_callback, setup_integration
|
||||
from .conftest import TEST_DEVICE_MAC, FakeListener
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_setup_entry(
|
||||
init_integration: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test successful integration setup."""
|
||||
assert init_integration.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_unload_entry(
|
||||
hass: HomeAssistant, init_integration: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test that unloading stops the coordinator and unloads platforms."""
|
||||
assert DATA_LISTENER_REGISTRY in hass.data
|
||||
assert await hass.config_entries.async_unload(init_integration.entry_id)
|
||||
assert init_integration.state is ConfigEntryState.NOT_LOADED
|
||||
assert DATA_LISTENER_REGISTRY not in hass.data
|
||||
|
||||
|
||||
async def test_two_entries_share_listener(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_second_config_entry: MockConfigEntry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that two entries on the same port share one library listener."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_second_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_second_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
assert mock_second_config_entry.state is ConfigEntryState.LOADED
|
||||
patch_shared_listener.start.assert_awaited_once()
|
||||
|
||||
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
patch_shared_listener.stop.assert_not_called()
|
||||
assert DATA_LISTENER_REGISTRY in hass.data
|
||||
|
||||
assert await hass.config_entries.async_unload(mock_second_config_entry.entry_id)
|
||||
patch_shared_listener.stop.assert_awaited_once()
|
||||
assert DATA_LISTENER_REGISTRY not in hass.data
|
||||
|
||||
|
||||
async def test_unload_after_dynamic_entities(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test unload succeeds after HAN sensors have been created dynamically."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.sample.power_active_delivered_to_client_kw = 2.0
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_entities_unavailable_before_data(
|
||||
hass: HomeAssistant,
|
||||
init_integration: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test diagnostic entities are unavailable after setup before any packet."""
|
||||
wifi_entity_id = entity_registry.async_get_entity_id(
|
||||
"sensor",
|
||||
DOMAIN,
|
||||
f"{TEST_DEVICE_MAC}_wifi_rssi",
|
||||
)
|
||||
assert wifi_entity_id is not None
|
||||
state = hass.states.get(wifi_entity_id)
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
@@ -0,0 +1,293 @@
|
||||
"""Tests for the Bitvis Power Hub sensor platform."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from bitvis_protobuf import powerhub_pb2
|
||||
from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.bitvis.const import DOMAIN
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
|
||||
|
||||
from . import find_listener_callback, setup_integration
|
||||
from .conftest import TEST_DEVICE_MAC, FakeListener
|
||||
|
||||
from tests.common import MockConfigEntry, snapshot_platform
|
||||
|
||||
DIAGNOSTIC_UNIQUE_IDS = {
|
||||
f"{TEST_DEVICE_MAC}_uptime",
|
||||
f"{TEST_DEVICE_MAC}_wifi_rssi",
|
||||
f"{TEST_DEVICE_MAC}_han_msg_successfully_parsed",
|
||||
f"{TEST_DEVICE_MAC}_han_msg_buffer_overflow",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def enable_all_entities(entity_registry_enabled_by_default: None) -> None:
|
||||
"""Make sure all entities are enabled."""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_payload() -> PayloadSample:
|
||||
"""Return a sample payload with test data."""
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.sample.phase_voltage_l1_v = 230.0
|
||||
payload.sample.phase_voltage_l2_v = 229.5
|
||||
payload.sample.phase_voltage_l3_v = 231.2
|
||||
payload.sample.phase_current_l1_a = 10.5
|
||||
payload.sample.phase_current_l2_a = 8.3
|
||||
payload.sample.phase_current_l3_a = 12.1
|
||||
payload.sample.power_active_delivered_to_client_kw = 2.415
|
||||
payload.sample.power_active_delivered_by_client_kw = 0.0
|
||||
payload.sample.power_reactive_delivered_to_client_kvar = 0.5
|
||||
payload.sample.power_reactive_delivered_by_client_kvar = 0.0
|
||||
payload.sample.power_active_l1_delivered_to_client_kw = 0.8
|
||||
payload.sample.power_active_l2_delivered_to_client_kw = 0.7
|
||||
payload.sample.power_active_l3_delivered_to_client_kw = 0.915
|
||||
payload.sample.power_active_l1_delivered_by_client_kw = 0.0
|
||||
payload.sample.power_active_l2_delivered_by_client_kw = 0.0
|
||||
payload.sample.power_active_l3_delivered_by_client_kw = 0.0
|
||||
payload.sample.power_reactive_l1_delivered_to_client_kvar = 0.2
|
||||
payload.sample.power_reactive_l2_delivered_to_client_kvar = 0.15
|
||||
payload.sample.power_reactive_l3_delivered_to_client_kvar = 0.15
|
||||
payload.sample.power_reactive_l1_delivered_by_client_kvar = 0.0
|
||||
payload.sample.power_reactive_l2_delivered_by_client_kvar = 0.0
|
||||
payload.sample.power_reactive_l3_delivered_by_client_kvar = 0.0
|
||||
payload.sample.energy_active_delivered_to_client_kwh = 1234.56
|
||||
payload.sample.energy_active_delivered_by_client_kwh = 789.12
|
||||
payload.sample.energy_reactive_delivered_to_client_kvarh = 45.67
|
||||
payload.sample.energy_reactive_delivered_by_client_kvarh = 23.45
|
||||
return PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def diagnostic_payload() -> PayloadDiagnostic:
|
||||
"""Return a diagnostic payload with test data."""
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.diagnostic.uptime_s = 86400
|
||||
payload.diagnostic.wifi_rssi_dbm = -65
|
||||
payload.diagnostic.device_info.model_name = "PowerHub Gen2"
|
||||
payload.diagnostic.device_info.sw_version = "2.0.0"
|
||||
payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff"
|
||||
payload.diagnostic.han_msg_successfully_parsed = 1000
|
||||
payload.diagnostic.han_msg_buffer_overflow = 5
|
||||
return PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic)
|
||||
|
||||
|
||||
@pytest.mark.freeze_time("2026-01-01 12:00:00")
|
||||
async def test_all_entities(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
sample_payload: PayloadSample,
|
||||
diagnostic_payload: PayloadDiagnostic,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test all entities with snapshot."""
|
||||
with patch("homeassistant.components.bitvis._PLATFORMS", [Platform.SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
callback = find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)
|
||||
callback(sample_payload, ("192.168.1.100", 1234))
|
||||
callback(diagnostic_payload, ("192.168.1.100", 1234))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_entities_added_when_fields_become_available(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that HAN sensors are created when their fields first appear."""
|
||||
base_unique_id = mock_config_entry.unique_id
|
||||
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.sample.power_active_delivered_to_client_kw = 2.0
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
unique_ids = {
|
||||
entry.unique_id
|
||||
for entry in er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
}
|
||||
assert unique_ids == DIAGNOSTIC_UNIQUE_IDS | {
|
||||
f"{base_unique_id}_power_active_delivered_to_client"
|
||||
}
|
||||
|
||||
payload.sample.phase_voltage_l1_v = 230.0
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
unique_ids = {
|
||||
entry.unique_id
|
||||
for entry in er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
}
|
||||
assert unique_ids == DIAGNOSTIC_UNIQUE_IDS | {
|
||||
f"{base_unique_id}_power_active_delivered_to_client",
|
||||
f"{base_unique_id}_phase_voltage_l1",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_sensors_become_available_with_data(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that sensors become available when data arrives."""
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.sample.power_active_delivered_to_client_kw = 2.0
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_id = entity_registry.async_get_entity_id(
|
||||
"sensor",
|
||||
DOMAIN,
|
||||
f"{mock_config_entry.unique_id}_power_active_delivered_to_client",
|
||||
)
|
||||
assert entity_id is not None
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None
|
||||
assert state.state != "unavailable"
|
||||
assert float(state.state) == pytest.approx(2.0)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_diagnostic_sensors_update_with_data(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: er.EntityRegistry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that diagnostic sensors update when a diagnostic payload arrives."""
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.diagnostic.uptime_s = 999
|
||||
payload.diagnostic.wifi_rssi_dbm = -70
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
wifi_entity_id = entity_registry.async_get_entity_id(
|
||||
"sensor",
|
||||
DOMAIN,
|
||||
f"{TEST_DEVICE_MAC}_wifi_rssi",
|
||||
)
|
||||
assert wifi_entity_id is not None
|
||||
wifi_state = hass.states.get(wifi_entity_id)
|
||||
assert wifi_state is not None
|
||||
assert wifi_state.state != "unavailable"
|
||||
assert float(wifi_state.state) == pytest.approx(-70)
|
||||
|
||||
uptime_entity_id = entity_registry.async_get_entity_id(
|
||||
"sensor",
|
||||
DOMAIN,
|
||||
f"{TEST_DEVICE_MAC}_uptime",
|
||||
)
|
||||
assert uptime_entity_id is not None
|
||||
uptime_state = hass.states.get(uptime_entity_id)
|
||||
assert uptime_state is not None
|
||||
assert uptime_state.state != "unavailable"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_device_info_updated_from_diagnostic(
|
||||
hass: HomeAssistant,
|
||||
init_integration: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that device info is updated from a diagnostic payload."""
|
||||
device = device_registry.async_get_device_by_identifier(
|
||||
(DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id
|
||||
)
|
||||
assert device is not None
|
||||
assert device.model is None
|
||||
assert device.sw_version is None
|
||||
assert (CONNECTION_NETWORK_MAC, TEST_DEVICE_MAC) in device.connections
|
||||
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.diagnostic.uptime_s = 10
|
||||
payload.diagnostic.device_info.model_name = "PowerHub Gen2"
|
||||
payload.diagnostic.device_info.sw_version = "1.2.3"
|
||||
payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff"
|
||||
find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)(
|
||||
PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device = device_registry.async_get_device_by_identifier(
|
||||
(DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id
|
||||
)
|
||||
assert device is not None
|
||||
assert device.model == "PowerHub Gen2"
|
||||
assert device.sw_version == "1.2.3"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_device_info_kept_when_absent_in_later_payload(
|
||||
hass: HomeAssistant,
|
||||
init_integration: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
sample_payload: PayloadSample,
|
||||
patch_shared_listener: FakeListener,
|
||||
) -> None:
|
||||
"""Test that known model/sw_version are kept when later payloads omit them."""
|
||||
payload = powerhub_pb2.Payload()
|
||||
payload.diagnostic.uptime_s = 10
|
||||
payload.diagnostic.device_info.model_name = "PowerHub"
|
||||
payload.diagnostic.device_info.sw_version = "1.0"
|
||||
payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff"
|
||||
callback = find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)
|
||||
callback(
|
||||
PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device = device_registry.async_get_device_by_identifier(
|
||||
(DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id
|
||||
)
|
||||
assert device is not None
|
||||
assert device.model == "PowerHub"
|
||||
assert device.sw_version == "1.0"
|
||||
|
||||
payload2 = powerhub_pb2.Payload()
|
||||
payload2.diagnostic.uptime_s = 20
|
||||
callback(
|
||||
PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload2.diagnostic),
|
||||
("192.168.1.100", 1234),
|
||||
)
|
||||
callback(sample_payload, ("192.168.1.100", 1234))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device = device_registry.async_get_device_by_identifier(
|
||||
(DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id
|
||||
)
|
||||
assert device is not None
|
||||
assert device.model == "PowerHub"
|
||||
assert device.sw_version == "1.0"
|
||||
Reference in New Issue
Block a user