Initial integration

README.md

@@ -0,0 +1,3 @@
+# NRGKick Integration
+
+A Home Assistant Integration to connect to your NRGKick device

custom_components/nrgkick/__init__.py

@@ -0,0 +1,39 @@
+"""The NRGKick integration."""
+from __future__ import annotations
+
+from .coordinator import NRGKickCoordinator
+from .websocket import NRGKickWebsocket
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+
+PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.NUMBER]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up NRGKick from a config entry."""
+
+    websocket = NRGKickWebsocket(entry.data["ip"], entry.data["uuid"])
+    await websocket.connect()
+
+    coordinator = NRGKickCoordinator(hass, websocket)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

custom_components/nrgkick/config_flow.py

@@ -0,0 +1,81 @@
+"""Config flow for NRGKick integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from .websocket import NRGKickWebsocket
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.exceptions import HomeAssistantError
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required("ip"): str,
+        vol.Required("uuid"): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+
+    nrgkicksocket = NRGKickWebsocket(data["ip"], data["uuid"])
+    details = None
+
+    try:
+        await nrgkicksocket.connect()
+        details = await nrgkicksocket.get_device_control_info()
+    except:
+        raise CannotConnect
+
+    # Return info that you want to store in the config entry.
+    return {"serial": details.serialNumber, "name": details.deviceName.value}
+
+
+class NRGKickConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for NRGKick."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+                user_input["serial"] = info["serial"]
+                user_input["name"] = info["name"]
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(title=info["name"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""

custom_components/nrgkick/const.py

@@ -0,0 +1,3 @@
+"""Constants for the NRGKick integration."""
+
+DOMAIN = "nrgkick"

custom_components/nrgkick/coordinator.py

@@ -0,0 +1,43 @@
+from datetime import timedelta
+import logging
+from typing import Any
+
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class NRGKickCoordinator(DataUpdateCoordinator):
+    """NRGKick coordinator."""
+
+    def __init__(self, hass: HomeAssistant, websocket) -> None:
+        """Initialize NRGKick."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=30),
+        )
+        self.websocket = websocket
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch data from API endpoint.
+
+        This is the place to pre-process the data to lookup tables
+        so entities can quickly look up their data.
+        """
+        try:
+            data = {}
+            data["cc_dv"] = await self.websocket.get_charge_control_dynamic_values()
+            data["cc_s"] = await self.websocket.get_charge_control_settings()
+            data["w_s"] = await self.websocket.get_wifi_status()
+            return data
+        except:
+            # Raising ConfigEntryAuthFailed will cancel future updates
+            # and start a config flow with SOURCE_REAUTH (async_step_reauth)
+            raise ConfigEntryAuthFailed

custom_components/nrgkick/entity.py

@@ -0,0 +1,40 @@
+from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
+from homeassistant.helpers.entity import EntityDescription
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import NRGKickCoordinator
+
+
+class NRGKickEntity(CoordinatorEntity[NRGKickCoordinator]):
+    """An entity using CoordinatorEntity.
+
+    The CoordinatorEntity class provides:
+      should_poll
+      async_update
+      async_added_to_hass
+      available
+
+    """
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: NRGKickCoordinator,
+        description: EntityDescription,
+    ) -> None:
+        """Pass coordinator to CoordinatorEntity."""
+        super().__init__(coordinator)
+        self.entity_description = description
+
+        serial = coordinator.config_entry.data["serial"]
+        dev_name = coordinator.config_entry.data["name"]
+        self._attr_unique_id = f"{serial}_{description.key}"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, serial)},
+            manufacturer="DiniTech",
+            model="NRGKick",
+            name=dev_name,
+            serial_number=serial,
+        )

custom_components/nrgkick/manifest.json

@@ -0,0 +1,15 @@
+{
+  "domain": "nrgkick",
+  "name": "NRGKick",
+  "codeowners": [
+    "@dupondje"
+  ],
+  "config_flow": true,
+  "dependencies": [],
+  "documentation": "https://www.home-assistant.io/integrations/nrgkick",
+  "homekit": {},
+  "iot_class": "local_polling",
+  "requirements": [],
+  "ssdp": [],
+  "zeroconf": []
+}

custom_components/nrgkick/number.py

@@ -0,0 +1,83 @@
+"""Number platform for NRGKick."""
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+import logging
+from typing import Any
+
+from homeassistant.components.number import (
+    NumberDeviceClass,
+    NumberEntity,
+    NumberEntityDescription,
+    NumberMode,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import UnitOfElectricCurrent
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+from .coordinator import NRGKickCoordinator
+from .entity import NRGKickEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class NRGKickRequiredKeysMixin:
+    """Mixin for required keys."""
+
+    value_fn: Callable[[Any], float | int | None]
+    api_fn: Callable[[Any, float | int], Coroutine[Any, Any, Any]]
+
+
+@dataclass
+class NRGKickNumberEntityDescription(NumberEntityDescription, NRGKickRequiredKeysMixin):
+    """Describes NRGKick number entity."""
+
+
+NUMBERS: list[NRGKickNumberEntityDescription] = [
+    NRGKickNumberEntityDescription(
+        key="charge_current_limit",
+        translation_key="charge_current_limit",
+        device_class=NumberDeviceClass.CURRENT,
+        native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
+        native_max_value=32.0,
+        native_min_value=6.0,
+        native_step=1.0,
+        mode=NumberMode.SLIDER,
+        value_fn=lambda data: data["cc_s"].chargeCurrent.userSet,
+        api_fn=lambda c, v: c.set_charge_current_limit(v),
+    ),
+]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the NRGKick number from config entry."""
+    coordinator: NRGKickCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    async_add_entities(NRGKickNumber(coordinator, number) for number in NUMBERS)
+
+
+class NRGKickNumber(NRGKickEntity, NumberEntity):
+    """Representation of NRGKick Number entity."""
+
+    entity_description: NRGKickNumberEntityDescription
+
+    @property
+    def native_value(self) -> float | None:
+        """Return the entity value to represent the entity state."""
+        return self.entity_description.value_fn(self.coordinator.data)
+
+    async def async_set_native_value(self, value: float) -> None:
+        """Update to the cable."""
+        _LOGGER.debug(
+            "Settings charge level to '%s'",
+            value,
+        )
+        await self.entity_description.api_fn(self.coordinator.websocket, value)
+        self.coordinator.async_update_listeners()