Skip to content
Merged
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
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
custom_components/victron/__pycache__
.DS_Store
28 changes: 22 additions & 6 deletions custom_components/victron/__init__.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,22 @@
"""The victron integration."""

from __future__ import annotations

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

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

PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH, Platform.NUMBER, Platform.SELECT, Platform.BINARY_SENSOR, Platform.BUTTON]
PLATFORMS: list[Platform] = [
Platform.SENSOR,
Platform.SWITCH,
Platform.NUMBER,
Platform.SELECT,
Platform.BINARY_SENSOR,
Platform.BUTTON,
]


async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
Expand All @@ -20,8 +28,13 @@ 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(hass, config_entry.options[CONF_HOST], config_entry.options[CONF_PORT],
config_entry.data[SCAN_REGISTERS], config_entry.options[CONF_INTERVAL])
coordinator = Coordinator(
hass,
config_entry.options[CONF_HOST],
config_entry.options[CONF_PORT],
config_entry.data[SCAN_REGISTERS],
config_entry.options[CONF_INTERVAL],
)
# try:
# await coordinator.async_config_entry_first_refresh()
# except ConfigEntryNotReady:
Expand All @@ -41,11 +54,14 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b

async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS):
if unload_ok := await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS
):
hass.data[DOMAIN].pop(config_entry.entry_id)

return unload_ok


async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Update listener."""
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.config_entries.async_reload(config_entry.entry_id)
18 changes: 12 additions & 6 deletions custom_components/victron/base.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,20 @@
from collections.abc import Callable
from homeassistant.helpers.typing import StateType

from dataclasses import dataclass

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


@dataclass
class VictronBaseEntityDescription(EntityDescription):
slave: int = None
value_fn: Callable[[dict], StateType] = lambda data, slave, key: data["data"][str(slave) + "." + str(key)]
@staticmethod
def lambda_func():
return lambda data, slave, key: data["data"][str(slave) + "." + str(key)]

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

@dataclass

@dataclass
class VictronWriteBaseEntityDescription(VictronBaseEntityDescription):
address: int = None
address: int = None
86 changes: 47 additions & 39 deletions custom_components/victron/binary_sensor.py
Original file line number Diff line number Diff line change
@@ -1,54 +1,59 @@
"""Support for Victron Energy binary sensors."""
from __future__ import annotations

from contextlib import suppress
from typing import cast

from homeassistant.core import HomeAssistant, HassJob

from collections.abc import Callable
from homeassistant.helpers.typing import StateType
from __future__ import annotations

from dataclasses import dataclass
import logging
from typing import cast

from homeassistant.helpers.update_coordinator import CoordinatorEntity

from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.core import HassJob, HomeAssistant
from homeassistant.helpers import entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers import event, entity

from homeassistant.components.binary_sensor import BinarySensorEntityDescription, BinarySensorEntity, DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.helpers.update_coordinator import CoordinatorEntity

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

import logging
_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Victron energy binary sensor entries."""
_LOGGER.debug("attempting to setup binary sensor entities")
victron_coordinator: victronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
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
# TODO cleanup
register_set = 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 == " + str(slave) + " registerLedger == " + str(registerLedger) + " registerInfo ")
_LOGGER.debug(
"unit == "
+ str(slave)
+ " registerLedger == "
+ str(registerLedger)
+ " registerInfo "
)

if isinstance(registerInfo.entityType, BoolReadEntityType):
description = VictronEntityDescription(
key=register_name,
name=register_name.replace('_', ' '),
name=register_name.replace("_", " "),
slave=slave,
)
_LOGGER.debug("composed description == " + str(description))
Expand All @@ -58,28 +63,29 @@ async def async_setup_entry(
entity = {}
for description in descriptions:
entity = description
entities.append(
VictronBinarySensor(
victron_coordinator,
entity
))
entities.append(VictronBinarySensor(victron_coordinator, entity))

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


@dataclass
class VictronEntityDescription(BinarySensorEntityDescription, VictronBaseEntityDescription):
class VictronEntityDescription(
BinarySensorEntityDescription, VictronBaseEntityDescription
):
"""Describes victron sensor entity."""


class VictronBinarySensor(CoordinatorEntity, BinarySensorEntity):
"""A binary sensor implementation for Victron energy device."""

def __init__(self, coordinator: victronEnergyDeviceUpdateCoordinator, description: VictronEntityDescription) -> None:
def __init__(
self,
coordinator: victronEnergyDeviceUpdateCoordinator,
description: VictronEntityDescription,
) -> None:
"""Initialize the binary sensor."""
self.description: VictronEntityDescription = description
#this needs to be changed to allow multiple of the same type
# this needs to be changed to allow multiple of the same type
self._attr_device_class = description.device_class
self._attr_name = f"{description.name}"

Expand All @@ -97,7 +103,11 @@ def __init__(self, coordinator: victronEnergyDeviceUpdateCoordinator, descriptio
@property
def is_on(self) -> bool:
"""Return True if the binary sensor is on."""
data = self.description.value_fn(self.coordinator.processed_data(), self.description.slave, self.description.key)
data = self.description.value_fn(
self.coordinator.processed_data(),
self.description.slave,
self.description.key,
)
return cast(bool, data)

@property
Expand All @@ -109,10 +119,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("_")[0])},
name=self.unique_id.split("_")[1],
model=self.unique_id.split("_")[0],
manufacturer="victron",
)
)
78 changes: 45 additions & 33 deletions custom_components/victron/button.py
Original file line number Diff line number Diff line change
@@ -1,48 +1,59 @@
"""Support for Victron energy button sensors."""
from __future__ import annotations

from homeassistant.core import HomeAssistant, HassJob
from __future__ import annotations

from dataclasses import dataclass
import logging

from homeassistant.helpers.update_coordinator import CoordinatorEntity

from homeassistant.components.button import (
DOMAIN as BUTTON_DOMAIN,
ButtonDeviceClass,
ButtonEntity,
ButtonEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.core import HassJob, HomeAssistant
from homeassistant.helpers import entity

from homeassistant.components.button import ButtonEntityDescription, ButtonDeviceClass, ButtonEntity, DOMAIN as BUTTON_DOMAIN
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

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

import logging
_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Victron energy binary sensor entries."""
_LOGGER.debug("attempting to setup button entities")
victron_coordinator: victronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
victron_coordinator: victronEnergyDeviceUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
descriptions = []
#TODO cleanup
# TODO cleanup
register_set = 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 == " + str(slave) + " registerLedger == " + str(registerLedger) + " registerInfo ")
_LOGGER.debug(
"unit == "
+ str(slave)
+ " registerLedger == "
+ str(registerLedger)
+ " registerInfo "
)
if not config_entry.options[CONF_ADVANCED_OPTIONS]:
continue
continue

if isinstance(registerInfo.entityType, ButtonWriteType):
description = VictronEntityDescription(
key=register_name,
name=register_name.replace('_', ' '),
name=register_name.replace("_", " "),
slave=slave,
device_class=ButtonDeviceClass.RESTART,
address=registerInfo.register,
Expand All @@ -54,25 +65,26 @@ async def async_setup_entry(
entity = {}
for description in descriptions:
entity = description
entities.append(
VictronBinarySensor(
victron_coordinator,
entity
))
entities.append(VictronBinarySensor(victron_coordinator, entity))

async_add_entities(entities, True)

async_add_entities(
entities, True
)

@dataclass
class VictronEntityDescription(ButtonEntityDescription, VictronWriteBaseEntityDescription):
class VictronEntityDescription(
ButtonEntityDescription, VictronWriteBaseEntityDescription
):
"""Describes victron sensor entity."""


class VictronBinarySensor(CoordinatorEntity, ButtonEntity):
"""A button implementation for Victron energy device."""

def __init__(self, coordinator: victronEnergyDeviceUpdateCoordinator, description: VictronEntityDescription) -> None:
def __init__(
self,
coordinator: victronEnergyDeviceUpdateCoordinator,
description: VictronEntityDescription,
) -> None:
"""Initialize the sensor."""
self.description: VictronEntityDescription = description
self._attr_device_class = description.device_class
Expand All @@ -91,7 +103,9 @@ def __init__(self, coordinator: victronEnergyDeviceUpdateCoordinator, descriptio

async def async_press(self) -> None:
"""Handle the button press."""
self.coordinator.write_register(unit=self.description.slave, address=self.description.address, value=1)
self.coordinator.write_register(
unit=self.description.slave, address=self.description.address, value=1
)

@property
def available(self) -> bool:
Expand All @@ -102,10 +116,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],
manufacturer="victron",
)
identifiers={(DOMAIN, self.unique_id.split("_")[0])},
name=self.unique_id.split("_")[1],
model=self.unique_id.split("_")[0],
manufacturer="victron",
)
Loading