1
0
mirror of https://github.com/home-assistant/core.git synced 2025-12-20 02:48:57 +00:00

Add support for Victron bluetooth low energy devices (#148043)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Raj Laud
2025-11-18 14:12:48 -06:00
committed by GitHub
parent 97ef4a35b9
commit 4b69543515
20 changed files with 3395 additions and 5 deletions

2
CODEOWNERS generated
View File

@@ -1736,6 +1736,8 @@ build.json @home-assistant/supervisor
/tests/components/vesync/ @markperdue @webdjoe @thegardenmonkey @cdnninja @iprak @sapuseven
/homeassistant/components/vicare/ @CFenner
/tests/components/vicare/ @CFenner
/homeassistant/components/victron_ble/ @rajlaud
/tests/components/victron_ble/ @rajlaud
/homeassistant/components/victron_remote_monitoring/ @AndyTempel
/tests/components/victron_remote_monitoring/ @AndyTempel
/homeassistant/components/vilfo/ @ManneW

View File

@@ -0,0 +1,5 @@
{
"domain": "victron",
"name": "Victron",
"integrations": ["victron_ble", "victron_remote_monitoring"]
}

View File

@@ -0,0 +1,54 @@
"""The Victron Bluetooth Low Energy integration."""
from __future__ import annotations
import logging
from victron_ble_ha_parser import VictronBluetoothDeviceData
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
async_rediscover_address,
)
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.core import HomeAssistant
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Victron BLE device from a config entry."""
address = entry.unique_id
assert address is not None
key = entry.data[CONF_ACCESS_TOKEN]
data = VictronBluetoothDeviceData(key)
coordinator = PassiveBluetoothProcessorCoordinator(
hass,
_LOGGER,
address=address,
mode=BluetoothScanningMode.ACTIVE,
update_method=data.update,
)
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, [Platform.SENSOR])
entry.async_on_unload(coordinator.async_start())
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(
entry, [Platform.SENSOR]
)
if unload_ok:
async_rediscover_address(hass, entry.entry_id)
return unload_ok

View File

@@ -0,0 +1,123 @@
"""Config flow for Victron Bluetooth Low Energy integration."""
from __future__ import annotations
import logging
from typing import Any
from victron_ble_ha_parser import VictronBluetoothDeviceData
import voluptuous as vol
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_ADDRESS
from .const import DOMAIN, VICTRON_IDENTIFIER
_LOGGER = logging.getLogger(__name__)
STEP_ACCESS_TOKEN_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_ACCESS_TOKEN): str,
}
)
class VictronBLEConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Victron Bluetooth Low Energy."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovered_device: str | None = None
self._discovered_devices: dict[str, str] = {}
self._discovered_devices_info: dict[str, BluetoothServiceInfoBleak] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> ConfigFlowResult:
"""Handle the bluetooth discovery step."""
_LOGGER.debug("async_step_bluetooth: %s", discovery_info.address)
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
device = VictronBluetoothDeviceData()
if not device.supported(discovery_info):
_LOGGER.debug("device %s not supported", discovery_info.address)
return self.async_abort(reason="not_supported")
self._discovered_device = discovery_info.address
self._discovered_devices_info[discovery_info.address] = discovery_info
self._discovered_devices[discovery_info.address] = discovery_info.name
self.context["title_placeholders"] = {"title": discovery_info.name}
return await self.async_step_access_token()
async def async_step_access_token(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle advertisement key input."""
# should only be called if there are discovered devices
assert self._discovered_device is not None
discovery_info = self._discovered_devices_info[self._discovered_device]
title = discovery_info.name
if user_input is not None:
# see if we can create a device with the access token
device = VictronBluetoothDeviceData(user_input[CONF_ACCESS_TOKEN])
if device.validate_advertisement_key(
discovery_info.manufacturer_data[VICTRON_IDENTIFIER]
):
return self.async_create_entry(
title=title,
data=user_input,
)
return self.async_abort(reason="invalid_access_token")
return self.async_show_form(
step_id="access_token",
data_schema=STEP_ACCESS_TOKEN_DATA_SCHEMA,
description_placeholders={"title": title},
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle select a device to set up."""
if user_input is not None:
address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
self._discovered_device = address
title = self._discovered_devices_info[address].name
return self.async_show_form(
step_id="access_token",
data_schema=STEP_ACCESS_TOKEN_DATA_SCHEMA,
description_placeholders={"title": title},
)
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass, False):
address = discovery_info.address
if address in current_addresses or address in self._discovered_devices:
continue
device = VictronBluetoothDeviceData()
if device.supported(discovery_info):
self._discovered_devices_info[address] = discovery_info
self._discovered_devices[address] = discovery_info.name
if len(self._discovered_devices) < 1:
return self.async_abort(reason="no_devices_found")
_LOGGER.debug("Discovered %s devices", len(self._discovered_devices))
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_ADDRESS): vol.In(self._discovered_devices)}
),
)

View File

@@ -0,0 +1,4 @@
"""Constants for the Victron Bluetooth Low Energy integration."""
DOMAIN = "victron_ble"
VICTRON_IDENTIFIER = 0x02E1

View File

@@ -0,0 +1,19 @@
{
"domain": "victron_ble",
"name": "Victron BLE",
"bluetooth": [
{
"connectable": false,
"manufacturer_data_start": [16],
"manufacturer_id": 737
}
],
"codeowners": ["@rajlaud"],
"config_flow": true,
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/victron_ble",
"integration_type": "device",
"iot_class": "local_push",
"quality_scale": "bronze",
"requirements": ["victron-ble-ha-parser==0.4.9"]
}

View File

@@ -0,0 +1,85 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling:
status: exempt
comment: |
This 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: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup:
status: exempt
comment: |
There is nothing to test, the integration just passively receives BLE advertisements.
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: This integration does not provide additional actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: No options to configure
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates:
status: done
reauthentication-flow:
status: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
This integration does not use IP addresses. Bluetooth MAC addresses do not change.
discovery: done
docs-data-update: done
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: |
This integration has a fixed single device per instance, and each device needs a user-supplied encryption key to set up.
entity-category: done
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices:
status: exempt
comment: |
This integration has a fixed single device.
# Platinum
async-dependency: todo
inject-websession: todo
strict-typing: todo

View File

@@ -0,0 +1,474 @@
"""Sensor platform for Victron BLE."""
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Any
from sensor_state_data import DeviceKey
from victron_ble_ha_parser import Keys, Units
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothDataProcessor,
PassiveBluetoothDataUpdate,
PassiveBluetoothEntityKey,
PassiveBluetoothProcessorEntity,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfPower,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.sensor import sensor_device_info_to_hass_device_info
LOGGER = logging.getLogger(__name__)
AC_IN_OPTIONS = [
"ac_in_1",
"ac_in_2",
"not_connected",
]
ALARM_OPTIONS = [
"low_voltage",
"high_voltage",
"low_soc",
"low_starter_voltage",
"high_starter_voltage",
"low_temperature",
"high_temperature",
"mid_voltage",
"overload",
"dc_ripple",
"low_v_ac_out",
"high_v_ac_out",
"short_circuit",
"bms_lockout",
]
CHARGER_ERROR_OPTIONS = [
"no_error",
"temperature_battery_high",
"voltage_high",
"remote_temperature_auto_reset",
"remote_temperature_not_auto_reset",
"remote_battery",
"high_ripple",
"temperature_battery_low",
"temperature_charger",
"over_current",
"bulk_time",
"current_sensor",
"internal_temperature",
"fan",
"overheated",
"short_circuit",
"converter_issue",
"over_charge",
"input_voltage",
"input_current",
"input_power",
"input_shutdown_voltage",
"input_shutdown_current",
"input_shutdown_failure",
"inverter_shutdown_pv_isolation",
"inverter_shutdown_ground_fault",
"inverter_overload",
"inverter_temperature",
"inverter_peak_current",
"inverter_output_voltage",
"inverter_self_test",
"inverter_ac",
"communication",
"synchronisation",
"bms",
"network",
"pv_input_shutdown",
"cpu_temperature",
"calibration_lost",
"firmware",
"settings",
"tester_fail",
"internal_dc_voltage",
"self_test",
"internal_supply",
]
def error_to_state(value: float | str | None) -> str | None:
"""Convert error code to state string."""
value_map: dict[Any, str] = {
"internal_supply_a": "internal_supply",
"internal_supply_b": "internal_supply",
"internal_supply_c": "internal_supply",
"internal_supply_d": "internal_supply",
"inverter_shutdown_41": "inverter_shutdown_pv_isolation",
"inverter_shutdown_42": "inverter_shutdown_pv_isolation",
"inverter_shutdown_43": "inverter_shutdown_ground_fault",
"internal_temperature_a": "internal_temperature",
"internal_temperature_b": "internal_temperature",
"inverter_output_voltage_a": "inverter_output_voltage",
"inverter_output_voltage_b": "inverter_output_voltage",
"internal_dc_voltage_a": "internal_dc_voltage",
"internal_dc_voltage_b": "internal_dc_voltage",
"remote_temperature_a": "remote_temperature_auto_reset",
"remote_temperature_b": "remote_temperature_auto_reset",
"remote_temperature_c": "remote_temperature_not_auto_reset",
"remote_battery_a": "remote_battery",
"remote_battery_b": "remote_battery",
"remote_battery_c": "remote_battery",
"pv_input_shutdown_80": "pv_input_shutdown",
"pv_input_shutdown_81": "pv_input_shutdown",
"pv_input_shutdown_82": "pv_input_shutdown",
"pv_input_shutdown_83": "pv_input_shutdown",
"pv_input_shutdown_84": "pv_input_shutdown",
"pv_input_shutdown_85": "pv_input_shutdown",
"pv_input_shutdown_86": "pv_input_shutdown",
"pv_input_shutdown_87": "pv_input_shutdown",
"inverter_self_test_a": "inverter_self_test",
"inverter_self_test_b": "inverter_self_test",
"inverter_self_test_c": "inverter_self_test",
"network_a": "network",
"network_b": "network",
"network_c": "network",
"network_d": "network",
}
return value_map.get(value)
DEVICE_STATE_OPTIONS = [
"off",
"low_power",
"fault",
"bulk",
"absorption",
"float",
"storage",
"equalize_manual",
"inverting",
"power_supply",
"starting_up",
"repeated_absorption",
"recondition",
"battery_safe",
"active",
"external_control",
"not_available",
]
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class VictronBLESensorEntityDescription(SensorEntityDescription):
"""Describes Victron BLE sensor entity."""
value_fn: Callable[[float | int | str | None], float | int | str | None] = (
lambda x: x
)
SENSOR_DESCRIPTIONS = {
Keys.AC_IN_POWER: VictronBLESensorEntityDescription(
key=Keys.AC_IN_POWER,
translation_key=Keys.AC_IN_POWER,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.AC_IN_STATE: VictronBLESensorEntityDescription(
key=Keys.AC_IN_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key="ac_in_state",
options=AC_IN_OPTIONS,
),
Keys.AC_OUT_POWER: VictronBLESensorEntityDescription(
key=Keys.AC_OUT_POWER,
translation_key=Keys.AC_OUT_POWER,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.AC_OUT_STATE: VictronBLESensorEntityDescription(
key=Keys.AC_OUT_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key="device_state",
options=DEVICE_STATE_OPTIONS,
),
Keys.ALARM: VictronBLESensorEntityDescription(
key=Keys.ALARM,
device_class=SensorDeviceClass.ENUM,
translation_key="alarm",
options=ALARM_OPTIONS,
),
Keys.BALANCER_STATUS: VictronBLESensorEntityDescription(
key=Keys.BALANCER_STATUS,
device_class=SensorDeviceClass.ENUM,
translation_key="balancer_status",
options=["balanced", "balancing", "imbalance"],
),
Keys.BATTERY_CURRENT: VictronBLESensorEntityDescription(
key=Keys.BATTERY_CURRENT,
translation_key=Keys.BATTERY_CURRENT,
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.BATTERY_TEMPERATURE: VictronBLESensorEntityDescription(
key=Keys.BATTERY_TEMPERATURE,
translation_key=Keys.BATTERY_TEMPERATURE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.BATTERY_VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.BATTERY_VOLTAGE,
translation_key=Keys.BATTERY_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.CHARGER_ERROR: VictronBLESensorEntityDescription(
key=Keys.CHARGER_ERROR,
device_class=SensorDeviceClass.ENUM,
translation_key="charger_error",
options=CHARGER_ERROR_OPTIONS,
value_fn=error_to_state,
),
Keys.CONSUMED_AMPERE_HOURS: VictronBLESensorEntityDescription(
key=Keys.CONSUMED_AMPERE_HOURS,
translation_key=Keys.CONSUMED_AMPERE_HOURS,
native_unit_of_measurement=Units.ELECTRIC_CURRENT_FLOW_AMPERE_HOUR,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.CURRENT: VictronBLESensorEntityDescription(
key=Keys.CURRENT,
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.DEVICE_STATE: VictronBLESensorEntityDescription(
key=Keys.DEVICE_STATE,
device_class=SensorDeviceClass.ENUM,
translation_key="device_state",
options=DEVICE_STATE_OPTIONS,
),
Keys.ERROR_CODE: VictronBLESensorEntityDescription(
key=Keys.ERROR_CODE,
device_class=SensorDeviceClass.ENUM,
translation_key="charger_error",
options=CHARGER_ERROR_OPTIONS,
),
Keys.EXTERNAL_DEVICE_LOAD: VictronBLESensorEntityDescription(
key=Keys.EXTERNAL_DEVICE_LOAD,
translation_key=Keys.EXTERNAL_DEVICE_LOAD,
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.INPUT_VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.INPUT_VOLTAGE,
translation_key=Keys.INPUT_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.METER_TYPE: VictronBLESensorEntityDescription(
key=Keys.METER_TYPE,
device_class=SensorDeviceClass.ENUM,
translation_key="meter_type",
options=[
"solar_charger",
"wind_charger",
"shaft_generator",
"alternator",
"fuel_cell",
"water_generator",
"dc_dc_charger",
"ac_charger",
"generic_source",
"generic_load",
"electric_drive",
"fridge",
"water_pump",
"bilge_pump",
"dc_system",
"inverter",
"water_heater",
],
),
Keys.MIDPOINT_VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.MIDPOINT_VOLTAGE,
translation_key=Keys.MIDPOINT_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.OFF_REASON: VictronBLESensorEntityDescription(
key=Keys.OFF_REASON,
device_class=SensorDeviceClass.ENUM,
translation_key="off_reason",
options=[
"no_reason",
"no_input_power",
"switched_off_switch",
"switched_off_register",
"remote_input",
"protection_active",
"pay_as_you_go_out_of_credit",
"bms",
"engine_shutdown",
"analysing_input_voltage",
],
),
Keys.OUTPUT_VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.OUTPUT_VOLTAGE,
translation_key=Keys.OUTPUT_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.REMAINING_MINUTES: VictronBLESensorEntityDescription(
key=Keys.REMAINING_MINUTES,
translation_key=Keys.REMAINING_MINUTES,
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.MINUTES,
state_class=SensorStateClass.MEASUREMENT,
),
SensorDeviceClass.SIGNAL_STRENGTH: VictronBLESensorEntityDescription(
key=SensorDeviceClass.SIGNAL_STRENGTH.value,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.SOLAR_POWER: VictronBLESensorEntityDescription(
key=Keys.SOLAR_POWER,
translation_key=Keys.SOLAR_POWER,
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.STARTER_VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.STARTER_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.STATE_OF_CHARGE: VictronBLESensorEntityDescription(
key=Keys.STATE_OF_CHARGE,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.TEMPERATURE: VictronBLESensorEntityDescription(
key=Keys.TEMPERATURE,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.VOLTAGE: VictronBLESensorEntityDescription(
key=Keys.VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
),
Keys.WARNING: VictronBLESensorEntityDescription(
key=Keys.WARNING,
device_class=SensorDeviceClass.ENUM,
translation_key="alarm",
options=ALARM_OPTIONS,
),
Keys.YIELD_TODAY: VictronBLESensorEntityDescription(
key=Keys.YIELD_TODAY,
translation_key=Keys.YIELD_TODAY,
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
),
}
for i in range(1, 8):
cell_key = getattr(Keys, f"CELL_{i}_VOLTAGE")
SENSOR_DESCRIPTIONS[cell_key] = VictronBLESensorEntityDescription(
key=cell_key,
translation_key="cell_voltage",
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
)
def _device_key_to_bluetooth_entity_key(
device_key: DeviceKey,
) -> PassiveBluetoothEntityKey:
"""Convert a device key to an entity key."""
return PassiveBluetoothEntityKey(device_key.key, device_key.device_id)
def sensor_update_to_bluetooth_data_update(
sensor_update,
) -> PassiveBluetoothDataUpdate:
"""Convert a sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
device_id: sensor_device_info_to_hass_device_info(device_info)
for device_id, device_info in sensor_update.devices.items()
},
entity_descriptions={
_device_key_to_bluetooth_entity_key(device_key): SENSOR_DESCRIPTIONS[
device_key.key
]
for device_key in sensor_update.entity_descriptions
if device_key.key in SENSOR_DESCRIPTIONS
},
entity_data={
_device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value
for device_key, sensor_values in sensor_update.entity_values.items()
if device_key.key in SENSOR_DESCRIPTIONS
},
entity_names={},
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Victron BLE sensor."""
coordinator = entry.runtime_data
processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update)
entry.async_on_unload(
processor.async_add_entities_listener(
VictronBLESensorEntity, async_add_entities
)
)
entry.async_on_unload(coordinator.async_register_processor(processor))
class VictronBLESensorEntity(PassiveBluetoothProcessorEntity, SensorEntity):
"""Representation of Victron BLE sensor."""
entity_description: VictronBLESensorEntityDescription
@property
def native_value(self) -> float | int | str | None:
"""Return the state of the sensor."""
value = self.processor.entity_data.get(self.entity_key)
return self.entity_description.value_fn(value)

View File

@@ -0,0 +1,234 @@
{
"common": {
"high_voltage": "High voltage",
"low_voltage": "Low voltage",
"midpoint_voltage": "Midpoint voltage",
"starter_voltage": "Starter voltage"
},
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_access_token": "Invalid encryption key for instant readout",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
},
"flow_title": "{title}",
"step": {
"access_token": {
"data": {
"access_token": "The encryption key for instant readout of the Victron device."
},
"data_description": {
"access_token": "The encryption key for instant readout may be found in the VictronConnect app under Settings > Product info > Instant readout details > Encryption data > Encryption Key."
},
"title": "{title}"
},
"user": {
"data": {
"address": "The Bluetooth address of the Victron device."
},
"data_description": {
"address": "This Bluetooth address is automatically discovered. You may view a device's Bluetooth address in the VictronConnect app under Settings > Product info > Instant readout details > Encryption data > MAC Address."
}
}
}
},
"entity": {
"sensor": {
"ac_in_power": {
"name": "AC-in power"
},
"ac_in_state": {
"name": "AC-in state",
"state": {
"ac_in_1": "AC-in 1",
"ac_in_2": "AC-in 2",
"not_connected": "Not connected"
}
},
"ac_out_power": {
"name": "AC-out power"
},
"alarm": {
"name": "Alarm",
"state": {
"bms_lockout": "Battery management system lockout",
"dc_ripple": "DC ripple",
"high_starter_voltage": "High starter voltage",
"high_temperature": "High temperature",
"high_v_ac_out": "AC-out overvoltage",
"high_voltage": "Overvoltage",
"low_soc": "Low state of charge",
"low_starter_voltage": "Low starter voltage",
"low_temperature": "Low temperature",
"low_v_ac_out": "AC-out undervoltage",
"low_voltage": "Undervoltage",
"mid_voltage": "[%key:component::victron_ble::common::midpoint_voltage%]",
"overload": "Overload",
"short_circuit": "Short circuit"
}
},
"balancer_status": {
"name": "Balancer status",
"state": {
"balanced": "Balanced",
"balancing": "Balancing",
"imbalance": "Imbalance"
}
},
"battery_current": {
"name": "Battery current"
},
"battery_temperature": {
"name": "Battery temperature"
},
"battery_voltage": {
"name": "Battery voltage"
},
"cell_voltage": {
"name": "Cell {cell} voltage"
},
"charger_error": {
"name": "Charger error",
"state": {
"bms": "BMS connection lost",
"bulk_time": "Bulk time limit exceeded",
"calibration_lost": "Factory calibration data lost",
"communication": "Communication warning",
"converter_issue": "Converter issue",
"cpu_temperature": "CPU temperature too high",
"current_sensor": "Current sensor issue",
"fan": "Fan failure",
"firmware": "Invalid or incompatible firmware",
"high_ripple": "Battery high ripple voltage",
"input_current": "Input overcurrent",
"input_power": "Input overpower",
"input_shutdown_current": "Input shutdown (current flow during off mode)",
"input_shutdown_failure": "PV input failed to shutdown",
"input_shutdown_voltage": "Input shutdown (battery overvoltage)",
"input_voltage": "Input overvoltage",
"internal_dc_voltage": "Internal DC voltage error",
"internal_supply": "Internal supply voltage error",
"internal_temperature": "Internal temperature sensor failure",
"inverter_ac": "Inverter AC voltage on output",
"inverter_output_voltage": "Inverter output voltage",
"inverter_overload": "Inverter overload",
"inverter_peak_current": "Inverter peak current",
"inverter_self_test": "Inverter self-test failed",
"inverter_shutdown_ground_fault": "Inverter shutdown (Ground fault)",
"inverter_shutdown_pv_isolation": "Inverter shutdown (PV isolation)",
"inverter_temperature": "Inverter temperature too high",
"network": "Network misconfigured",
"no_error": "No error",
"over_charge": "Overcharge protection",
"over_current": "Charger overcurrent",
"overheated": "Terminals overheated",
"pv_input_shutdown": "PV input shutdown",
"remote_battery": "Remote battery voltage sense failure",
"remote_temperature_auto_reset": "Remote temperature sensor failure (auto-reset)",
"remote_temperature_not_auto_reset": "Remote temperature sensor failure (not auto-reset)",
"self_test": "PV residual current sensor self-test failure",
"settings": "Settings data lost",
"short_circuit": "Charger short circuit",
"synchronisation": "Synchronized charging device configuration issue",
"temperature_battery_high": "Battery temperature too high",
"temperature_battery_low": "Battery temperature too low",
"temperature_charger": "Charger temperature too high",
"tester_fail": "Tester fail",
"voltage_high": "Battery overvoltage"
}
},
"consumed_ampere_hours": {
"name": "Consumed ampere hours"
},
"device_state": {
"name": "Device state",
"state": {
"absorption": "Absorption",
"active": "Active",
"battery_safe": "Battery safe",
"bulk": "Bulk",
"equalize_manual": "Equalize (manual)",
"external_control": "External control",
"fault": "Fault",
"float": "Float",
"inverting": "Inverting",
"low_power": "Low power",
"not_available": "Not available",
"off": "[%key:common::state::off%]",
"power_supply": "Power supply",
"recondition": "Recondition",
"repeated_absorption": "Repeated absorption",
"starting_up": "Starting up",
"storage": "Storage"
}
},
"error_code": {
"name": "Error code"
},
"external_device_load": {
"name": "External device load"
},
"input_voltage": {
"name": "Input voltage"
},
"meter_type": {
"name": "Meter type",
"state": {
"ac_charger": "AC charger",
"alternator": "Alternator",
"bilge_pump": "Bilge pump",
"dc_dc_charger": "DC-DC charger",
"dc_system": "DC system",
"electric_drive": "Electric drive",
"fridge": "Fridge",
"fuel_cell": "Fuel cell",
"generic_load": "Generic load",
"generic_source": "Generic source",
"inverter": "Inverter",
"shaft_generator": "Shaft generator",
"solar_charger": "Solar charger",
"water_generator": "Water generator",
"water_heater": "Water heater",
"water_pump": "Water pump",
"wind_charger": "Wind charger"
}
},
"midpoint_voltage": {
"name": "[%key:component::victron_ble::common::midpoint_voltage%]"
},
"off_reason": {
"name": "Off reason",
"state": {
"analysing_input_voltage": "Analyzing input voltage",
"bms": "Battery management system",
"engine_shutdown": "Engine shutdown",
"no_input_power": "No input power",
"no_reason": "No reason",
"pay_as_you_go_out_of_credit": "Pay-as-you-go out of credit",
"protection_active": "Protection active",
"remote_input": "Remote input",
"switched_off_register": "Switched off by register",
"switched_off_switch": "Switched off by switch"
}
},
"output_voltage": {
"name": "Output voltage"
},
"remaining_minutes": {
"name": "Remaining minutes"
},
"solar_power": {
"name": "Solar power"
},
"starter_voltage": {
"name": "[%key:component::victron_ble::common::starter_voltage%]"
},
"warning": {
"name": "Warning"
},
"yield_today": {
"name": "Yield today"
}
}
}
}

View File

@@ -849,6 +849,14 @@ BLUETOOTH: Final[list[dict[str, bool | str | int | list[int]]]] = [
"manufacturer_id": 34714,
"service_uuid": "0000cee0-0000-1000-8000-00805f9b34fb",
},
{
"connectable": False,
"domain": "victron_ble",
"manufacturer_data_start": [
16,
],
"manufacturer_id": 737,
},
{
"connectable": False,
"domain": "xiaomi_ble",

View File

@@ -723,6 +723,7 @@ FLOWS = {
"version",
"vesync",
"vicare",
"victron_ble",
"victron_remote_monitoring",
"vilfo",
"vizio",

View File

@@ -7282,11 +7282,22 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"victron_remote_monitoring": {
"name": "Victron Remote Monitoring",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
"victron": {
"name": "Victron",
"integrations": {
"victron_ble": {
"integration_type": "device",
"config_flow": true,
"iot_class": "local_push",
"name": "Victron BLE"
},
"victron_remote_monitoring": {
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Victron Remote Monitoring"
}
}
},
"vilfo": {
"name": "Vilfo Router",

3
requirements_all.txt generated
View File

@@ -3088,6 +3088,9 @@ velbus-aio==2025.11.0
# homeassistant.components.venstar
venstarcolortouch==0.21
# homeassistant.components.victron_ble
victron-ble-ha-parser==0.4.9
# homeassistant.components.victron_remote_monitoring
victron-vrm==0.1.8

View File

@@ -2555,6 +2555,9 @@ velbus-aio==2025.11.0
# homeassistant.components.venstar
venstarcolortouch==0.21
# homeassistant.components.victron_ble
victron-ble-ha-parser==0.4.9
# homeassistant.components.victron_remote_monitoring
victron-vrm==0.1.8

View File

@@ -0,0 +1 @@
"""Tests for the Victron Bluetooth Low Energy integration."""

View File

@@ -0,0 +1,75 @@
"""Test the Victron Bluetooth Low Energy config flow."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from home_assistant_bluetooth import BluetoothServiceInfo
import pytest
from homeassistant.components.victron_ble.const import DOMAIN
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_ADDRESS
from homeassistant.core import HomeAssistant
from .fixtures import VICTRON_VEBUS_SERVICE_INFO, VICTRON_VEBUS_TOKEN
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.victron_ble.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_discovered_service_info() -> Generator[AsyncMock]:
"""Mock discovered service info."""
with patch(
"homeassistant.components.victron_ble.config_flow.async_discovered_service_info",
return_value=[VICTRON_VEBUS_SERVICE_INFO],
) as mock_discovered_service_info:
yield mock_discovered_service_info
@pytest.fixture
def service_info() -> BluetoothServiceInfo:
"""Return service info."""
return VICTRON_VEBUS_SERVICE_INFO
@pytest.fixture
def access_token() -> str:
"""Return access token."""
return VICTRON_VEBUS_TOKEN
@pytest.fixture
def mock_config_entry(
service_info: BluetoothServiceInfo, access_token: str
) -> MockConfigEntry:
"""Mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: service_info.address,
CONF_ACCESS_TOKEN: access_token,
},
unique_id=service_info.address,
)
@pytest.fixture
def mock_config_entry_added_to_hass(
mock_config_entry,
hass: HomeAssistant,
service_info: BluetoothServiceInfo,
access_token: str,
) -> MockConfigEntry:
"""Mock config entry factory that added to hass."""
entry = mock_config_entry
entry.add_to_hass(hass)
return entry

View File

@@ -0,0 +1,147 @@
"""Fixtures for testing victron_ble."""
from home_assistant_bluetooth import BluetoothServiceInfo
NOT_VICTRON_SERVICE_INFO = BluetoothServiceInfo(
name="Not it",
address="61DE521B-F0BF-9F44-64D4-75BBE1738105",
rssi=-63,
manufacturer_data={3234: b"\x00\x01"},
service_data={},
service_uuids=[],
source="local",
)
VICTRON_TEST_WRONG_TOKEN = "00000000000000000000000000000000"
# battery monitor
VICTRON_BATTERY_MONITOR_SERVICE_INFO = BluetoothServiceInfo(
name="Battery Monitor",
address="01:02:03:04:05:07",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("100289a302b040af925d09a4d89aa0128bdef48c6298a9")
},
service_data={},
service_uuids=[],
source="local",
)
VICTRON_BATTERY_MONITOR_TOKEN = "aff4d0995b7d1e176c0c33ecb9e70dcd"
VICTRON_BATTERY_MONITOR_SENSORS = {
"battery_monitor_aux_mode": "disabled",
"battery_monitor_consumed_ampere_hours": "-50.0",
"battery_monitor_current": "0.0",
"battery_monitor_remaining_minutes": "unknown",
"battery_monitor_state_of_charge": "50.0",
"battery_monitor_voltage": "12.53",
"battery_monitor_alarm": "none",
"battery_monitor_temperature": "unknown",
"battery_monitor_starter_voltage": "unknown",
"battery_monitor_midpoint_voltage": "unknown",
}
# DC/DC converter
VICTRON_DC_DC_CONVERTER_SERVICE_INFO = BluetoothServiceInfo(
name="DC/DC Converter",
address="01:02:03:04:05:08",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("1000c0a304121d64ca8d442b90bbdf6a8cba"),
},
service_data={},
service_uuids=[],
source="local",
)
# DC energy meter
VICTRON_DC_ENERGY_METER_SERVICE_INFO = BluetoothServiceInfo(
name="DC Energy Meter",
address="01:02:03:04:05:09",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("100289a30d787fafde83ccec982199fd815286"),
},
service_data={},
service_uuids=[],
source="local",
)
VICTRON_DC_ENERGY_METER_TOKEN = "aff4d0995b7d1e176c0c33ecb9e70dcd"
VICTRON_DC_ENERGY_METER_SENSORS = {
"dc_energy_meter_meter_type": "dc_dc_charger",
"dc_energy_meter_aux_mode": "starter_voltage",
"dc_energy_meter_current": "0.0",
"dc_energy_meter_voltage": "12.52",
"dc_energy_meter_starter_voltage": "-0.01",
"dc_energy_meter_alarm": "none",
"dc_energy_meter_temperature": "unknown",
}
# Inverter
VICTRON_INVERTER_SERVICE_INFO = BluetoothServiceInfo(
name="Inverter",
address="01:02:03:04:05:10",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("1003a2a2031252dad26f0b8eb39162074d140df410"),
}, # not a valid advertisement, but model id mangled to match inverter
service_data={},
service_uuids=[],
source="local",
)
# Solar charger
VICTRON_SOLAR_CHARGER_SERVICE_INFO = BluetoothServiceInfo(
name="Solar Charger",
address="01:02:03:04:05:11",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("100242a0016207adceb37b605d7e0ee21b24df5c"),
},
service_data={},
service_uuids=[],
source="local",
)
VICTRON_SOLAR_CHARGER_TOKEN = "adeccb947395801a4dd45a2eaa44bf17"
VICTRON_SOLAR_CHARGER_SENSORS = {
"solar_charger_charge_state": "absorption",
"solar_charger_battery_voltage": "13.88",
"solar_charger_battery_current": "1.4",
"solar_charger_yield_today": "30",
"solar_charger_solar_power": "19",
"solar_charger_external_device_load": "0.0",
}
# ve.bus
VICTRON_VEBUS_SERVICE_INFO = BluetoothServiceInfo(
name="Inverter Charger",
address="01:02:03:04:05:06",
rssi=-60,
manufacturer_data={
0x02E1: bytes.fromhex("100380270c1252dad26f0b8eb39162074d140df410")
},
service_data={},
service_uuids=[],
source="local",
)
VICTRON_VEBUS_TOKEN = "da3f5fa2860cb1cf86ba7a6d1d16b9dd"
VICTRON_VEBUS_SENSORS = {
"inverter_charger_device_state": "float",
"inverter_charger_battery_voltage": "14.45",
"inverter_charger_battery_current": "23.2",
"inverter_charger_ac_in_state": "AC_IN_1",
"inverter_charger_ac_in_power": "1459",
"inverter_charger_ac_out_power": "1046",
"inverter_charger_battery_temperature": "32",
"inverter_charger_state_of_charge": "unknown",
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,189 @@
"""Test the Victron Bluetooth Low Energy config flow."""
from unittest.mock import AsyncMock
from home_assistant_bluetooth import BluetoothServiceInfo
import pytest
from homeassistant import config_entries
from homeassistant.components.victron_ble.const import DOMAIN
from homeassistant.config_entries import SOURCE_BLUETOOTH, SOURCE_USER
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .fixtures import (
NOT_VICTRON_SERVICE_INFO,
VICTRON_INVERTER_SERVICE_INFO,
VICTRON_TEST_WRONG_TOKEN,
VICTRON_VEBUS_SERVICE_INFO,
VICTRON_VEBUS_TOKEN,
)
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth: None) -> None:
"""Mock bluetooth for all tests in this module."""
async def test_async_step_bluetooth_valid_device(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=VICTRON_VEBUS_SERVICE_INFO,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "access_token"
# test valid access token
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_ACCESS_TOKEN: VICTRON_VEBUS_TOKEN},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == VICTRON_VEBUS_SERVICE_INFO.name
flow_result = result.get("result")
assert flow_result is not None
assert flow_result.unique_id == VICTRON_VEBUS_SERVICE_INFO.address
assert flow_result.data == {
CONF_ACCESS_TOKEN: VICTRON_VEBUS_TOKEN,
}
assert set(flow_result.data.keys()) == {CONF_ACCESS_TOKEN}
@pytest.mark.parametrize(
("source", "service_info", "expected_reason"),
[
(
SOURCE_BLUETOOTH,
NOT_VICTRON_SERVICE_INFO,
"not_supported",
),
(
SOURCE_BLUETOOTH,
VICTRON_INVERTER_SERVICE_INFO,
"not_supported",
),
(
SOURCE_USER,
None,
"no_devices_found",
),
],
ids=["bluetooth_not_victron", "bluetooth_unsupported_device", "user_no_devices"],
)
async def test_abort_scenarios(
hass: HomeAssistant,
source: str,
service_info: BluetoothServiceInfo | None,
expected_reason: str,
) -> None:
"""Test flows that result in abort."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": source},
data=service_info,
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == expected_reason
async def test_async_step_user_with_devices_found(
hass: HomeAssistant, mock_discovered_service_info: AsyncMock
) -> None:
"""Test setup from service info cache with devices found."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_ADDRESS: VICTRON_VEBUS_SERVICE_INFO.address},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "access_token"
# test invalid access token (valid already tested above)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_ACCESS_TOKEN: VICTRON_TEST_WRONG_TOKEN}
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "invalid_access_token"
async def test_async_step_user_device_added_between_steps(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_discovered_service_info: AsyncMock,
) -> None:
"""Test abort when the device gets added via another flow between steps."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": VICTRON_VEBUS_SERVICE_INFO.address},
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_configured"
async def test_async_step_user_with_found_devices_already_setup(
hass: HomeAssistant,
mock_config_entry_added_to_hass: MockConfigEntry,
mock_discovered_service_info: AsyncMock,
) -> None:
"""Test setup from service info cache with devices found."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "no_devices_found"
async def test_async_step_bluetooth_devices_already_setup(
hass: HomeAssistant, mock_config_entry_added_to_hass: MockConfigEntry
) -> None:
"""Test we can't start a flow if there is already a config entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VICTRON_VEBUS_SERVICE_INFO,
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_configured"
async def test_async_step_bluetooth_already_in_progress(hass: HomeAssistant) -> None:
"""Test we can't start a flow for the same device twice."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VICTRON_VEBUS_SERVICE_INFO,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "access_token"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VICTRON_VEBUS_SERVICE_INFO,
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_in_progress"

View File

@@ -0,0 +1,61 @@
"""Test updating sensors in the victron_ble integration."""
from home_assistant_bluetooth import BluetoothServiceInfo
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .fixtures import (
VICTRON_BATTERY_MONITOR_SERVICE_INFO,
VICTRON_BATTERY_MONITOR_TOKEN,
VICTRON_DC_ENERGY_METER_SERVICE_INFO,
VICTRON_DC_ENERGY_METER_TOKEN,
VICTRON_SOLAR_CHARGER_SERVICE_INFO,
VICTRON_SOLAR_CHARGER_TOKEN,
VICTRON_VEBUS_SERVICE_INFO,
VICTRON_VEBUS_TOKEN,
)
from tests.common import MockConfigEntry, snapshot_platform
from tests.components.bluetooth import inject_bluetooth_service_info
@pytest.mark.usefixtures("enable_bluetooth")
@pytest.mark.parametrize(
(
"service_info",
"access_token",
),
[
(VICTRON_BATTERY_MONITOR_SERVICE_INFO, VICTRON_BATTERY_MONITOR_TOKEN),
(VICTRON_DC_ENERGY_METER_SERVICE_INFO, VICTRON_DC_ENERGY_METER_TOKEN),
(VICTRON_SOLAR_CHARGER_SERVICE_INFO, VICTRON_SOLAR_CHARGER_TOKEN),
(VICTRON_VEBUS_SERVICE_INFO, VICTRON_VEBUS_TOKEN),
],
ids=["battery_monitor", "dc_energy_meter", "solar_charger", "vebus"],
)
async def test_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry_added_to_hass: MockConfigEntry,
service_info: BluetoothServiceInfo,
access_token: str,
) -> None:
"""Test sensor entities."""
entry = mock_config_entry_added_to_hass
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Initially no entities should be created until bluetooth data is received
assert len(hass.states.async_all()) == 0
# Inject bluetooth service info to trigger entity creation
inject_bluetooth_service_info(hass, service_info)
await hass.async_block_till_done()
# Use snapshot testing to verify all entity states and registry entries
await snapshot_platform(hass, entity_registry, snapshot, entry.entry_id)