Victron preparation for ha integration submission

custom_components/victron/__init__.py

@@ -1,61 +1,47 @@
-"""The victron integration."""
+"""The Victron GX modbusTCP integration."""
 
 from __future__ import annotations
 
 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
-
-PLATFORMS: list[Platform] = [
-    Platform.SENSOR,
-    Platform.SWITCH,
-    Platform.NUMBER,
-    Platform.SELECT,
-    Platform.BINARY_SENSOR,
-    Platform.BUTTON,
-]
+from .coordinator import VictronEnergyDeviceUpdateCoordinator as Coordinator
+# For your initial PR, limit it to 1 platform.
+# There will be also: SWITCH, NUMBER, SELECT, BINARY_SENSOR, BUTTON
+_PLATFORMS: list[Platform] = [Platform.SENSOR]
 
 
 async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Set up victron from a config entry."""
 
     hass.data.setdefault(DOMAIN, {})
-    # TODO 1. Create API instance
-    # TODO 2. Validate the API connection (and authentication)
-    # 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],
         config_entry.data[SCAN_REGISTERS],
         config_entry.options[CONF_INTERVAL],
     )
-    # try:
-    #     await coordinator.async_config_entry_first_refresh()
-    # except ConfigEntryNotReady:
-    #     await coordinator.api.close()
-    #     raise
 
     # 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)
+    config_entry.runtime_data = coordinator_
+    await hass.config_entries.async_forward_entry_setups(config_entry, _PLATFORMS)
 
     return True
 
 
 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
+        config_entry, _PLATFORMS
     ):
         hass.data[DOMAIN].pop(config_entry.entry_id)
 

custom_components/victron/binary_sensor.py

@@ -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,
@@ -13,13 +14,14 @@
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HassJob, HomeAssistant
-from homeassistant.helpers import entity
+from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from .base import VictronBaseEntityDescription
 from .const import DOMAIN, BoolReadEntityType, register_info_dict
-from .coordinator import victronEnergyDeviceUpdateCoordinator
+from .coordinator import VictronEnergyDeviceUpdateCoordinator
+from .entity import VictronBaseEntityDescription
+from .select import VictronSelect
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -31,44 +33,46 @@ 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 = {}
-    for description in descriptions:
-        entity = description
-        entities.append(VictronBinarySensor(victron_coordinator, entity))
+    _entities = [
+        VictronSelect(hass, victron_coordinator, description)
+        for description in descriptions
+    ]
 
-    async_add_entities(entities, True)
+    async_add_entities(_entities, True)
 
 
-@dataclass
+@dataclass(frozen=True)
 class VictronEntityDescription(
-    BinarySensorEntityDescription, VictronBaseEntityDescription
+    BinarySensorEntityDescription,
+    VictronBaseEntityDescription,
 ):
     """Describes victron sensor entity."""
 
@@ -78,7 +82,7 @@ class VictronBinarySensor(CoordinatorEntity, BinarySensorEntity):
 
     def __init__(
         self,
-        coordinator: victronEnergyDeviceUpdateCoordinator,
+        coordinator: VictronEnergyDeviceUpdateCoordinator,
         description: VictronEntityDescription,
     ) -> None:
         """Initialize the binary sensor."""
@@ -101,25 +105,25 @@ 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]
 
     @property
-    def device_info(self) -> entity.DeviceInfo:
+    def device_info(self) -> 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],
+        return DeviceInfo(
+            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",
         )

custom_components/victron/button.py

@@ -4,6 +4,7 @@
 
 from dataclasses import dataclass
 import logging
+from typing import Any
 
 from homeassistant.components.button import (
     DOMAIN as BUTTON_DOMAIN,
@@ -13,13 +14,14 @@
 )
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HassJob, HomeAssistant
-from homeassistant.helpers import entity
+from homeassistant.helpers.device_registry import DeviceInfo
 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 .coordinator import VictronEnergyDeviceUpdateCoordinator
+from .entity import VictronWriteBaseEntityDescription
+from .select import VictronSelect
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -31,24 +33,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("_", " "),
@@ -59,18 +61,18 @@ async def async_setup_entry(
                     _LOGGER.debug("composed description == %s", description)
                     descriptions.append(description)
 
-    entities = []
-    entity = {}
-    for description in descriptions:
-        entity = description
-        entities.append(VictronBinarySensor(victron_coordinator, entity))
+    _entities = [
+        VictronSelect(hass, victron_coordinator, description)
+        for description in descriptions
+    ]
 
-    async_add_entities(entities, True)
+    async_add_entities(_entities, True)
 
 
-@dataclass
+@dataclass(frozen=True)
 class VictronEntityDescription(
-    ButtonEntityDescription, VictronWriteBaseEntityDescription
+    ButtonEntityDescription,
+    VictronWriteBaseEntityDescription,
 ):
     """Describes victron sensor entity."""
 
@@ -80,7 +82,7 @@ class VictronBinarySensor(CoordinatorEntity, ButtonEntity):
 
     def __init__(
         self,
-        coordinator: victronEnergyDeviceUpdateCoordinator,
+        coordinator: VictronEnergyDeviceUpdateCoordinator,
         description: VictronEntityDescription,
     ) -> None:
         """Initialize the sensor."""
@@ -106,17 +108,17 @@ 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]
 
     @property
-    def device_info(self) -> entity.DeviceInfo:
+    def device_info(self) -> 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],
+        return DeviceInfo(
+            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",
         )