Skip to content

Commit

Permalink
from PR sfstar#264: ruff format and ruff check --fix and some mypy/py…
Browse files Browse the repository at this point in the history
…lint erros and fix PR sfstar#264 and fix issues sfstar#259, sfstar#261
  • Loading branch information
julesxxl committed Dec 26, 2024
1 parent 1ec1567 commit 4b8f946
Show file tree
Hide file tree
Showing 14 changed files with 2,601 additions and 1,063 deletions.
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

0 comments on commit 4b8f946

Please sign in to comment.