Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions custom_components/victron/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,14 @@

from __future__ import annotations

from typing import Any

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant

from .const import CONF_HOST, CONF_INTERVAL, CONF_PORT, DOMAIN, SCAN_REGISTERS
from .coordinator import victronEnergyDeviceUpdateCoordinator as Coordinator
from .coordinator import VictronEnergyDeviceUpdateCoordinator as Coordinator

PLATFORMS: list[Platform] = [
Platform.SENSOR,
Expand All @@ -28,7 +30,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
# TODO 3. Store an API object for your platforms to access
# hass.data[DOMAIN][entry.entry_id] = MyApi(...)

coordinator = Coordinator(
coordinator_ = Coordinator(
hass,
config_entry.options[CONF_HOST],
config_entry.options[CONF_PORT],
Expand All @@ -43,16 +45,16 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b

# Finalize
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = coordinator
hass.data[DOMAIN][config_entry.entry_id] = coordinator_

await coordinator.async_config_entry_first_refresh()
await coordinator_.async_config_entry_first_refresh()
config_entry.async_on_unload(config_entry.add_update_listener(update_listener))
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)

return True


async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> Any:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS
Expand Down
13 changes: 7 additions & 6 deletions custom_components/victron/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,26 +2,27 @@

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any

from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.typing import StateType


@dataclass
class VictronBaseEntityDescription(EntityDescription):
@dataclass(frozen=True)
class VictronBaseEntityDescription(EntityDescription): # type: ignore[misc]
"""An extension of EntityDescription for Victron components."""

@staticmethod
def lambda_func():
def lambda_func() -> Any:
"""Return an entitydescription."""
return lambda data, slave, key: data["data"][str(slave) + "." + str(key)]

slave: int = None
slave: int | None = None
value_fn: Callable[[dict], StateType] = lambda_func()


@dataclass
@dataclass(frozen=True)
class VictronWriteBaseEntityDescription(VictronBaseEntityDescription):
"""An extension of VictronBaseEntityDescription for writeable Victron components."""

address: int = None
address: int | None = None
46 changes: 25 additions & 21 deletions custom_components/victron/binary_sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

from __future__ import annotations

from collections import OrderedDict
from dataclasses import dataclass
import logging
from typing import cast
from typing import Any, cast

from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
Expand All @@ -19,7 +20,7 @@

from .base import VictronBaseEntityDescription
from .const import DOMAIN, BoolReadEntityType, register_info_dict
from .coordinator import victronEnergyDeviceUpdateCoordinator
from .coordinator import VictronEnergyDeviceUpdateCoordinator

_LOGGER = logging.getLogger(__name__)

Expand All @@ -31,54 +32,57 @@ async def async_setup_entry(
) -> None:
"""Set up Victron energy binary sensor entries."""
_LOGGER.debug("attempting to setup binary sensor entities")
victron_coordinator: victronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][
victron_coordinator: VictronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
_LOGGER.debug(victron_coordinator.processed_data()["register_set"])
_LOGGER.debug(victron_coordinator.processed_data()["data"])
descriptions = []
# TODO cleanup
register_set = victron_coordinator.processed_data()["register_set"]
register_set: dict[str, OrderedDict] = victron_coordinator.processed_data()[
"register_set"
]
for slave, registerLedger in register_set.items():
for name in registerLedger:
for register_name, registerInfo in register_info_dict[name].items():
_LOGGER.debug(
"unit == %s registerLedger == %s registerInfo",
"unit == %s registerLedger == %s register_info",
slave,
registerLedger,
)

if isinstance(registerInfo.entityType, BoolReadEntityType):
if isinstance(registerInfo.entity_type, BoolReadEntityType):
description = VictronEntityDescription(
key=register_name,
name=register_name.replace("_", " "),
slave=slave,
slave=slave, # type: ignore[arg-type]
)
_LOGGER.debug("composed description == %s", description)
descriptions.append(description)

entities = []
entity = {}
_entities = []
_entity: dict = {}
for description in descriptions:
entity = description
entities.append(VictronBinarySensor(victron_coordinator, entity))
_entity = description
_entities.append(VictronBinarySensor(victron_coordinator, _entity))

async_add_entities(entities, True)
async_add_entities(_entities, True)


@dataclass
@dataclass(frozen=True)
class VictronEntityDescription(
BinarySensorEntityDescription, VictronBaseEntityDescription
BinarySensorEntityDescription, # type: ignore[misc]
VictronBaseEntityDescription,
):
"""Describes victron sensor entity."""


class VictronBinarySensor(CoordinatorEntity, BinarySensorEntity):
class VictronBinarySensor(CoordinatorEntity, BinarySensorEntity): # type: ignore[misc]
"""A binary sensor implementation for Victron energy device."""

def __init__(
self,
coordinator: victronEnergyDeviceUpdateCoordinator,
coordinator: VictronEnergyDeviceUpdateCoordinator,
description: VictronEntityDescription,
) -> None:
"""Initialize the binary sensor."""
Expand All @@ -101,15 +105,15 @@ def __init__(
@property
def is_on(self) -> bool:
"""Return True if the binary sensor is on."""
data = self.description.value_fn(
data = self.description.value_fn( # type: ignore[call-arg]
self.coordinator.processed_data(),
self.description.slave,
self.description.key,
)
return cast("bool", data)

@property
def available(self) -> bool:
def available(self) -> Any:
"""Return True if entity is available."""
full_key = str(self.description.slave) + "." + self.description.key
return self.coordinator.processed_data()["availability"][full_key]
Expand All @@ -118,8 +122,8 @@ def available(self) -> bool:
def device_info(self) -> entity.DeviceInfo:
"""Return the device info."""
return entity.DeviceInfo(
identifiers={(DOMAIN, self.unique_id.split("_")[0])},
name=self.unique_id.split("_")[1],
model=self.unique_id.split("_")[0],
identifiers={(DOMAIN, self.unique_id.split("_", maxsplit=1)[0])},
name=self.unique_id.split("_", maxsplit=1)[1],
model=self.unique_id.split("_", maxsplit=1)[0],
manufacturer="victron",
)
38 changes: 20 additions & 18 deletions custom_components/victron/button.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from dataclasses import dataclass
import logging
from typing import Any

from homeassistant.components.button import (
DOMAIN as BUTTON_DOMAIN,
Expand All @@ -19,7 +20,7 @@

from .base import VictronWriteBaseEntityDescription
from .const import CONF_ADVANCED_OPTIONS, DOMAIN, ButtonWriteType, register_info_dict
from .coordinator import victronEnergyDeviceUpdateCoordinator
from .coordinator import VictronEnergyDeviceUpdateCoordinator

_LOGGER = logging.getLogger(__name__)

Expand All @@ -31,24 +32,24 @@ async def async_setup_entry(
) -> None:
"""Set up Victron energy binary sensor entries."""
_LOGGER.debug("attempting to setup button entities")
victron_coordinator: victronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][
victron_coordinator: VictronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
descriptions = []
# TODO cleanup
register_set = victron_coordinator.processed_data()["register_set"]
register_set: Any = victron_coordinator.processed_data()["register_set"]
for slave, registerLedger in register_set.items():
for name in registerLedger:
for register_name, registerInfo in register_info_dict[name].items():
_LOGGER.debug(
"unit == %s registerLedger == %s registerInfo",
"unit == %s registerLedger == %s register_info",
slave,
registerLedger,
)
if not config_entry.options[CONF_ADVANCED_OPTIONS]:
continue

if isinstance(registerInfo.entityType, ButtonWriteType):
if isinstance(registerInfo.entity_type, ButtonWriteType):
description = VictronEntityDescription(
key=register_name,
name=register_name.replace("_", " "),
Expand All @@ -59,28 +60,29 @@ async def async_setup_entry(
_LOGGER.debug("composed description == %s", description)
descriptions.append(description)

entities = []
entity = {}
_entities = []
_entity: dict = {}
for description in descriptions:
entity = description
entities.append(VictronBinarySensor(victron_coordinator, entity))
_entity = description
_entities.append(VictronBinarySensor(victron_coordinator, _entity))

async_add_entities(entities, True)
async_add_entities(_entities, True)


@dataclass
@dataclass(frozen=True)
class VictronEntityDescription(
ButtonEntityDescription, VictronWriteBaseEntityDescription
ButtonEntityDescription, # type: ignore[misc]
VictronWriteBaseEntityDescription,
):
"""Describes victron sensor entity."""


class VictronBinarySensor(CoordinatorEntity, ButtonEntity):
class VictronBinarySensor(CoordinatorEntity, ButtonEntity): # type: ignore[misc]
"""A button implementation for Victron energy device."""

def __init__(
self,
coordinator: victronEnergyDeviceUpdateCoordinator,
coordinator: VictronEnergyDeviceUpdateCoordinator,
description: VictronEntityDescription,
) -> None:
"""Initialize the sensor."""
Expand All @@ -106,7 +108,7 @@ async def async_press(self) -> None:
)

@property
def available(self) -> bool:
def available(self) -> Any:
"""Return True if entity available."""
full_key = str(self.description.slave) + "." + self.description.key
return self.coordinator.processed_data()["availability"][full_key]
Expand All @@ -115,8 +117,8 @@ def available(self) -> bool:
def device_info(self) -> entity.DeviceInfo:
"""Return the device info."""
return entity.DeviceInfo(
identifiers={(DOMAIN, self.unique_id.split("_")[0])},
name=self.unique_id.split("_")[1],
model=self.unique_id.split("_")[0],
identifiers={(DOMAIN, self.unique_id.split("_", maxsplit=1)[0])},
name=self.unique_id.split("_", maxsplit=1)[1],
model=self.unique_id.split("_", maxsplit=1)[0],
manufacturer="victron",
)
Loading
Loading