Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 03:03:17 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add NRGkick integration and tests (#159995)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Andreas Jakl
2026-01-27 21:33:12 +01:00
committed by GitHub
parent b84022f88b
commit 37b4bfc9fc
30 changed files with 6320 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@@ -376,6 +376,7 @@ homeassistant.components.no_ip.*
homeassistant.components.nordpool.*
homeassistant.components.notify.*
homeassistant.components.notion.*
homeassistant.components.nrgkick.*
homeassistant.components.ntfy.*
homeassistant.components.number.*
homeassistant.components.nut.*

2
CODEOWNERS generated
View File

@@ -1126,6 +1126,8 @@ build.json @home-assistant/supervisor
/tests/components/notify_events/ @matrozov @papajojo
/homeassistant/components/notion/ @bachya
/tests/components/notion/ @bachya
/homeassistant/components/nrgkick/ @andijakl
/tests/components/nrgkick/ @andijakl
/homeassistant/components/nsw_fuel_station/ @nickw444
/tests/components/nsw_fuel_station/ @nickw444
/homeassistant/components/nsw_rural_fire_service_feed/ @exxamalte

40
homeassistant/components/nrgkick/__init__.py Normal file
View File

@@ -0,0 +1,40 @@
"""The NRGkick integration."""
from __future__ import annotations
from nrgkick_api import NRGkickAPI
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import NRGkickConfigEntry, NRGkickDataUpdateCoordinator
PLATFORMS: list[Platform] = [
Platform.SENSOR,
]
async def async_setup_entry(hass: HomeAssistant, entry: NRGkickConfigEntry) -> bool:
"""Set up NRGkick from a config entry."""
api = NRGkickAPI(
host=entry.data[CONF_HOST],
username=entry.data.get(CONF_USERNAME),
password=entry.data.get(CONF_PASSWORD),
session=async_get_clientsession(hass),
)
coordinator = NRGkickDataUpdateCoordinator(hass, api, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
# Set up platforms.
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: NRGkickConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

39
homeassistant/components/nrgkick/api.py Normal file
View File

@@ -0,0 +1,39 @@
"""Home Assistant exceptions for the NRGkick integration."""
from __future__ import annotations
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
class NRGkickApiClientError(HomeAssistantError):
"""Base exception for NRGkick API client errors."""
translation_domain = DOMAIN
translation_key = "unknown_error"
class NRGkickApiClientCommunicationError(NRGkickApiClientError):
"""Exception for NRGkick API client communication errors."""
translation_domain = DOMAIN
translation_key = "communication_error"
class NRGkickApiClientAuthenticationError(NRGkickApiClientError):
"""Exception for NRGkick API client authentication errors."""
translation_domain = DOMAIN
translation_key = "authentication_error"
class NRGkickApiClientApiDisabledError(NRGkickApiClientError):
"""Exception for disabled NRGkick JSON API."""
translation_domain = DOMAIN
translation_key = "json_api_disabled"
class NRGkickApiClientInvalidResponseError(NRGkickApiClientError):
"""Exception for invalid responses from the device."""

340
homeassistant/components/nrgkick/config_flow.py Normal file
View File

@@ -0,0 +1,340 @@
"""Config flow for NRGkick integration."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import aiohttp
from nrgkick_api import (
NRGkickAPI,
NRGkickAPIDisabledError,
NRGkickAuthenticationError,
NRGkickConnectionError,
)
import voluptuous as vol
import yarl
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .api import (
NRGkickApiClientApiDisabledError,
NRGkickApiClientAuthenticationError,
NRGkickApiClientCommunicationError,
NRGkickApiClientError,
NRGkickApiClientInvalidResponseError,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
def _normalize_host(value: str) -> str:
"""Normalize user input to host[:port] (no scheme/path).
Accepts either a plain host/IP (optionally with a port) or a full URL.
If a URL is provided, we strip the scheme.
"""
value = value.strip()
if not value:
raise vol.Invalid("host is required")
if "://" in value:
try:
url = yarl.URL(cv.url(value))
except ValueError as err:
raise vol.Invalid("invalid url") from err
if not url.host:
raise vol.Invalid("invalid url")
if url.port is not None:
return f"{url.host}:{url.port}"
return url.host
return value.strip("/").split("/", 1)[0]
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): TextSelector(TextSelectorConfig(autocomplete="off")),
}
)
STEP_AUTH_DATA_SCHEMA = vol.Schema(
{
vol.Optional(CONF_USERNAME): TextSelector(
TextSelectorConfig(autocomplete="off")
),
vol.Optional(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
}
)
async def validate_input(
hass: HomeAssistant,
host: str,
username: str | None = None,
password: str | None = None,
) -> dict[str, Any]:
"""Validate the user input allows us to connect."""
session = async_get_clientsession(hass)
api = NRGkickAPI(
host=host,
username=username,
password=password,
session=session,
)
try:
await api.test_connection()
info = await api.get_info(["general"], raw=True)
except NRGkickAuthenticationError as err:
raise NRGkickApiClientAuthenticationError from err
except NRGkickAPIDisabledError as err:
raise NRGkickApiClientApiDisabledError from err
except (NRGkickConnectionError, TimeoutError, aiohttp.ClientError, OSError) as err:
raise NRGkickApiClientCommunicationError from err
device_name = info.get("general", {}).get("device_name")
if not device_name:
device_name = "NRGkick"
serial = info.get("general", {}).get("serial_number")
if not serial:
raise NRGkickApiClientInvalidResponseError
return {
"title": device_name,
"serial": serial,
}
class NRGkickConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for NRGkick."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovered_host: str | None = None
self._discovered_name: str | None = None
self._pending_host: str | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
try:
host = _normalize_host(user_input[CONF_HOST])
except vol.Invalid:
errors["base"] = "cannot_connect"
else:
try:
info = await validate_input(self.hass, host)
except NRGkickApiClientApiDisabledError:
errors["base"] = "json_api_disabled"
except NRGkickApiClientAuthenticationError:
self._pending_host = host
return await self.async_step_user_auth()
except NRGkickApiClientInvalidResponseError:
errors["base"] = "invalid_response"
except NRGkickApiClientCommunicationError:
errors["base"] = "cannot_connect"
except NRGkickApiClientError:
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(
info["serial"], raise_on_progress=False
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=info["title"], data={CONF_HOST: host}
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
async def async_step_user_auth(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the authentication step only when needed."""
errors: dict[str, str] = {}
if TYPE_CHECKING:
assert self._pending_host is not None
if user_input is not None:
username = user_input.get(CONF_USERNAME)
password = user_input.get(CONF_PASSWORD)
try:
info = await validate_input(
self.hass,
self._pending_host,
username=username,
password=password,
)
except NRGkickApiClientApiDisabledError:
errors["base"] = "json_api_disabled"
except NRGkickApiClientAuthenticationError:
errors["base"] = "invalid_auth"
except NRGkickApiClientInvalidResponseError:
errors["base"] = "invalid_response"
except NRGkickApiClientCommunicationError:
errors["base"] = "cannot_connect"
except NRGkickApiClientError:
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(info["serial"], raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=info["title"],
data={
CONF_HOST: self._pending_host,
CONF_USERNAME: username,
CONF_PASSWORD: password,
},
)
return self.async_show_form(
step_id="user_auth",
data_schema=STEP_AUTH_DATA_SCHEMA,
errors=errors,
description_placeholders={
"device_ip": self._pending_host,
},
)
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle zeroconf discovery."""
_LOGGER.debug("Discovered NRGkick device: %s", discovery_info)
# Extract device information from mDNS metadata.
serial = discovery_info.properties.get("serial_number")
device_name = discovery_info.properties.get("device_name")
model_type = discovery_info.properties.get("model_type")
json_api_enabled = discovery_info.properties.get("json_api_enabled", "0")
if not serial:
_LOGGER.debug("NRGkick device discovered without serial number")
return self.async_abort(reason="no_serial_number")
# Set unique ID to prevent duplicate entries.
await self.async_set_unique_id(serial)
# Update the host if the device is already configured (IP might have changed).
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
# Store discovery info for the confirmation step.
self._discovered_host = discovery_info.host
# Fallback: device_name -> model_type -> "NRGkick".
self._discovered_name = device_name or model_type or "NRGkick"
self.context["title_placeholders"] = {"name": self._discovered_name}
# If JSON API is disabled, guide the user through enabling it.
if json_api_enabled != "1":
_LOGGER.debug("NRGkick device %s does not have JSON API enabled", serial)
return await self.async_step_zeroconf_enable_json_api()
try:
await validate_input(self.hass, self._discovered_host)
except NRGkickApiClientAuthenticationError:
self._pending_host = self._discovered_host
return await self.async_step_user_auth()
except NRGkickApiClientApiDisabledError:
# mDNS metadata may be stale; fall back to the enable guidance.
return await self.async_step_zeroconf_enable_json_api()
except (
NRGkickApiClientCommunicationError,
NRGkickApiClientInvalidResponseError,
):
return self.async_abort(reason="cannot_connect")
except NRGkickApiClientError:
_LOGGER.exception("Unexpected error")
return self.async_abort(reason="unknown")
# Proceed to confirmation step (no auth required upfront).
return await self.async_step_zeroconf_confirm()
async def async_step_zeroconf_enable_json_api(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Guide the user to enable JSON API after discovery."""
errors: dict[str, str] = {}
if TYPE_CHECKING:
assert self._discovered_host is not None
assert self._discovered_name is not None
if user_input is not None:
try:
info = await validate_input(self.hass, self._discovered_host)
except NRGkickApiClientApiDisabledError:
errors["base"] = "json_api_disabled"
except NRGkickApiClientAuthenticationError:
self._pending_host = self._discovered_host
return await self.async_step_user_auth()
except NRGkickApiClientInvalidResponseError:
errors["base"] = "invalid_response"
except NRGkickApiClientCommunicationError:
errors["base"] = "cannot_connect"
except NRGkickApiClientError:
_LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=info["title"], data={CONF_HOST: self._discovered_host}
)
return self.async_show_form(
step_id="zeroconf_enable_json_api",
data_schema=vol.Schema({}),
description_placeholders={
"name": self._discovered_name,
"device_ip": _normalize_host(self._discovered_host),
},
errors=errors,
)
async def async_step_zeroconf_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm discovery."""
errors: dict[str, str] = {}
if TYPE_CHECKING:
assert self._discovered_host is not None
assert self._discovered_name is not None
if user_input is not None:
return self.async_create_entry(
title=self._discovered_name, data={CONF_HOST: self._discovered_host}
)
return self.async_show_form(
step_id="zeroconf_confirm",
data_schema=vol.Schema({}),
description_placeholders={
"name": self._discovered_name,
"device_ip": self._discovered_host,
},
errors=errors,
)

124
homeassistant/components/nrgkick/const.py Normal file
View File

@@ -0,0 +1,124 @@
"""Constants for the NRGkick integration."""
from typing import Final
from nrgkick_api import (
CellularMode,
ChargingStatus,
ConnectorType,
ErrorCode,
RcdTriggerStatus,
WarningCode,
)
DOMAIN: Final = "nrgkick"
# Default polling interval (seconds).
DEFAULT_SCAN_INTERVAL: Final = 30
# Note: API Endpoints are in the nrgkick-api library.
# Import from nrgkick_api if needed: from nrgkick_api import ENDPOINT_INFO, ...
# Human-readable status mapping for the status sensor.
# Values are translation keys that match translations/<lang>.json
STATUS_MAP: Final[dict[int, str | None]] = {
ChargingStatus.UNKNOWN: None,
ChargingStatus.STANDBY: "standby",
ChargingStatus.CONNECTED: "connected",
ChargingStatus.CHARGING: "charging",
ChargingStatus.ERROR: "error",
ChargingStatus.WAKEUP: "wakeup",
}
# Human-readable RCD trigger mapping.
# Values are translation keys that match translations/<lang>.json
RCD_TRIGGER_MAP: Final[dict[int, str]] = {
RcdTriggerStatus.NO_FAULT: "no_fault",
RcdTriggerStatus.AC_30MA_FAULT: "ac_30ma_fault",
RcdTriggerStatus.AC_60MA_FAULT: "ac_60ma_fault",
RcdTriggerStatus.AC_150MA_FAULT: "ac_150ma_fault",
RcdTriggerStatus.DC_POSITIVE_6MA_FAULT: "dc_positive_6ma_fault",
RcdTriggerStatus.DC_NEGATIVE_6MA_FAULT: "dc_negative_6ma_fault",
}
# Human-readable warning code mapping.
# Values are translation keys that match translations/<lang>.json
WARNING_CODE_MAP: Final[dict[int, str]] = {
WarningCode.NO_WARNING: "no_warning",
WarningCode.NO_PE: "no_pe",
WarningCode.BLACKOUT_PROTECTION: "blackout_protection",
WarningCode.ENERGY_LIMIT_REACHED: "energy_limit_reached",
WarningCode.EV_DOES_NOT_COMPLY_STANDARD: "ev_does_not_comply_standard",
WarningCode.UNSUPPORTED_CHARGING_MODE: "unsupported_charging_mode",
WarningCode.NO_ATTACHMENT_DETECTED: "no_attachment_detected",
WarningCode.NO_COMM_WITH_TYPE2_ATTACHMENT: "no_comm_with_type2_attachment",
WarningCode.INCREASED_TEMPERATURE: "increased_temperature",
WarningCode.INCREASED_HOUSING_TEMPERATURE: "increased_housing_temperature",
WarningCode.INCREASED_ATTACHMENT_TEMPERATURE: "increased_attachment_temperature",
WarningCode.INCREASED_DOMESTIC_PLUG_TEMPERATURE: (
"increased_domestic_plug_temperature"
),
}
# Human-readable error code mapping.
# Values are translation keys that match translations/<lang>.json
ERROR_CODE_MAP: Final[dict[int, str]] = {
ErrorCode.NO_ERROR: "no_error",
ErrorCode.GENERAL_ERROR: "general_error",
ErrorCode.ATTACHMENT_32A_ON_16A_UNIT: "32a_attachment_on_16a_unit",
ErrorCode.VOLTAGE_DROP_DETECTED: "voltage_drop_detected",
ErrorCode.UNPLUG_DETECTION_TRIGGERED: "unplug_detection_triggered",
ErrorCode.TYPE2_NOT_AUTHORIZED: "type2_not_authorized",
ErrorCode.RESIDUAL_CURRENT_DETECTED: "residual_current_detected",
ErrorCode.CP_SIGNAL_VOLTAGE_ERROR: "cp_signal_voltage_error",
ErrorCode.CP_SIGNAL_IMPERMISSIBLE: "cp_signal_impermissible",
ErrorCode.EV_DIODE_FAULT: "ev_diode_fault",
ErrorCode.PE_SELF_TEST_FAILED: "pe_self_test_failed",
ErrorCode.RCD_SELF_TEST_FAILED: "rcd_self_test_failed",
ErrorCode.RELAY_SELF_TEST_FAILED: "relay_self_test_failed",
ErrorCode.PE_AND_RCD_SELF_TEST_FAILED: "pe_and_rcd_self_test_failed",
ErrorCode.PE_AND_RELAY_SELF_TEST_FAILED: "pe_and_relay_self_test_failed",
ErrorCode.RCD_AND_RELAY_SELF_TEST_FAILED: "rcd_and_relay_self_test_failed",
ErrorCode.PE_AND_RCD_AND_RELAY_SELF_TEST_FAILED: (
"pe_and_rcd_and_relay_self_test_failed"
),
ErrorCode.SUPPLY_VOLTAGE_ERROR: "supply_voltage_error",
ErrorCode.PHASE_SHIFT_ERROR: "phase_shift_error",
ErrorCode.OVERVOLTAGE_DETECTED: "overvoltage_detected",
ErrorCode.UNDERVOLTAGE_DETECTED: "undervoltage_detected",
ErrorCode.OVERVOLTAGE_WITHOUT_PE_DETECTED: "overvoltage_without_pe_detected",
ErrorCode.UNDERVOLTAGE_WITHOUT_PE_DETECTED: "undervoltage_without_pe_detected",
ErrorCode.UNDERFREQUENCY_DETECTED: "underfrequency_detected",
ErrorCode.OVERFREQUENCY_DETECTED: "overfrequency_detected",
ErrorCode.UNKNOWN_FREQUENCY_TYPE: "unknown_frequency_type",
ErrorCode.UNKNOWN_GRID_TYPE: "unknown_grid_type",
ErrorCode.GENERAL_OVERTEMPERATURE: "general_overtemperature",
ErrorCode.HOUSING_OVERTEMPERATURE: "housing_overtemperature",
ErrorCode.ATTACHMENT_OVERTEMPERATURE: "attachment_overtemperature",
ErrorCode.DOMESTIC_PLUG_OVERTEMPERATURE: "domestic_plug_overtemperature",
}
# Human-readable connector type mapping.
# Values are translation keys that match translations/<lang>.json
CONNECTOR_TYPE_MAP: Final[dict[int, str | None]] = {
ConnectorType.UNKNOWN: None,
ConnectorType.CEE: "cee",
ConnectorType.DOMESTIC: "domestic",
ConnectorType.TYPE2: "type2",
ConnectorType.WALL: "wall",
ConnectorType.AUS: "aus",
}
# Human-readable cellular mode mapping.
# Values are translation keys that match translations/<lang>.json
CELLULAR_MODE_MAP: Final[dict[int, str | None]] = {
CellularMode.UNKNOWN: None,
CellularMode.NO_SERVICE: "no_service",
CellularMode.GSM: "gsm",
CellularMode.LTE_CAT_M1: "lte_cat_m1",
CellularMode.LTE_NB_IOT: "lte_nb_iot",
}

89
homeassistant/components/nrgkick/coordinator.py Normal file
View File

@@ -0,0 +1,89 @@
"""DataUpdateCoordinator for NRGkick integration."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
import aiohttp
from nrgkick_api import (
NRGkickAPI,
NRGkickAPIDisabledError,
NRGkickAuthenticationError,
NRGkickConnectionError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
_LOGGER = logging.getLogger(__name__)
# Type alias for typed config entry with runtime_data.
type NRGkickConfigEntry = ConfigEntry[NRGkickDataUpdateCoordinator]
@dataclass(slots=True)
class NRGkickData:
"""Container for coordinator data."""
info: dict[str, Any]
control: dict[str, Any]
values: dict[str, Any]
class NRGkickDataUpdateCoordinator(DataUpdateCoordinator[NRGkickData]):
"""Class to manage fetching NRGkick data from the API."""
config_entry: NRGkickConfigEntry
def __init__(
self, hass: HomeAssistant, api: NRGkickAPI, entry: NRGkickConfigEntry
) -> None:
"""Initialize."""
self.api = api
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
config_entry=entry,
always_update=False,
)
async def _async_update_data(self) -> NRGkickData:
"""Update data via library."""
try:
info = await self.api.get_info(raw=True)
control = await self.api.get_control()
values = await self.api.get_values(raw=True)
except NRGkickAuthenticationError as error:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from error
except NRGkickAPIDisabledError as error:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="json_api_disabled",
) from error
except NRGkickConnectionError as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={"error": str(error)},
) from error
except (TimeoutError, aiohttp.ClientError, OSError) as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={"error": str(error)},
) from error
return NRGkickData(info=info, control=control, values=values)

57
homeassistant/components/nrgkick/entity.py Normal file
View File

@@ -0,0 +1,57 @@
"""Base entity for NRGkick integration."""
from __future__ import annotations
from typing import Any
from homeassistant.const import CONF_HOST
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import NRGkickDataUpdateCoordinator
class NRGkickEntity(CoordinatorEntity[NRGkickDataUpdateCoordinator]):
"""Base class for NRGkick entities with common device info setup."""
_attr_has_entity_name = True
def __init__(self, coordinator: NRGkickDataUpdateCoordinator, key: str) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._key = key
data = self.coordinator.data
assert data is not None
info_data: dict[str, Any] = data.info
device_info: dict[str, Any] = info_data.get("general", {})
network_info: dict[str, Any] = info_data.get("network", {})
# The config flow requires a serial number and sets it as unique_id.
serial = self.coordinator.config_entry.unique_id
assert serial is not None
# Get additional device info fields.
versions: dict[str, Any] = info_data.get("versions", {})
connections: set[tuple[str, str]] | None = None
if (mac_address := network_info.get("mac_address")) and isinstance(
mac_address, str
):
connections = {(CONNECTION_NETWORK_MAC, mac_address)}
self._attr_unique_id = f"{serial}_{self._key}"
device_info_typed = DeviceInfo(
configuration_url=f"http://{self.coordinator.config_entry.data[CONF_HOST]}",
identifiers={(DOMAIN, serial)},
serial_number=serial,
manufacturer="DiniTech",
model=device_info.get("model_type", "NRGkick Gen2"),
sw_version=versions.get("sw_sm"),
hw_version=versions.get("hw_sm"),
)
if connections is not None:
device_info_typed["connections"] = connections
self._attr_device_info = device_info_typed

87
homeassistant/components/nrgkick/icons.json Normal file
View File

@@ -0,0 +1,87 @@
{
"entity": {
"sensor": {
"charge_count": {
"default": "mdi:counter"
},
"charged_energy": {
"default": "mdi:lightning-bolt"
},
"charging_current": {
"default": "mdi:current-ac"
},
"charging_rate": {
"default": "mdi:speedometer"
},
"charging_voltage": {
"default": "mdi:flash"
},
"connector_max_current": {
"default": "mdi:current-ac"
},
"connector_phase_count": {
"default": "mdi:sine-wave"
},
"connector_serial": {
"default": "mdi:barcode"
},
"connector_type": {
"default": "mdi:ev-plug-type2"
},
"error_code": {
"default": "mdi:alert-circle"
},
"grid_frequency": {
"default": "mdi:sine-wave"
},
"grid_voltage": {
"default": "mdi:flash"
},
"network_rssi": {
"default": "mdi:wifi-strength-3"
},
"network_ssid": {
"default": "mdi:wifi"
},
"peak_power": {
"default": "mdi:flash"
},
"powerflow_grid_frequency": {
"default": "mdi:sine-wave"
},
"rated_current": {
"default": "mdi:current-ac"
},
"rcd_trigger": {
"default": "mdi:flash-alert"
},
"status": {
"default": "mdi:ev-station"
},
"total_active_power": {
"default": "mdi:flash"
},
"total_apparent_power": {
"default": "mdi:flash-outline"
},
"total_charged_energy": {
"default": "mdi:lightning-bolt"
},
"total_power_factor": {
"default": "mdi:percent"
},
"total_reactive_power": {
"default": "mdi:flash-outline"
},
"vehicle_charging_time": {
"default": "mdi:timer"
},
"vehicle_connected_since": {
"default": "mdi:timer-outline"
},
"warning_code": {
"default": "mdi:alert"
}
}
}
}

12
homeassistant/components/nrgkick/manifest.json Normal file
View File

@@ -0,0 +1,12 @@
{
"domain": "nrgkick",
"name": "NRGkick",
"codeowners": ["@andijakl"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nrgkick",
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["nrgkick-api==1.6.0"],
"zeroconf": ["_nrgkick._tcp.local."]
}

80
homeassistant/components/nrgkick/quality_scale.yaml Normal file
View File

@@ -0,0 +1,80 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not provide additional actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
Entities of this integration use DataUpdateCoordinator and do not explicitly subscribe to events.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
This integration does not provide service actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: |
This integration does not provide configuration options (no options flow).
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery: done
discovery-update-info: done
docs-data-update: done
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: |
This integration represents a single device per config entry.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices:
status: exempt
comment: |
This integration represents a single device per config entry.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done

800
homeassistant/components/nrgkick/sensor.py Normal file
View File

@@ -0,0 +1,800 @@
"""Sensor platform for NRGkick."""
from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, cast
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
UnitOfApparentPower,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfFrequency,
UnitOfPower,
UnitOfReactivePower,
UnitOfSpeed,
UnitOfTemperature,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow
from homeassistant.util.variance import ignore_variance
from .const import (
CELLULAR_MODE_MAP,
CONNECTOR_TYPE_MAP,
ERROR_CODE_MAP,
RCD_TRIGGER_MAP,
STATUS_MAP,
WARNING_CODE_MAP,
)
from .coordinator import NRGkickConfigEntry, NRGkickData, NRGkickDataUpdateCoordinator
from .entity import NRGkickEntity
PARALLEL_UPDATES = 0
def _get_nested_dict_value(data: Any, *keys: str) -> Any:
"""Safely get a nested value from dict-like API responses."""
current: Any = data
for key in keys:
try:
current = current.get(key)
except AttributeError:
return None
return current
@dataclass(frozen=True, kw_only=True)
class NRGkickSensorEntityDescription(SensorEntityDescription):
"""Class describing NRGkick sensor entities."""
value_fn: Callable[[NRGkickData], StateType | datetime | None]
requires_sim_module: bool = False
def _seconds_to_datetime(value: int) -> datetime:
"""Convert seconds to a UTC timestamp."""
return utcnow().replace(microsecond=0) - timedelta(seconds=value)
_seconds_to_stable_datetime = ignore_variance(
_seconds_to_datetime, timedelta(minutes=1)
)
def _seconds_to_stable_timestamp(value: StateType) -> datetime | None:
"""Convert seconds to a stable timestamp.
This is used for durations that represent "seconds since X" coming from the
device. Converting to a timestamp avoids UI drift due to polling cadence.
"""
if value is None:
return None
try:
return _seconds_to_stable_datetime(cast(int, value))
except (TypeError, OverflowError):
return None
def _map_code_to_translation_key(
value: StateType,
mapping: Mapping[int, str | None],
) -> StateType:
"""Map numeric API codes to translation keys.
The NRGkick API returns `int` (including `IntEnum`) values for code-like
fields used as sensor states.
"""
if value is None:
return None
# The API returns ints (including IntEnum). Use a cast to satisfy typing
# without paying for repeated runtime type checks in this hot path.
return mapping.get(cast(int, value))
def _enum_options_from_mapping(mapping: Mapping[int, str | None]) -> list[str]:
"""Build stable enum options from a numeric->translation-key mapping."""
# Keep ordering stable by sorting keys.
unique_options: dict[str, None] = {}
for key in sorted(mapping):
if (option := mapping[key]) is not None:
unique_options[option] = None
return list(unique_options)
async def async_setup_entry(
_hass: HomeAssistant,
entry: NRGkickConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up NRGkick sensors based on a config entry."""
coordinator: NRGkickDataUpdateCoordinator = entry.runtime_data
data = coordinator.data
assert data is not None
info_data: dict[str, Any] = data.info
general_info: dict[str, Any] = info_data.get("general", {})
model_type = general_info.get("model_type")
# The cellular and GPS modules are optional. There is no dedicated API to query
# module availability, but SIM-capable models include "SIM" in their model
# type (e.g. "NRGkick Gen2 SIM").
# Note: GPS to be added back with future pull request, currently only cellular.
has_sim_module = isinstance(model_type, str) and "SIM" in model_type.upper()
async_add_entities(
NRGkickSensor(coordinator, description)
for description in SENSORS
if has_sim_module or not description.requires_sim_module
)
SENSORS: tuple[NRGkickSensorEntityDescription, ...] = (
# INFO - General
NRGkickSensorEntityDescription(
key="rated_current",
translation_key="rated_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.info, "general", "rated_current"
),
),
# INFO - Connector
NRGkickSensorEntityDescription(
key="connector_phase_count",
translation_key="connector_phase_count",
value_fn=lambda data: _get_nested_dict_value(
data.info, "connector", "phase_count"
),
),
NRGkickSensorEntityDescription(
key="connector_max_current",
translation_key="connector_max_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.info, "connector", "max_current"
),
),
NRGkickSensorEntityDescription(
key="connector_type",
translation_key="connector_type",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(CONNECTOR_TYPE_MAP),
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _map_code_to_translation_key(
cast(StateType, _get_nested_dict_value(data.info, "connector", "type")),
CONNECTOR_TYPE_MAP,
),
),
NRGkickSensorEntityDescription(
key="connector_serial",
translation_key="connector_serial",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(data.info, "connector", "serial"),
),
# INFO - Grid
NRGkickSensorEntityDescription(
key="grid_voltage",
translation_key="grid_voltage",
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(data.info, "grid", "voltage"),
),
NRGkickSensorEntityDescription(
key="grid_frequency",
translation_key="grid_frequency",
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(data.info, "grid", "frequency"),
),
# INFO - Network
NRGkickSensorEntityDescription(
key="network_ssid",
translation_key="network_ssid",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(data.info, "network", "ssid"),
),
NRGkickSensorEntityDescription(
key="network_rssi",
translation_key="network_rssi",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(data.info, "network", "rssi"),
),
# INFO - Cellular (optional, only if cellular module is available)
NRGkickSensorEntityDescription(
key="cellular_mode",
translation_key="cellular_mode",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(CELLULAR_MODE_MAP),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
requires_sim_module=True,
value_fn=lambda data: _map_code_to_translation_key(
cast(StateType, _get_nested_dict_value(data.info, "cellular", "mode")),
CELLULAR_MODE_MAP,
),
),
NRGkickSensorEntityDescription(
key="cellular_rssi",
translation_key="cellular_rssi",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
requires_sim_module=True,
value_fn=lambda data: _get_nested_dict_value(data.info, "cellular", "rssi"),
),
NRGkickSensorEntityDescription(
key="cellular_operator",
translation_key="cellular_operator",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
requires_sim_module=True,
value_fn=lambda data: _get_nested_dict_value(data.info, "cellular", "operator"),
),
# VALUES - Energy
NRGkickSensorEntityDescription(
key="total_charged_energy",
translation_key="total_charged_energy",
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_display_precision=3,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
value_fn=lambda data: _get_nested_dict_value(
data.values, "energy", "total_charged_energy"
),
),
NRGkickSensorEntityDescription(
key="charged_energy",
translation_key="charged_energy",
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
suggested_display_precision=3,
suggested_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
value_fn=lambda data: _get_nested_dict_value(
data.values, "energy", "charged_energy"
),
),
# VALUES - Powerflow (Total)
NRGkickSensorEntityDescription(
key="charging_voltage",
translation_key="charging_voltage",
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "charging_voltage"
),
),
NRGkickSensorEntityDescription(
key="charging_current",
translation_key="charging_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "charging_current"
),
),
NRGkickSensorEntityDescription(
key="powerflow_grid_frequency",
translation_key="powerflow_grid_frequency",
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "grid_frequency"
),
),
NRGkickSensorEntityDescription(
key="peak_power",
translation_key="peak_power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "peak_power"
),
),
NRGkickSensorEntityDescription(
key="total_active_power",
translation_key="total_active_power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "total_active_power"
),
),
NRGkickSensorEntityDescription(
key="total_reactive_power",
translation_key="total_reactive_power",
device_class=SensorDeviceClass.REACTIVE_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfReactivePower.VOLT_AMPERE_REACTIVE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "total_reactive_power"
),
),
NRGkickSensorEntityDescription(
key="total_apparent_power",
translation_key="total_apparent_power",
device_class=SensorDeviceClass.APPARENT_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfApparentPower.VOLT_AMPERE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "total_apparent_power"
),
),
NRGkickSensorEntityDescription(
key="total_power_factor",
translation_key="total_power_factor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "total_power_factor"
),
),
# VALUES - Powerflow L1
NRGkickSensorEntityDescription(
key="l1_voltage",
translation_key="l1_voltage",
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "voltage"
),
),
NRGkickSensorEntityDescription(
key="l1_current",
translation_key="l1_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "current"
),
),
NRGkickSensorEntityDescription(
key="l1_active_power",
translation_key="l1_active_power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "active_power"
),
),
NRGkickSensorEntityDescription(
key="l1_reactive_power",
translation_key="l1_reactive_power",
device_class=SensorDeviceClass.REACTIVE_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfReactivePower.VOLT_AMPERE_REACTIVE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "reactive_power"
),
),
NRGkickSensorEntityDescription(
key="l1_apparent_power",
translation_key="l1_apparent_power",
device_class=SensorDeviceClass.APPARENT_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfApparentPower.VOLT_AMPERE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "apparent_power"
),
),
NRGkickSensorEntityDescription(
key="l1_power_factor",
translation_key="l1_power_factor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l1", "power_factor"
),
),
# VALUES - Powerflow L2
NRGkickSensorEntityDescription(
key="l2_voltage",
translation_key="l2_voltage",
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "voltage"
),
),
NRGkickSensorEntityDescription(
key="l2_current",
translation_key="l2_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "current"
),
),
NRGkickSensorEntityDescription(
key="l2_active_power",
translation_key="l2_active_power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "active_power"
),
),
NRGkickSensorEntityDescription(
key="l2_reactive_power",
translation_key="l2_reactive_power",
device_class=SensorDeviceClass.REACTIVE_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfReactivePower.VOLT_AMPERE_REACTIVE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "reactive_power"
),
),
NRGkickSensorEntityDescription(
key="l2_apparent_power",
translation_key="l2_apparent_power",
device_class=SensorDeviceClass.APPARENT_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfApparentPower.VOLT_AMPERE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "apparent_power"
),
),
NRGkickSensorEntityDescription(
key="l2_power_factor",
translation_key="l2_power_factor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l2", "power_factor"
),
),
# VALUES - Powerflow L3
NRGkickSensorEntityDescription(
key="l3_voltage",
translation_key="l3_voltage",
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "voltage"
),
),
NRGkickSensorEntityDescription(
key="l3_current",
translation_key="l3_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "current"
),
),
NRGkickSensorEntityDescription(
key="l3_active_power",
translation_key="l3_active_power",
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
suggested_display_precision=2,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "active_power"
),
),
NRGkickSensorEntityDescription(
key="l3_reactive_power",
translation_key="l3_reactive_power",
device_class=SensorDeviceClass.REACTIVE_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfReactivePower.VOLT_AMPERE_REACTIVE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "reactive_power"
),
),
NRGkickSensorEntityDescription(
key="l3_apparent_power",
translation_key="l3_apparent_power",
device_class=SensorDeviceClass.APPARENT_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfApparentPower.VOLT_AMPERE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "apparent_power"
),
),
NRGkickSensorEntityDescription(
key="l3_power_factor",
translation_key="l3_power_factor",
device_class=SensorDeviceClass.POWER_FACTOR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "l3", "power_factor"
),
),
# VALUES - Powerflow Neutral
NRGkickSensorEntityDescription(
key="n_current",
translation_key="n_current",
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
suggested_display_precision=2,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: _get_nested_dict_value(
data.values, "powerflow", "n", "current"
),
),
# VALUES - General
NRGkickSensorEntityDescription(
key="charging_rate",
translation_key="charging_rate",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR,
value_fn=lambda data: _get_nested_dict_value(
data.values, "general", "charging_rate"
),
),
NRGkickSensorEntityDescription(
key="vehicle_connected_since",
translation_key="vehicle_connected_since",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: _seconds_to_stable_timestamp(
cast(
StateType,
_get_nested_dict_value(data.values, "general", "vehicle_connect_time"),
)
),
),
NRGkickSensorEntityDescription(
key="vehicle_charging_time",
translation_key="vehicle_charging_time",
device_class=SensorDeviceClass.DURATION,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.SECONDS,
suggested_unit_of_measurement=UnitOfTime.MINUTES,
value_fn=lambda data: _get_nested_dict_value(
data.values, "general", "vehicle_charging_time"
),
),
NRGkickSensorEntityDescription(
key="status",
translation_key="status",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(STATUS_MAP),
value_fn=lambda data: _map_code_to_translation_key(
cast(StateType, _get_nested_dict_value(data.values, "general", "status")),
STATUS_MAP,
),
),
NRGkickSensorEntityDescription(
key="charge_count",
translation_key="charge_count",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.TOTAL_INCREASING,
suggested_display_precision=0,
value_fn=lambda data: _get_nested_dict_value(
data.values, "general", "charge_count"
),
),
NRGkickSensorEntityDescription(
key="rcd_trigger",
translation_key="rcd_trigger",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(RCD_TRIGGER_MAP),
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _map_code_to_translation_key(
cast(
StateType, _get_nested_dict_value(data.values, "general", "rcd_trigger")
),
RCD_TRIGGER_MAP,
),
),
NRGkickSensorEntityDescription(
key="warning_code",
translation_key="warning_code",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(WARNING_CODE_MAP),
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _map_code_to_translation_key(
cast(
StateType,
_get_nested_dict_value(data.values, "general", "warning_code"),
),
WARNING_CODE_MAP,
),
),
NRGkickSensorEntityDescription(
key="error_code",
translation_key="error_code",
device_class=SensorDeviceClass.ENUM,
options=_enum_options_from_mapping(ERROR_CODE_MAP),
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _map_code_to_translation_key(
cast(
StateType, _get_nested_dict_value(data.values, "general", "error_code")
),
ERROR_CODE_MAP,
),
),
# VALUES - Temperatures
NRGkickSensorEntityDescription(
key="housing_temperature",
translation_key="housing_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "housing"
),
),
NRGkickSensorEntityDescription(
key="connector_l1_temperature",
translation_key="connector_l1_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "connector_l1"
),
),
NRGkickSensorEntityDescription(
key="connector_l2_temperature",
translation_key="connector_l2_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "connector_l2"
),
),
NRGkickSensorEntityDescription(
key="connector_l3_temperature",
translation_key="connector_l3_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "connector_l3"
),
),
NRGkickSensorEntityDescription(
key="domestic_plug_1_temperature",
translation_key="domestic_plug_1_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "domestic_plug_1"
),
),
NRGkickSensorEntityDescription(
key="domestic_plug_2_temperature",
translation_key="domestic_plug_2_temperature",
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: _get_nested_dict_value(
data.values, "temperatures", "domestic_plug_2"
),
),
)
class NRGkickSensor(NRGkickEntity, SensorEntity):
"""Representation of a NRGkick sensor."""
entity_description: NRGkickSensorEntityDescription
def __init__(
self,
coordinator: NRGkickDataUpdateCoordinator,
entity_description: NRGkickSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, entity_description.key)
self.entity_description = entity_description
@property
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)

323
homeassistant/components/nrgkick/strings.json Normal file
View File

@@ -0,0 +1,323 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"json_api_disabled": "JSON API is disabled on the device. Enable it in the NRGkick mobile app under Extended \u2192 Local API \u2192 API Variants.",
"no_serial_number": "Device does not provide a serial number"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_response": "The device returned an invalid response. Verify the host or IP address and make sure JSON API is enabled in the NRGkick mobile app.",
"json_api_disabled": "JSON API is disabled on the device.\n\nEnable JSON API in the NRGkick mobile app:\n1. Open the NRGkick mobile app \u2192 Extended \u2192 Local API\n2. Enable JSON API under API Variants\n3. (Optional \u0026 recommended) Enable Authentication (JSON) and set credentials\n\nAfter enabling JSON API, submit this form again.",
"no_serial_number": "Device does not provide a serial number",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]"
},
"data_description": {
"host": "The hostname or IP address of the NRGkick device."
},
"description": "Set up your NRGkick device. Ensure the device is powered on and reachable on the network."
},
"user_auth": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "Password for your NRGkick device.",
"username": "Username for your NRGkick device."
},
"description": "Authentication is required for {device_ip}.\n\nGet your username and password in the NRGkick mobile app:\n1. Open the NRGkick mobile app \u2192 Extended \u2192 Local API\n2. Under Authentication (JSON), check or set your username and password\n\nIf you changed the credentials in the app, use the updated values here."
},
"zeroconf_confirm": {
"description": "Do you want to add the NRGkick device ({name}) at {device_ip} to Home Assistant?"
},
"zeroconf_enable_json_api": {
"description": "The NRGkick device ({name}) at {device_ip} was discovered, but JSON API is disabled.\n\nEnable JSON API in the NRGkick mobile app:\n1. Open the NRGkick mobile app \u2192 Extended \u2192 Local API\n2. Enable JSON API under API Variants\n3. (Optional \u0026 recommended) Enable Authentication (JSON) and set credentials\n\nAfter enabling JSON API, submit this form to continue."
}
}
},
"entity": {
"sensor": {
"cellular_mode": {
"name": "Cellular mode",
"state": {
"gsm": "GSM",
"lte_cat_m1": "LTE Cat-M1",
"lte_nb_iot": "LTE NB-IoT",
"no_service": "No service"
}
},
"cellular_operator": {
"name": "Cellular operator"
},
"cellular_rssi": {
"name": "Cellular signal strength"
},
"charge_count": {
"name": "Charge count",
"unit_of_measurement": "cycles"
},
"charged_energy": {
"name": "Charged energy"
},
"charging_current": {
"name": "Charging current"
},
"charging_rate": {
"name": "Charging rate"
},
"charging_voltage": {
"name": "Charging voltage"
},
"connector_l1_temperature": {
"name": "Connector L1 temperature"
},
"connector_l2_temperature": {
"name": "Connector L2 temperature"
},
"connector_l3_temperature": {
"name": "Connector L3 temperature"
},
"connector_max_current": {
"name": "Connector max current"
},
"connector_phase_count": {
"name": "Connector phase count"
},
"connector_serial": {
"name": "Connector serial"
},
"connector_type": {
"name": "Connector type",
"state": {
"aus": "Australian (32A 5-pin)",
"cee": "CEE",
"domestic": "Domestic plug",
"type2": "Type 2",
"wall": "Wall socket"
}
},
"domestic_plug_1_temperature": {
"name": "Domestic plug 1 temperature"
},
"domestic_plug_2_temperature": {
"name": "Domestic plug 2 temperature"
},
"error_code": {
"name": "Error",
"state": {
"32a_attachment_on_16a_unit": "32A attachment on 16A unit",
"attachment_overtemperature": "Attachment overtemperature",
"cp_signal_impermissible": "CP signal impermissible",
"cp_signal_voltage_error": "CP signal voltage error",
"domestic_plug_overtemperature": "Domestic plug overtemperature",
"ev_diode_fault": "EV diode fault",
"general_error": "General error",
"general_overtemperature": "General overtemperature",
"housing_overtemperature": "Housing overtemperature",
"no_error": "No error",
"overfrequency_detected": "Overfrequency detected",
"overvoltage_detected": "Overvoltage detected",
"overvoltage_without_pe_detected": "Overvoltage without PE detected",
"pe_and_rcd_and_relay_self_test_failed": "PE, RCD, and relay self-test failed",
"pe_and_rcd_self_test_failed": "PE and RCD self-test failed",
"pe_and_relay_self_test_failed": "PE and relay self-test failed",
"pe_self_test_failed": "PE self-test failed",
"phase_shift_error": "Phase shift error",
"rcd_and_relay_self_test_failed": "RCD and relay self-test failed",
"rcd_self_test_failed": "RCD self-test failed",
"relay_self_test_failed": "Relay self-test failed",
"residual_current_detected": "Residual current detected",
"supply_voltage_error": "Supply voltage error",
"type2_not_authorized": "Type 2 not authorized",
"underfrequency_detected": "Underfrequency detected",
"undervoltage_detected": "Undervoltage detected",
"undervoltage_without_pe_detected": "Undervoltage without PE detected",
"unknown_frequency_type": "Unknown frequency type",
"unknown_grid_type": "Unknown grid type",
"unplug_detection_triggered": "Unplug detection triggered",
"voltage_drop_detected": "Voltage drop detected"
}
},
"grid_frequency": {
"name": "Grid frequency"
},
"grid_voltage": {
"name": "Grid voltage"
},
"housing_temperature": {
"name": "Housing temperature"
},
"l1_active_power": {
"name": "L1 active power"
},
"l1_apparent_power": {
"name": "L1 apparent power"
},
"l1_current": {
"name": "L1 current"
},
"l1_power_factor": {
"name": "L1 power factor"
},
"l1_reactive_power": {
"name": "L1 reactive power"
},
"l1_voltage": {
"name": "L1 voltage"
},
"l2_active_power": {
"name": "L2 active power"
},
"l2_apparent_power": {
"name": "L2 apparent power"
},
"l2_current": {
"name": "L2 current"
},
"l2_power_factor": {
"name": "L2 power factor"
},
"l2_reactive_power": {
"name": "L2 reactive power"
},
"l2_voltage": {
"name": "L2 voltage"
},
"l3_active_power": {
"name": "L3 active power"
},
"l3_apparent_power": {
"name": "L3 apparent power"
},
"l3_current": {
"name": "L3 current"
},
"l3_power_factor": {
"name": "L3 power factor"
},
"l3_reactive_power": {
"name": "L3 reactive power"
},
"l3_voltage": {
"name": "L3 voltage"
},
"n_current": {
"name": "Neutral current"
},
"network_rssi": {
"name": "Signal strength"
},
"network_ssid": {
"name": "SSID"
},
"peak_power": {
"name": "Peak power"
},
"powerflow_grid_frequency": {
"name": "Powerflow grid frequency"
},
"rated_current": {
"name": "Rated current"
},
"rcd_trigger": {
"name": "RCD trigger",
"state": {
"ac_150ma_fault": "AC 150mA fault",
"ac_30ma_fault": "AC 30mA fault",
"ac_60ma_fault": "AC 60mA fault",
"dc_negative_6ma_fault": "DC -6mA fault",
"dc_positive_6ma_fault": "DC +6mA fault",
"no_fault": "No fault"
}
},
"status": {
"name": "Status",
"state": {
"charging": "[%key:common::state::charging%]",
"connected": "[%key:common::state::connected%]",
"error": "[%key:common::state::error%]",
"standby": "[%key:common::state::standby%]",
"wakeup": "Wakeup"
}
},
"total_active_power": {
"name": "Total active power"
},
"total_apparent_power": {
"name": "Total apparent power"
},
"total_charged_energy": {
"name": "Total charged energy"
},
"total_power_factor": {
"name": "Total power factor"
},
"total_reactive_power": {
"name": "Total reactive power"
},
"vehicle_charging_time": {
"name": "Vehicle charging time"
},
"vehicle_connected_since": {
"name": "Vehicle connected since"
},
"warning_code": {
"name": "Warning",
"state": {
"blackout_protection": "Blackout protection",
"energy_limit_reached": "Energy limit reached",
"ev_does_not_comply_standard": "EV does not comply with standard",
"increased_attachment_temperature": "Increased attachment temperature",
"increased_domestic_plug_temperature": "Increased domestic plug temperature",
"increased_housing_temperature": "Increased housing temperature",
"increased_temperature": "Increased temperature",
"no_attachment_detected": "No attachment detected",
"no_comm_with_type2_attachment": "No communication with Type 2 attachment",
"no_pe": "No protective earth",
"no_warning": "No warning",
"unsupported_charging_mode": "Unsupported charging mode"
}
}
}
},
"exceptions": {
"auth_failure": {
"message": "Authentication failed (HTTP {status_code}). Verify BasicAuth settings. Target: {url}"
},
"authentication_error": {
"message": "Authentication failed. Please check your credentials."
},
"communication_error": {
"message": "Communication error with NRGkick device: {error}"
},
"connection_failed": {
"message": "Network connection failed: {error_details}. Target: {url}"
},
"connection_timeout": {
"message": "Connection timeout after {attempts} attempts. Check power and network. Target: {url}"
},
"generic_error": {
"message": "Connection failed: {error_details}. Target: {url}"
},
"http_error": {
"message": "Device returned HTTP error {status_code} ({status_message}). URL: {url}"
},
"invalid_response": {
"message": "The device returned an invalid response."
},
"json_api_disabled": {
"message": "JSON API is disabled on the device. Enable it in the NRGkick mobile app under Extended \u2192 Local API \u2192 API Variants."
},
"unknown_error": {
"message": "An unknown error occurred."
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -469,6 +469,7 @@ FLOWS = {
"nobo_hub",
"nordpool",
"notion",
"nrgkick",
"ntfy",
"nuheat",
"nuki",

6
homeassistant/generated/integrations.json
View File

@@ -4579,6 +4579,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"nrgkick": {
"name": "NRGkick",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling"
},
"nsw_fuel_station": {
"name": "NSW Fuel Station Price",
"integration_type": "hub",

5
homeassistant/generated/zeroconf.py generated
View File

@@ -782,6 +782,11 @@ ZEROCONF = {
"domain": "nanoleaf",
},
],
"_nrgkick._tcp.local.": [
{
"domain": "nrgkick",
},
],
"_nut._tcp.local.": [
{
"domain": "nut",

10
mypy.ini generated
View File

@@ -3516,6 +3516,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.nrgkick.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.ntfy.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt generated
View File

@@ -1600,6 +1600,9 @@ notifications-android-tv==0.1.5
# homeassistant.components.notify_events
notify-events==1.0.4
# homeassistant.components.nrgkick
nrgkick-api==1.6.0
# homeassistant.components.nederlandse_spoorwegen
nsapi==3.1.3

3
requirements_test_all.txt generated
View File

@@ -1389,6 +1389,9 @@ notifications-android-tv==0.1.5
# homeassistant.components.notify_events
notify-events==1.0.4
# homeassistant.components.nrgkick
nrgkick-api==1.6.0
# homeassistant.components.nederlandse_spoorwegen
nsapi==3.1.3

15
tests/components/nrgkick/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
"""Tests for the NRGkick integration."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the component for tests."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

93
tests/components/nrgkick/conftest.py Normal file
View File

@@ -0,0 +1,93 @@
"""Fixtures for NRGkick integration tests."""
from __future__ import annotations
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, patch
from nrgkick_api import ConnectorType, GridPhases
import pytest
from homeassistant.components.nrgkick.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry, load_json_object_fixture
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.nrgkick.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_nrgkick_api(
mock_info_data: dict[str, Any],
mock_control_data: dict[str, Any],
mock_values_data: dict[str, Any],
) -> Generator[AsyncMock]:
"""Mock the NRGkick API client and patch it where used."""
with (
patch(
"homeassistant.components.nrgkick.NRGkickAPI",
autospec=True,
) as mock_api_cls,
patch(
"homeassistant.components.nrgkick.config_flow.NRGkickAPI",
new=mock_api_cls,
),
):
api = mock_api_cls.return_value
api.test_connection.return_value = True
api.get_info.return_value = mock_info_data
api.get_control.return_value = mock_control_data
api.get_values.return_value = mock_values_data
api.set_current.return_value = {"current_set": 16.0}
api.set_charge_pause.return_value = {"charge_pause": 0}
api.set_energy_limit.return_value = {"energy_limit": 0}
api.set_phase_count.return_value = {"phase_count": 3}
yield api
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="NRGkick Test",
data={
CONF_HOST: "192.168.1.100",
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_pass",
},
entry_id="test_entry_id",
unique_id="TEST123456",
)
@pytest.fixture
def mock_info_data() -> dict[str, Any]:
"""Mock device info data."""
res = load_json_object_fixture("info.json", DOMAIN)
res["connector"]["type"] = ConnectorType.TYPE2
res["grid"]["phases"] = GridPhases.L1_L2_L3
return res
@pytest.fixture
def mock_control_data() -> dict[str, Any]:
"""Mock control data."""
return load_json_object_fixture("control.json", DOMAIN)
@pytest.fixture
def mock_values_data() -> dict[str, Any]:
"""Mock values data."""
return load_json_object_fixture("values_sensor.json", DOMAIN)

6
tests/components/nrgkick/fixtures/control.json Normal file
View File

@@ -0,0 +1,6 @@
{
"current_set": 16.0,
"charge_pause": 0,
"energy_limit": 0,
"phase_count": 3
}

34
tests/components/nrgkick/fixtures/info.json Normal file
View File

@@ -0,0 +1,34 @@
{
"general": {
"device_name": "NRGkick Test",
"serial_number": "TEST123456",
"model_type": "NRGkick Gen2 SIM",
"rated_current": 32.0,
"json_api_version": "v1"
},
"connector": {
"type": "TYPE2",
"serial_number": "CONN123",
"max_current": 32.0,
"phase_count": 3
},
"cellular": { "mode": 3, "rssi": -85, "operator": "Test operator" },
"grid": {
"voltage": 230,
"frequency": 50.0,
"phases": "L1, L2, L3"
},
"network": {
"ip_address": "192.168.1.100",
"mac_address": "AA:BB:CC:DD:EE:FF",
"wifi_ssid": "TestNetwork",
"wifi_rssi": -45
},
"hardware": {
"smartmodule_version": "4.0.0.0",
"bluetooth_version": "1.2.3"
},
"software": {
"firmware_version": "2.1.0"
}
}

37
tests/components/nrgkick/fixtures/values.json Normal file
View File

@@ -0,0 +1,37 @@
{
"powerflow": {
"power": {
"total": 11000,
"l1": 3666,
"l2": 3667,
"l3": 3667
},
"current": {
"total": 16.0,
"l1": 5.33,
"l2": 5.33,
"l3": 5.34
},
"voltage": {
"total": 230.0,
"l1": 230.0,
"l2": 230.0,
"l3": 230.0
},
"frequency": 50.0,
"power_factor": 0.98
},
"energy": {
"charged_energy": 5000,
"session_energy": 2500
},
"status": {
"charging_status": 3
},
"temperatures": {
"housing": 35.0,
"connector_l1": 28.0,
"connector_l2": 29.0,
"connector_l3": 28.5
}
}

61
tests/components/nrgkick/fixtures/values_sensor.json Normal file
View File

@@ -0,0 +1,61 @@
{
"powerflow": {
"total_active_power": 11000,
"l1": {
"voltage": 230.0,
"current": 16.0,
"active_power": 3680,
"reactive_power": 0,
"apparent_power": 3680,
"power_factor": 100
},
"l2": {
"voltage": 230.0,
"current": 16.0,
"active_power": 3680,
"reactive_power": 0,
"apparent_power": 3680,
"power_factor": 100
},
"l3": {
"voltage": 230.0,
"current": 16.0,
"active_power": 3680,
"reactive_power": 0,
"apparent_power": 3680,
"power_factor": 100
},
"charging_voltage": 230.0,
"charging_current": 16.0,
"grid_frequency": 50.0,
"peak_power": 11000,
"total_reactive_power": 0,
"total_apparent_power": 11040,
"total_power_factor": 100,
"n": {
"current": 0.0
}
},
"general": {
"status": 3,
"charging_rate": 11.0,
"vehicle_connect_time": 100,
"vehicle_charging_time": 50,
"charge_count": 5,
"rcd_trigger": 0,
"warning_code": 0,
"error_code": 0
},
"temperatures": {
"housing": 35.0,
"connector_l1": 28.0,
"connector_l2": 29.0,
"connector_l3": 28.5,
"domestic_plug_1": 25.0,
"domestic_plug_2": 25.0
},
"energy": {
"total_charged_energy": 100000,
"charged_energy": 5000
}
}

36
tests/components/nrgkick/snapshots/test_init.ambr Normal file
View File

@@ -0,0 +1,36 @@
# serializer version: 1
# name: test_device
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'http://192.168.1.100',
'connections': set({
tuple(
'mac',
'aa:bb:cc:dd:ee:ff',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'nrgkick',
'TEST123456',
),
}),
'labels': set({
}),
'manufacturer': 'DiniTech',
'model': 'NRGkick Gen2 SIM',
'model_id': None,
'name': 'NRGkick Test',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'TEST123456',
'sw_version': None,
'via_device_id': None,
})
# ---

3197
tests/components/nrgkick/snapshots/test_sensor.ambr Normal file
View File

File diff suppressed because it is too large Load Diff

676
tests/components/nrgkick/test_config_flow.py Normal file
View File

@@ -0,0 +1,676 @@
"""Tests for the NRGkick config flow."""
from __future__ import annotations
from ipaddress import ip_address
from typing import Any
from unittest.mock import AsyncMock
from nrgkick_api import (
NRGkickAPIDisabledError,
NRGkickAuthenticationError,
NRGkickConnectionError,
)
import pytest
from homeassistant.components.nrgkick.api import (
NRGkickApiClientError,
NRGkickApiClientInvalidResponseError,
)
from homeassistant.components.nrgkick.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from tests.common import MockConfigEntry
ZEROCONF_DISCOVERY_INFO = ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.101"),
ip_addresses=[ip_address("192.168.1.101")],
hostname="nrgkick.local.",
name="NRGkick Test._nrgkick._tcp.local.",
port=80,
properties={
"serial_number": "TEST123456",
"device_name": "NRGkick Test",
"model_type": "NRGkick Gen2",
"json_api_enabled": "1",
"json_api_version": "v1",
},
type="_nrgkick._tcp.local.",
)
ZEROCONF_DISCOVERY_INFO_DISABLED_JSON_API = ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.101"),
ip_addresses=[ip_address("192.168.1.101")],
hostname="nrgkick.local.",
name="NRGkick Test._nrgkick._tcp.local.",
port=80,
properties={
"serial_number": "TEST123456",
"device_name": "NRGkick Test",
"model_type": "NRGkick Gen2",
"json_api_enabled": "0",
"json_api_version": "v1",
},
type="_nrgkick._tcp.local.",
)
ZEROCONF_DISCOVERY_INFO_NO_SERIAL = ZeroconfServiceInfo(
ip_address=ip_address("192.168.1.101"),
ip_addresses=[ip_address("192.168.1.101")],
hostname="nrgkick.local.",
name="NRGkick Test._nrgkick._tcp.local.",
port=80,
properties={
"device_name": "NRGkick Test",
"model_type": "NRGkick Gen2",
"json_api_enabled": "1",
"json_api_version": "v1",
},
type="_nrgkick._tcp.local.",
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_flow(hass: HomeAssistant, mock_nrgkick_api: AsyncMock) -> None:
"""Test we can set up successfully without credentials."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {CONF_HOST: "192.168.1.100"}
assert result["result"].unique_id == "TEST123456"
async def test_user_flow_with_credentials(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test we can setup when authentication is required."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "test_user", CONF_PASSWORD: "test_pass"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {
CONF_HOST: "192.168.1.100",
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_pass",
}
assert result["result"].unique_id == "TEST123456"
mock_setup_entry.assert_called_once()
@pytest.mark.parametrize("url", ["http://", ""])
async def test_form_invalid_host_input(
hass: HomeAssistant,
mock_nrgkick_api: AsyncMock,
mock_setup_entry: AsyncMock,
url: str,
) -> None:
"""Test we handle invalid host input during normalization."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: url}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_setup_entry")
async def test_form_fallback_title_when_device_name_missing(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock
) -> None:
"""Test we fall back to a default title when device name is missing."""
mock_nrgkick_api.get_info.return_value = {"general": {"serial_number": "ABC"}}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick"
assert result["data"] == {CONF_HOST: "192.168.1.100"}
@pytest.mark.usefixtures("mock_setup_entry")
async def test_form_invalid_response_when_serial_missing(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, mock_info_data: dict[str, Any]
) -> None:
"""Test we handle invalid device info response."""
mock_nrgkick_api.get_info.return_value = {"general": {"device_name": "NRGkick"}}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: "192.168.1.100"},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_response"}
mock_nrgkick_api.get_info.return_value = mock_info_data
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: "192.168.1.100"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.parametrize(
("exception", "error"),
[
(NRGkickAPIDisabledError, "json_api_disabled"),
(NRGkickApiClientInvalidResponseError, "invalid_response"),
(NRGkickConnectionError, "cannot_connect"),
(NRGkickApiClientError, "unknown"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_flow_errors(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, exception: Exception, error: str
) -> None:
"""Test errors are handled and the flow can recover to CREATE_ENTRY."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": error}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.parametrize(
("exception", "error"),
[
(NRGkickAPIDisabledError, "json_api_disabled"),
(NRGkickAuthenticationError, "invalid_auth"),
(NRGkickApiClientInvalidResponseError, "invalid_response"),
(NRGkickConnectionError, "cannot_connect"),
(NRGkickApiClientError, "unknown"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_flow_auth_errors(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, exception: Exception, error: str
) -> None:
"""Test errors are handled and the flow can recover to CREATE_ENTRY."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
assert result["errors"] == {}
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "user", CONF_PASSWORD: "pass"},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
assert result["errors"] == {"base": error}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "user", CONF_PASSWORD: "pass"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_nrgkick_api: AsyncMock
) -> None:
"""Test we handle already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_user_auth_already_configured(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_nrgkick_api: AsyncMock
) -> None:
"""Test we handle already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
assert result["errors"] == {}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "user", CONF_PASSWORD: "pass"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_discovery(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock
) -> None:
"""Test zeroconf discovery without credentials."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_confirm"
assert result["description_placeholders"] == {
"name": "NRGkick Test",
"device_ip": "192.168.1.101",
}
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_HOST: "192.168.1.101"}
assert result["result"].unique_id == "TEST123456"
async def test_zeroconf_discovery_with_credentials(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test zeroconf discovery flow (auth required)."""
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
assert result["description_placeholders"] == {"device_ip": "192.168.1.101"}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "test_user", CONF_PASSWORD: "test_pass"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {
CONF_HOST: "192.168.1.101",
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_pass",
}
assert result["result"].unique_id == "TEST123456"
mock_setup_entry.assert_called_once()
@pytest.mark.parametrize(
("exception", "reason"),
[
(NRGkickConnectionError, "cannot_connect"),
(NRGkickApiClientInvalidResponseError, "cannot_connect"),
(NRGkickApiClientError, "unknown"),
],
)
async def test_zeroconf_errors(
hass: HomeAssistant,
mock_nrgkick_api: AsyncMock,
exception: Exception,
reason: str,
) -> None:
"""Test zeroconf confirm step reports errors."""
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
async def test_zeroconf_already_configured(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, mock_config_entry: MockConfigEntry
) -> None:
"""Test zeroconf discovery when device is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_DISCOVERY_INFO
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_HOST] == "192.168.1.101"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_json_api_disabled(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock
) -> None:
"""Test zeroconf discovery when JSON API is disabled."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO_DISABLED_JSON_API,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_enable_json_api"
assert result["description_placeholders"] == {
"name": "NRGkick Test",
"device_ip": "192.168.1.101",
}
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {CONF_HOST: "192.168.1.101"}
assert result["result"].unique_id == "TEST123456"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_json_api_disabled_stale_mdns(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock
) -> None:
"""Test zeroconf discovery when JSON API is disabled."""
mock_nrgkick_api.test_connection.side_effect = NRGkickAPIDisabledError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_enable_json_api"
assert result["description_placeholders"] == {
"name": "NRGkick Test",
"device_ip": "192.168.1.101",
}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {CONF_HOST: "192.168.1.101"}
assert result["result"].unique_id == "TEST123456"
@pytest.mark.parametrize(
("exception", "error"),
[
(NRGkickAPIDisabledError, "json_api_disabled"),
(NRGkickApiClientInvalidResponseError, "invalid_response"),
(NRGkickConnectionError, "cannot_connect"),
(NRGkickApiClientError, "unknown"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_json_api_disabled_errors(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock, exception: Exception, error: str
) -> None:
"""Test zeroconf discovery when JSON API is disabled."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO_DISABLED_JSON_API,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_enable_json_api"
assert result["description_placeholders"] == {
"name": "NRGkick Test",
"device_ip": "192.168.1.101",
}
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_enable_json_api"
assert result["errors"] == {"base": error}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "NRGkick Test"
assert result["data"] == {CONF_HOST: "192.168.1.101"}
assert result["result"].unique_id == "TEST123456"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_zeroconf_json_api_disabled_with_credentials(
hass: HomeAssistant, mock_nrgkick_api: AsyncMock
) -> None:
"""Test JSON API disabled flow that requires authentication afterwards."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO_DISABLED_JSON_API,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "zeroconf_enable_json_api"
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "user", CONF_PASSWORD: "pass"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_HOST: "192.168.1.101",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
}
@pytest.mark.parametrize(
("exception", "error"),
[
(NRGkickAPIDisabledError, "json_api_disabled"),
(NRGkickAuthenticationError, "invalid_auth"),
(NRGkickConnectionError, "cannot_connect"),
(NRGkickApiClientError, "unknown"),
],
)
async def test_zeroconf_enable_json_api_auth_errors(
hass: HomeAssistant, mock_nrgkick_api, exception: Exception, error: str
) -> None:
"""Test JSON API enable auth step reports errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO_DISABLED_JSON_API,
)
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
assert result["errors"] == {}
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "user", CONF_PASSWORD: "pass"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "user", CONF_PASSWORD: "pass"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.parametrize(
("exception", "error"),
[
(NRGkickAuthenticationError, "invalid_auth"),
(NRGkickConnectionError, "cannot_connect"),
(NRGkickAPIDisabledError, "json_api_disabled"),
(NRGkickApiClientError, "unknown"),
],
)
async def test_zeroconf_auth_errors(
hass: HomeAssistant,
mock_nrgkick_api: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test zeroconf auth step reports errors."""
mock_nrgkick_api.test_connection.side_effect = NRGkickAuthenticationError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user_auth"
mock_nrgkick_api.test_connection.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "user", CONF_PASSWORD: "pass"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
mock_nrgkick_api.test_connection.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_USERNAME: "user", CONF_PASSWORD: "pass"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_zeroconf_no_serial_number(hass: HomeAssistant) -> None:
"""Test zeroconf discovery without serial number."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_ZEROCONF},
data=ZEROCONF_DISCOVERY_INFO_NO_SERIAL,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_serial_number"

78
tests/components/nrgkick/test_init.py Normal file
View File

@@ -0,0 +1,78 @@
"""Tests for the NRGkick integration initialization."""
from __future__ import annotations
from unittest.mock import AsyncMock
from nrgkick_api import (
NRGkickAPIDisabledError,
NRGkickAuthenticationError,
NRGkickConnectionError,
)
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.nrgkick.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_integration
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_nrgkick_api: AsyncMock
) -> None:
"""Test successful load and unload of entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("exception", "state"),
[
(NRGkickAuthenticationError, ConfigEntryState.SETUP_ERROR),
(NRGkickAPIDisabledError, ConfigEntryState.SETUP_ERROR),
(NRGkickConnectionError, ConfigEntryState.SETUP_RETRY),
(TimeoutError, ConfigEntryState.SETUP_RETRY),
(OSError, ConfigEntryState.SETUP_RETRY),
],
)
async def test_entry_setup_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_nrgkick_api: AsyncMock,
exception: Exception,
state: ConfigEntryState,
) -> None:
"""Test setup entry with failed connection."""
mock_nrgkick_api.get_info.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is state
async def test_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_nrgkick_api: AsyncMock,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test successful load and unload of entry."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device(
identifiers={(DOMAIN, mock_config_entry.unique_id)}
)
assert device is not None
assert device == snapshot

65
tests/components/nrgkick/test_sensor.py Normal file
View File

@@ -0,0 +1,65 @@
"""Tests for the NRGkick sensor platform."""
from unittest.mock import AsyncMock
from nrgkick_api import ChargingStatus, ConnectorType
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.freeze_time("2023-10-21")
async def test_sensor_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_nrgkick_api: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test sensor entities."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_mapped_unknown_values_become_state_unknown(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_nrgkick_api: AsyncMock,
) -> None:
"""Test that enum-like UNKNOWN values map to HA's unknown state."""
mock_nrgkick_api.get_info.return_value["connector"]["type"] = ConnectorType.UNKNOWN
mock_nrgkick_api.get_values.return_value["general"]["status"] = (
ChargingStatus.UNKNOWN
)
await setup_integration(hass, mock_config_entry)
assert hass.states.get("sensor.nrgkick_test_connector_type").state == STATE_UNKNOWN
assert hass.states.get("sensor.nrgkick_test_status").state == STATE_UNKNOWN
async def test_cellular_and_gps_entities_are_gated_by_model_type(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_nrgkick_api: AsyncMock,
) -> None:
"""Test that cellular entities are only created for SIM-capable models."""
mock_nrgkick_api.get_info.return_value["general"]["model_type"] = "NRGkick Gen2"
await setup_integration(hass, mock_config_entry)
assert hass.states.get("sensor.nrgkick_test_cellular_mode") is None
assert hass.states.get("sensor.nrgkick_test_cellular_signal_strength") is None
assert hass.states.get("sensor.nrgkick_test_cellular_operator") is None