From e1989e285896e07fb6f4a5f09dcf5039c722a16e Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 23 Feb 2022 01:15:31 -1000 Subject: [PATCH] Enable strict typing for powerwall (#65577) --- .strict-typing | 1 + .../components/powerwall/__init__.py | 303 ++++++++---------- .../components/powerwall/binary_sensor.py | 140 +++----- .../components/powerwall/config_flow.py | 29 +- homeassistant/components/powerwall/const.py | 28 +- homeassistant/components/powerwall/entity.py | 43 +-- .../components/powerwall/manifest.json | 2 +- homeassistant/components/powerwall/models.py | 50 +++ homeassistant/components/powerwall/sensor.py | 109 ++----- mypy.ini | 11 + requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- 12 files changed, 326 insertions(+), 394 deletions(-) create mode 100644 homeassistant/components/powerwall/models.py diff --git a/.strict-typing b/.strict-typing index cd148430340..be25c50ae38 100644 --- a/.strict-typing +++ b/.strict-typing @@ -143,6 +143,7 @@ homeassistant.components.openuv.* homeassistant.components.overkiz.* homeassistant.components.persistent_notification.* homeassistant.components.pi_hole.* +homeassistant.components.powerwall.* homeassistant.components.proximity.* homeassistant.components.pvoutput.* homeassistant.components.pure_energie.* diff --git a/homeassistant/components/powerwall/__init__.py b/homeassistant/components/powerwall/__init__.py index 8d91b984d46..10504e2aa06 100644 --- a/homeassistant/components/powerwall/__init__.py +++ b/homeassistant/components/powerwall/__init__.py @@ -1,4 +1,6 @@ """The Tesla Powerwall integration.""" +from __future__ import annotations + import contextlib from datetime import timedelta import logging @@ -16,9 +18,8 @@ from tesla_powerwall import ( from homeassistant.components import persistent_notification from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, Platform -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady -from homeassistant.helpers import entity_registry import homeassistant.helpers.config_validation as cv from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.util.network import is_ip_address @@ -26,21 +27,12 @@ from homeassistant.util.network import is_ip_address from .const import ( DOMAIN, POWERWALL_API_CHANGED, - POWERWALL_API_CHARGE, - POWERWALL_API_DEVICE_TYPE, - POWERWALL_API_GATEWAY_DIN, - POWERWALL_API_GRID_SERVICES_ACTIVE, - POWERWALL_API_GRID_STATUS, - POWERWALL_API_METERS, - POWERWALL_API_SERIAL_NUMBERS, - POWERWALL_API_SITE_INFO, - POWERWALL_API_SITEMASTER, - POWERWALL_API_STATUS, POWERWALL_COORDINATOR, POWERWALL_HTTP_SESSION, - POWERWALL_OBJECT, + POWERWALL_LOGIN_FAILED_COUNT, UPDATE_INTERVAL, ) +from .models import PowerwallBaseInfo, PowerwallData, PowerwallRuntimeData CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False) @@ -50,211 +42,194 @@ _LOGGER = logging.getLogger(__name__) MAX_LOGIN_FAILURES = 5 - -async def _migrate_old_unique_ids(hass, entry_id, powerwall_data): - serial_numbers = powerwall_data[POWERWALL_API_SERIAL_NUMBERS] - site_info = powerwall_data[POWERWALL_API_SITE_INFO] - - @callback - def _async_migrator(entity_entry: entity_registry.RegistryEntry): - parts = entity_entry.unique_id.split("_") - # Check if the unique_id starts with the serial_numbers of the powerwalls - if parts[0 : len(serial_numbers)] != serial_numbers: - # The old unique_id ended with the nomianal_system_engery_kWh so we can use that - # to find the old base unique_id and extract the device_suffix. - normalized_energy_index = ( - len(parts) - 1 - parts[::-1].index(str(site_info.nominal_system_energy)) - ) - device_suffix = parts[normalized_energy_index + 1 :] - - new_unique_id = "_".join([*serial_numbers, *device_suffix]) - _LOGGER.info( - "Migrating unique_id from [%s] to [%s]", - entity_entry.unique_id, - new_unique_id, - ) - return {"new_unique_id": new_unique_id} - return None - - await entity_registry.async_migrate_entries(hass, entry_id, _async_migrator) +API_CHANGED_ERROR_BODY = ( + "It seems like your powerwall uses an unsupported version. " + "Please update the software of your powerwall or if it is " + "already the newest consider reporting this issue.\nSee logs for more information" +) +API_CHANGED_TITLE = "Unknown powerwall software version" -async def _async_handle_api_changed_error( - hass: HomeAssistant, error: MissingAttributeError -): - # The error might include some important information about what exactly changed. - _LOGGER.error(str(error)) - persistent_notification.async_create( - hass, - "It seems like your powerwall uses an unsupported version. " - "Please update the software of your powerwall or if it is " - "already the newest consider reporting this issue.\nSee logs for more information", - title="Unknown powerwall software version", - ) +class PowerwallDataManager: + """Class to manager powerwall data and relogin on failure.""" + + def __init__( + self, + hass: HomeAssistant, + power_wall: Powerwall, + ip_address: str, + password: str | None, + runtime_data: PowerwallRuntimeData, + ) -> None: + """Init the data manager.""" + self.hass = hass + self.ip_address = ip_address + self.password = password + self.runtime_data = runtime_data + self.power_wall = power_wall + + @property + def login_failed_count(self) -> int: + """Return the current number of failed logins.""" + return self.runtime_data[POWERWALL_LOGIN_FAILED_COUNT] + + @property + def api_changed(self) -> int: + """Return true if the api has changed out from under us.""" + return self.runtime_data[POWERWALL_API_CHANGED] + + def _increment_failed_logins(self) -> None: + self.runtime_data[POWERWALL_LOGIN_FAILED_COUNT] += 1 + + def _clear_failed_logins(self) -> None: + self.runtime_data[POWERWALL_LOGIN_FAILED_COUNT] = 0 + + def _recreate_powerwall_login(self) -> None: + """Recreate the login on auth failure.""" + http_session = self.runtime_data[POWERWALL_HTTP_SESSION] + http_session.close() + http_session = requests.Session() + self.runtime_data[POWERWALL_HTTP_SESSION] = http_session + self.power_wall = Powerwall(self.ip_address, http_session=http_session) + self.power_wall.login(self.password or "") + + async def async_update_data(self) -> PowerwallData: + """Fetch data from API endpoint.""" + # Check if we had an error before + _LOGGER.debug("Checking if update failed") + if self.api_changed: + raise UpdateFailed("The powerwall api has changed") + return await self.hass.async_add_executor_job(self._update_data) + + def _update_data(self) -> PowerwallData: + """Fetch data from API endpoint.""" + _LOGGER.debug("Updating data") + for attempt in range(2): + try: + if attempt == 1: + self._recreate_powerwall_login() + data = _fetch_powerwall_data(self.power_wall) + except PowerwallUnreachableError as err: + raise UpdateFailed("Unable to fetch data from powerwall") from err + except MissingAttributeError as err: + _LOGGER.error("The powerwall api has changed: %s", str(err)) + # The error might include some important information about what exactly changed. + persistent_notification.create( + self.hass, API_CHANGED_ERROR_BODY, API_CHANGED_TITLE + ) + self.runtime_data[POWERWALL_API_CHANGED] = True + raise UpdateFailed("The powerwall api has changed") from err + except AccessDeniedError as err: + if attempt == 1: + self._increment_failed_logins() + raise ConfigEntryAuthFailed from err + if self.password is None: + raise ConfigEntryAuthFailed from err + raise UpdateFailed( + f"Login attempt {self.login_failed_count}/{MAX_LOGIN_FAILURES} failed, will retry: {err}" + ) from err + except APIError as err: + raise UpdateFailed(f"Updated failed due to {err}, will retry") from err + else: + self._clear_failed_logins() + return data + raise RuntimeError("unreachable") async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up Tesla Powerwall from a config entry.""" - - entry_id = entry.entry_id - - hass.data.setdefault(DOMAIN, {}) http_session = requests.Session() ip_address = entry.data[CONF_IP_ADDRESS] password = entry.data.get(CONF_PASSWORD) power_wall = Powerwall(ip_address, http_session=http_session) try: - powerwall_data = await hass.async_add_executor_job( - _login_and_fetch_base_info, power_wall, password + base_info = await hass.async_add_executor_job( + _login_and_fetch_base_info, power_wall, ip_address, password ) except PowerwallUnreachableError as err: http_session.close() raise ConfigEntryNotReady from err except MissingAttributeError as err: http_session.close() - await _async_handle_api_changed_error(hass, err) + # The error might include some important information about what exactly changed. + _LOGGER.error("The powerwall api has changed: %s", str(err)) + persistent_notification.async_create( + hass, API_CHANGED_ERROR_BODY, API_CHANGED_TITLE + ) return False except AccessDeniedError as err: _LOGGER.debug("Authentication failed", exc_info=err) http_session.close() raise ConfigEntryAuthFailed from err - await _migrate_old_unique_ids(hass, entry_id, powerwall_data) - - gateway_din = powerwall_data[POWERWALL_API_GATEWAY_DIN] + gateway_din = base_info.gateway_din if gateway_din and entry.unique_id is not None and is_ip_address(entry.unique_id): hass.config_entries.async_update_entry(entry, unique_id=gateway_din) - login_failed_count = 0 + runtime_data = PowerwallRuntimeData( + api_changed=False, + base_info=base_info, + http_session=http_session, + login_failed_count=0, + coordinator=None, + ) - runtime_data = hass.data[DOMAIN][entry.entry_id] = { - POWERWALL_API_CHANGED: False, - POWERWALL_HTTP_SESSION: http_session, - } - - def _recreate_powerwall_login(): - nonlocal http_session - nonlocal power_wall - http_session.close() - http_session = requests.Session() - power_wall = Powerwall(ip_address, http_session=http_session) - runtime_data[POWERWALL_OBJECT] = power_wall - runtime_data[POWERWALL_HTTP_SESSION] = http_session - power_wall.login(password) - - async def _async_login_and_retry_update_data(): - """Retry the update after a failed login.""" - nonlocal login_failed_count - # If the session expired, recreate, relogin, and try again - _LOGGER.debug("Retrying login and updating data") - try: - await hass.async_add_executor_job(_recreate_powerwall_login) - data = await _async_update_powerwall_data(hass, entry, power_wall) - except AccessDeniedError as err: - login_failed_count += 1 - if login_failed_count == MAX_LOGIN_FAILURES: - raise ConfigEntryAuthFailed from err - raise UpdateFailed( - f"Login attempt {login_failed_count}/{MAX_LOGIN_FAILURES} failed, will retry: {err}" - ) from err - except APIError as err: - raise UpdateFailed(f"Updated failed due to {err}, will retry") from err - else: - login_failed_count = 0 - return data - - async def async_update_data(): - """Fetch data from API endpoint.""" - # Check if we had an error before - nonlocal login_failed_count - _LOGGER.debug("Checking if update failed") - if runtime_data[POWERWALL_API_CHANGED]: - return runtime_data[POWERWALL_COORDINATOR].data - - _LOGGER.debug("Updating data") - try: - data = await _async_update_powerwall_data(hass, entry, power_wall) - except AccessDeniedError as err: - if password is None: - raise ConfigEntryAuthFailed from err - return await _async_login_and_retry_update_data() - except APIError as err: - raise UpdateFailed(f"Updated failed due to {err}, will retry") from err - else: - login_failed_count = 0 - return data + manager = PowerwallDataManager(hass, power_wall, ip_address, password, runtime_data) coordinator = DataUpdateCoordinator( hass, _LOGGER, name="Powerwall site", - update_method=async_update_data, + update_method=manager.async_update_data, update_interval=timedelta(seconds=UPDATE_INTERVAL), ) - runtime_data.update( - { - **powerwall_data, - POWERWALL_OBJECT: power_wall, - POWERWALL_COORDINATOR: coordinator, - } - ) - await coordinator.async_config_entry_first_refresh() + runtime_data[POWERWALL_COORDINATOR] = coordinator + + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = runtime_data + hass.config_entries.async_setup_platforms(entry, PLATFORMS) return True -async def _async_update_powerwall_data( - hass: HomeAssistant, entry: ConfigEntry, power_wall: Powerwall -): - """Fetch updated powerwall data.""" - try: - return await hass.async_add_executor_job(_fetch_powerwall_data, power_wall) - except PowerwallUnreachableError as err: - raise UpdateFailed("Unable to fetch data from powerwall") from err - except MissingAttributeError as err: - await _async_handle_api_changed_error(hass, err) - hass.data[DOMAIN][entry.entry_id][POWERWALL_API_CHANGED] = True - # Returns the cached data. This data can also be None - return hass.data[DOMAIN][entry.entry_id][POWERWALL_COORDINATOR].data - - -def _login_and_fetch_base_info(power_wall: Powerwall, password: str): +def _login_and_fetch_base_info( + power_wall: Powerwall, host: str, password: str +) -> PowerwallBaseInfo: """Login to the powerwall and fetch the base info.""" if password is not None: power_wall.login(password) - power_wall.detect_and_pin_version() - return call_base_info(power_wall) + return call_base_info(power_wall, host) -def call_base_info(power_wall): - """Wrap powerwall properties to be a callable.""" +def call_base_info(power_wall: Powerwall, host: str) -> PowerwallBaseInfo: + """Return PowerwallBaseInfo for the device.""" # Make sure the serial numbers always have the same order gateway_din = None - with contextlib.suppress((AssertionError, PowerwallError)): + with contextlib.suppress(AssertionError, PowerwallError): gateway_din = power_wall.get_gateway_din().upper() - return { - POWERWALL_API_SITE_INFO: power_wall.get_site_info(), - POWERWALL_API_STATUS: power_wall.get_status(), - POWERWALL_API_DEVICE_TYPE: power_wall.get_device_type(), - POWERWALL_API_SERIAL_NUMBERS: sorted(power_wall.get_serial_numbers()), - POWERWALL_API_GATEWAY_DIN: gateway_din, - } + return PowerwallBaseInfo( + gateway_din=gateway_din, + site_info=power_wall.get_site_info(), + status=power_wall.get_status(), + device_type=power_wall.get_device_type(), + serial_numbers=sorted(power_wall.get_serial_numbers()), + url=f"https://{host}", + ) -def _fetch_powerwall_data(power_wall): +def _fetch_powerwall_data(power_wall: Powerwall) -> PowerwallData: """Process and update powerwall data.""" - return { - POWERWALL_API_CHARGE: power_wall.get_charge(), - POWERWALL_API_SITEMASTER: power_wall.get_sitemaster(), - POWERWALL_API_METERS: power_wall.get_meters(), - POWERWALL_API_GRID_SERVICES_ACTIVE: power_wall.is_grid_services_active(), - POWERWALL_API_GRID_STATUS: power_wall.get_grid_status(), - } + return PowerwallData( + charge=power_wall.get_charge(), + site_master=power_wall.get_sitemaster(), + meters=power_wall.get_meters(), + grid_services_active=power_wall.is_grid_services_active(), + grid_status=power_wall.get_grid_status(), + ) async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: diff --git a/homeassistant/components/powerwall/binary_sensor.py b/homeassistant/components/powerwall/binary_sensor.py index 5c8ebbdcf19..868d9e3076d 100644 --- a/homeassistant/components/powerwall/binary_sensor.py +++ b/homeassistant/components/powerwall/binary_sensor.py @@ -1,4 +1,5 @@ """Support for powerwall binary sensors.""" + from tesla_powerwall import GridStatus, MeterType from homeassistant.components.binary_sensor import ( @@ -9,19 +10,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .const import ( - DOMAIN, - POWERWALL_API_DEVICE_TYPE, - POWERWALL_API_GRID_SERVICES_ACTIVE, - POWERWALL_API_GRID_STATUS, - POWERWALL_API_METERS, - POWERWALL_API_SERIAL_NUMBERS, - POWERWALL_API_SITE_INFO, - POWERWALL_API_SITEMASTER, - POWERWALL_API_STATUS, - POWERWALL_COORDINATOR, -) +from .const import DOMAIN from .entity import PowerWallEntity +from .models import PowerwallRuntimeData async def async_setup_entry( @@ -29,152 +20,103 @@ async def async_setup_entry( config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: - """Set up the August sensors.""" - powerwall_data = hass.data[DOMAIN][config_entry.entry_id] - - coordinator = powerwall_data[POWERWALL_COORDINATOR] - site_info = powerwall_data[POWERWALL_API_SITE_INFO] - device_type = powerwall_data[POWERWALL_API_DEVICE_TYPE] - status = powerwall_data[POWERWALL_API_STATUS] - powerwalls_serial_numbers = powerwall_data[POWERWALL_API_SERIAL_NUMBERS] - - entities = [] - for sensor_class in ( - PowerWallRunningSensor, - PowerWallGridServicesActiveSensor, - PowerWallGridStatusSensor, - PowerWallConnectedSensor, - PowerWallChargingStatusSensor, - ): - entities.append( - sensor_class( - coordinator, site_info, status, device_type, powerwalls_serial_numbers + """Set up the powerwall sensors.""" + powerwall_data: PowerwallRuntimeData = hass.data[DOMAIN][config_entry.entry_id] + async_add_entities( + [ + sensor_class(powerwall_data) + for sensor_class in ( + PowerWallRunningSensor, + PowerWallGridServicesActiveSensor, + PowerWallGridStatusSensor, + PowerWallConnectedSensor, + PowerWallChargingStatusSensor, ) - ) - - async_add_entities(entities, True) + ] + ) class PowerWallRunningSensor(PowerWallEntity, BinarySensorEntity): """Representation of an Powerwall running sensor.""" - @property - def name(self): - """Device Name.""" - return "Powerwall Status" + _attr_name = "Powerwall Status" + _attr_device_class = BinarySensorDeviceClass.POWER @property - def device_class(self): - """Device Class.""" - return BinarySensorDeviceClass.POWER - - @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_running" @property - def is_on(self): + def is_on(self) -> bool: """Get the powerwall running state.""" - return self.coordinator.data[POWERWALL_API_SITEMASTER].is_running + return self.data.site_master.is_running class PowerWallConnectedSensor(PowerWallEntity, BinarySensorEntity): """Representation of an Powerwall connected sensor.""" - @property - def name(self): - """Device Name.""" - return "Powerwall Connected to Tesla" + _attr_name = "Powerwall Connected to Tesla" + _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY @property - def device_class(self): - """Device Class.""" - return BinarySensorDeviceClass.CONNECTIVITY - - @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_connected_to_tesla" @property - def is_on(self): + def is_on(self) -> bool: """Get the powerwall connected to tesla state.""" - return self.coordinator.data[POWERWALL_API_SITEMASTER].is_connected_to_tesla + return self.data.site_master.is_connected_to_tesla class PowerWallGridServicesActiveSensor(PowerWallEntity, BinarySensorEntity): """Representation of a Powerwall grid services active sensor.""" - @property - def name(self): - """Device Name.""" - return "Grid Services Active" + _attr_name = "Grid Services Active" + _attr_device_class = BinarySensorDeviceClass.POWER @property - def device_class(self): - """Device Class.""" - return BinarySensorDeviceClass.POWER - - @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_grid_services_active" @property - def is_on(self): + def is_on(self) -> bool: """Grid services is active.""" - return self.coordinator.data[POWERWALL_API_GRID_SERVICES_ACTIVE] + return self.data.grid_services_active class PowerWallGridStatusSensor(PowerWallEntity, BinarySensorEntity): """Representation of an Powerwall grid status sensor.""" - @property - def name(self): - """Device Name.""" - return "Grid Status" + _attr_name = "Grid Status" + _attr_device_class = BinarySensorDeviceClass.POWER @property - def device_class(self): - """Device Class.""" - return BinarySensorDeviceClass.POWER - - @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_grid_status" @property - def is_on(self): + def is_on(self) -> bool: """Grid is online.""" - return self.coordinator.data[POWERWALL_API_GRID_STATUS] == GridStatus.CONNECTED + return self.data.grid_status == GridStatus.CONNECTED class PowerWallChargingStatusSensor(PowerWallEntity, BinarySensorEntity): """Representation of an Powerwall charging status sensor.""" - @property - def name(self): - """Device Name.""" - return "Powerwall Charging" + _attr_name = "Powerwall Charging" + _attr_device_class = BinarySensorDeviceClass.BATTERY_CHARGING @property - def device_class(self): - """Device Class.""" - return BinarySensorDeviceClass.BATTERY_CHARGING - - @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_powerwall_charging" @property - def is_on(self): + def is_on(self) -> bool: """Powerwall is charging.""" # is_sending_to returns true for values greater than 100 watts - return ( - self.coordinator.data[POWERWALL_API_METERS] - .get_meter(MeterType.BATTERY) - .is_sending_to() - ) + return self.data.meters.get_meter(MeterType.BATTERY).is_sending_to() diff --git a/homeassistant/components/powerwall/config_flow.py b/homeassistant/components/powerwall/config_flow.py index 9814a52d8a0..08e9f90df1b 100644 --- a/homeassistant/components/powerwall/config_flow.py +++ b/homeassistant/components/powerwall/config_flow.py @@ -9,6 +9,7 @@ from tesla_powerwall import ( MissingAttributeError, Powerwall, PowerwallUnreachableError, + SiteInfo, ) import voluptuous as vol @@ -23,11 +24,12 @@ from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -def _login_and_fetch_site_info(power_wall: Powerwall, password: str): +def _login_and_fetch_site_info( + power_wall: Powerwall, password: str +) -> tuple[SiteInfo, str]: """Login to the powerwall and fetch the base info.""" if password is not None: power_wall.login(password) - power_wall.detect_and_pin_version() return power_wall.get_site_info(), power_wall.get_gateway_din() @@ -60,7 +62,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize the powerwall flow.""" self.ip_address: str | None = None self.title: str | None = None @@ -101,7 +103,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_confirm_discovery() async def _async_try_connect( - self, user_input + self, user_input: dict[str, Any] ) -> tuple[dict[str, Any] | None, dict[str, str] | None]: """Try to connect to the powerwall.""" info = None @@ -120,7 +122,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return errors, info - async def async_step_confirm_discovery(self, user_input=None) -> FlowResult: + async def async_step_confirm_discovery( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Confirm a discovered powerwall.""" assert self.ip_address is not None assert self.unique_id is not None @@ -148,9 +152,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): }, ) - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle the initial step.""" - errors = {} + errors: dict[str, str] | None = {} if user_input is not None: errors, info = await self._async_try_connect(user_input) if not errors: @@ -176,9 +182,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_reauth_confirm(self, user_input=None): + async def async_step_reauth_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle reauth confirmation.""" - errors = {} + assert self.reauth_entry is not None + errors: dict[str, str] | None = {} if user_input is not None: entry_data = self.reauth_entry.data errors, _ = await self._async_try_connect( @@ -197,7 +206,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_reauth(self, data): + async def async_step_reauth(self, data: dict[str, str]) -> FlowResult: """Handle configuration by re-auth.""" self.reauth_entry = self.hass.config_entries.async_get_entry( self.context["entry_id"] diff --git a/homeassistant/components/powerwall/const.py b/homeassistant/components/powerwall/const.py index b2f0e3afe80..b2738bce4ac 100644 --- a/homeassistant/components/powerwall/const.py +++ b/homeassistant/components/powerwall/const.py @@ -1,34 +1,20 @@ """Constants for the Tesla Powerwall integration.""" +from typing import Final DOMAIN = "powerwall" -POWERWALL_OBJECT = "powerwall" -POWERWALL_COORDINATOR = "coordinator" -POWERWALL_API_CHANGED = "api_changed" +POWERWALL_BASE_INFO: Final = "base_info" +POWERWALL_COORDINATOR: Final = "coordinator" +POWERWALL_API_CHANGED: Final = "api_changed" +POWERWALL_HTTP_SESSION: Final = "http_session" +POWERWALL_LOGIN_FAILED_COUNT: Final = "login_failed_count" -UPDATE_INTERVAL = 30 +UPDATE_INTERVAL = 5 ATTR_FREQUENCY = "frequency" ATTR_INSTANT_AVERAGE_VOLTAGE = "instant_average_voltage" ATTR_INSTANT_TOTAL_CURRENT = "instant_total_current" ATTR_IS_ACTIVE = "is_active" -STATUS_VERSION = "version" - -POWERWALL_SITE_NAME = "site_name" - -POWERWALL_API_METERS = "meters" -POWERWALL_API_CHARGE = "charge" -POWERWALL_API_GRID_SERVICES_ACTIVE = "grid_services_active" -POWERWALL_API_GRID_STATUS = "grid_status" -POWERWALL_API_SITEMASTER = "sitemaster" -POWERWALL_API_STATUS = "status" -POWERWALL_API_DEVICE_TYPE = "device_type" -POWERWALL_API_SITE_INFO = "site_info" -POWERWALL_API_SERIAL_NUMBERS = "serial_numbers" -POWERWALL_API_GATEWAY_DIN = "gateway_din" - -POWERWALL_HTTP_SESSION = "http_session" - MODEL = "PowerWall 2" MANUFACTURER = "Tesla" diff --git a/homeassistant/components/powerwall/entity.py b/homeassistant/components/powerwall/entity.py index ae647a080c0..20871944663 100644 --- a/homeassistant/components/powerwall/entity.py +++ b/homeassistant/components/powerwall/entity.py @@ -3,30 +3,37 @@ from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.update_coordinator import CoordinatorEntity -from .const import DOMAIN, MANUFACTURER, MODEL +from .const import ( + DOMAIN, + MANUFACTURER, + MODEL, + POWERWALL_BASE_INFO, + POWERWALL_COORDINATOR, +) +from .models import PowerwallData, PowerwallRuntimeData -class PowerWallEntity(CoordinatorEntity): +class PowerWallEntity(CoordinatorEntity[PowerwallData]): """Base class for powerwall entities.""" - def __init__( - self, coordinator, site_info, status, device_type, powerwalls_serial_numbers - ): - """Initialize the sensor.""" + def __init__(self, powerwall_data: PowerwallRuntimeData) -> None: + """Initialize the entity.""" + base_info = powerwall_data[POWERWALL_BASE_INFO] + coordinator = powerwall_data[POWERWALL_COORDINATOR] + assert coordinator is not None super().__init__(coordinator) - self._site_info = site_info - self._device_type = device_type - self._version = status.version # The serial numbers of the powerwalls are unique to every site - self.base_unique_id = "_".join(powerwalls_serial_numbers) - - @property - def device_info(self) -> DeviceInfo: - """Powerwall device info.""" - return DeviceInfo( + self.base_unique_id = "_".join(base_info.serial_numbers) + self._attr_device_info = DeviceInfo( identifiers={(DOMAIN, self.base_unique_id)}, manufacturer=MANUFACTURER, - model=f"{MODEL} ({self._device_type.name})", - name=self._site_info.site_name, - sw_version=self._version, + model=f"{MODEL} ({base_info.device_type.name})", + name=base_info.site_info.site_name, + sw_version=base_info.status.version, + configuration_url=base_info.url, ) + + @property + def data(self) -> PowerwallData: + """Return the coordinator data.""" + return self.coordinator.data diff --git a/homeassistant/components/powerwall/manifest.json b/homeassistant/components/powerwall/manifest.json index 55c7ab41e64..be5d4678e27 100644 --- a/homeassistant/components/powerwall/manifest.json +++ b/homeassistant/components/powerwall/manifest.json @@ -3,7 +3,7 @@ "name": "Tesla Powerwall", "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/powerwall", - "requirements": ["tesla-powerwall==0.3.15"], + "requirements": ["tesla-powerwall==0.3.17"], "codeowners": ["@bdraco", "@jrester"], "dhcp": [ { diff --git a/homeassistant/components/powerwall/models.py b/homeassistant/components/powerwall/models.py new file mode 100644 index 00000000000..472d9e59304 --- /dev/null +++ b/homeassistant/components/powerwall/models.py @@ -0,0 +1,50 @@ +"""The powerwall integration models.""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import TypedDict + +from requests import Session +from tesla_powerwall import ( + DeviceType, + GridStatus, + MetersAggregates, + PowerwallStatus, + SiteInfo, + SiteMaster, +) + +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator + + +@dataclass +class PowerwallBaseInfo: + """Base information for the powerwall integration.""" + + gateway_din: None | str + site_info: SiteInfo + status: PowerwallStatus + device_type: DeviceType + serial_numbers: list[str] + url: str + + +@dataclass +class PowerwallData: + """Point in time data for the powerwall integration.""" + + charge: float + site_master: SiteMaster + meters: MetersAggregates + grid_services_active: bool + grid_status: GridStatus + + +class PowerwallRuntimeData(TypedDict): + """Run time data for the powerwall.""" + + coordinator: DataUpdateCoordinator | None + login_failed_count: int + base_info: PowerwallBaseInfo + api_changed: bool + http_session: Session diff --git a/homeassistant/components/powerwall/sensor.py b/homeassistant/components/powerwall/sensor.py index da0e7e7f599..a48726211b2 100644 --- a/homeassistant/components/powerwall/sensor.py +++ b/homeassistant/components/powerwall/sensor.py @@ -1,7 +1,7 @@ -"""Support for August sensors.""" +"""Support for powerwall sensors.""" from __future__ import annotations -import logging +from typing import Any from tesla_powerwall import MeterType @@ -21,72 +21,43 @@ from .const import ( ATTR_INSTANT_TOTAL_CURRENT, ATTR_IS_ACTIVE, DOMAIN, - POWERWALL_API_CHARGE, - POWERWALL_API_DEVICE_TYPE, - POWERWALL_API_METERS, - POWERWALL_API_SERIAL_NUMBERS, - POWERWALL_API_SITE_INFO, - POWERWALL_API_STATUS, POWERWALL_COORDINATOR, ) from .entity import PowerWallEntity +from .models import PowerwallData, PowerwallRuntimeData _METER_DIRECTION_EXPORT = "export" _METER_DIRECTION_IMPORT = "import" _METER_DIRECTIONS = [_METER_DIRECTION_EXPORT, _METER_DIRECTION_IMPORT] -_LOGGER = logging.getLogger(__name__) - - async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: - """Set up the August sensors.""" - powerwall_data = hass.data[DOMAIN][config_entry.entry_id] - _LOGGER.debug("Powerwall_data: %s", powerwall_data) - + """Set up the powerwall sensors.""" + powerwall_data: PowerwallRuntimeData = hass.data[DOMAIN][config_entry.entry_id] coordinator = powerwall_data[POWERWALL_COORDINATOR] - site_info = powerwall_data[POWERWALL_API_SITE_INFO] - device_type = powerwall_data[POWERWALL_API_DEVICE_TYPE] - status = powerwall_data[POWERWALL_API_STATUS] - powerwalls_serial_numbers = powerwall_data[POWERWALL_API_SERIAL_NUMBERS] - - entities: list[SensorEntity] = [] - # coordinator.data[POWERWALL_API_METERS].meters holds all meters that are available - for meter in coordinator.data[POWERWALL_API_METERS].meters: - entities.append( - PowerWallEnergySensor( - meter, - coordinator, - site_info, - status, - device_type, - powerwalls_serial_numbers, - ) - ) + assert coordinator is not None + data: PowerwallData = coordinator.data + entities: list[ + PowerWallEnergySensor | PowerWallEnergyDirectionSensor | PowerWallChargeSensor + ] = [] + for meter in data.meters.meters: + entities.append(PowerWallEnergySensor(powerwall_data, meter)) for meter_direction in _METER_DIRECTIONS: entities.append( PowerWallEnergyDirectionSensor( + powerwall_data, meter, - coordinator, - site_info, - status, - device_type, - powerwalls_serial_numbers, meter_direction, ) ) - entities.append( - PowerWallChargeSensor( - coordinator, site_info, status, device_type, powerwalls_serial_numbers - ) - ) + entities.append(PowerWallChargeSensor(powerwall_data)) - async_add_entities(entities, True) + async_add_entities(entities) class PowerWallChargeSensor(PowerWallEntity, SensorEntity): @@ -98,14 +69,14 @@ class PowerWallChargeSensor(PowerWallEntity, SensorEntity): _attr_device_class = SensorDeviceClass.BATTERY @property - def unique_id(self): + def unique_id(self) -> str: """Device Uniqueid.""" return f"{self.base_unique_id}_charge" @property - def native_value(self): + def native_value(self) -> int: """Get the current value in percentage.""" - return round(self.coordinator.data[POWERWALL_API_CHARGE]) + return round(self.data.charge) class PowerWallEnergySensor(PowerWallEntity, SensorEntity): @@ -115,19 +86,9 @@ class PowerWallEnergySensor(PowerWallEntity, SensorEntity): _attr_native_unit_of_measurement = POWER_KILO_WATT _attr_device_class = SensorDeviceClass.POWER - def __init__( - self, - meter: MeterType, - coordinator, - site_info, - status, - device_type, - powerwalls_serial_numbers, - ): + def __init__(self, powerwall_data: PowerwallRuntimeData, meter: MeterType) -> None: """Initialize the sensor.""" - super().__init__( - coordinator, site_info, status, device_type, powerwalls_serial_numbers - ) + super().__init__(powerwall_data) self._meter = meter self._attr_name = f"Powerwall {self._meter.value.title()} Now" self._attr_unique_id = ( @@ -135,18 +96,14 @@ class PowerWallEnergySensor(PowerWallEntity, SensorEntity): ) @property - def native_value(self): + def native_value(self) -> float: """Get the current value in kW.""" - return ( - self.coordinator.data[POWERWALL_API_METERS] - .get_meter(self._meter) - .get_power(precision=3) - ) + return self.data.meters.get_meter(self._meter).get_power(precision=3) @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the device specific state attributes.""" - meter = self.coordinator.data[POWERWALL_API_METERS].get_meter(self._meter) + meter = self.data.meters.get_meter(self._meter) return { ATTR_FREQUENCY: round(meter.frequency, 1), ATTR_INSTANT_AVERAGE_VOLTAGE: round(meter.average_voltage, 1), @@ -164,18 +121,12 @@ class PowerWallEnergyDirectionSensor(PowerWallEntity, SensorEntity): def __init__( self, + powerwall_data: PowerwallRuntimeData, meter: MeterType, - coordinator, - site_info, - status, - device_type, - powerwalls_serial_numbers, - meter_direction, - ): + meter_direction: str, + ) -> None: """Initialize the sensor.""" - super().__init__( - coordinator, site_info, status, device_type, powerwalls_serial_numbers - ) + super().__init__(powerwall_data) self._meter = meter self._meter_direction = meter_direction self._attr_name = ( @@ -186,9 +137,9 @@ class PowerWallEnergyDirectionSensor(PowerWallEntity, SensorEntity): ) @property - def native_value(self): + def native_value(self) -> float: """Get the current value in kWh.""" - meter = self.coordinator.data[POWERWALL_API_METERS].get_meter(self._meter) + meter = self.data.meters.get_meter(self._meter) if self._meter_direction == _METER_DIRECTION_EXPORT: return meter.get_energy_exported() return meter.get_energy_imported() diff --git a/mypy.ini b/mypy.ini index dbbbe824540..b508045d623 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1382,6 +1382,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.powerwall.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.proximity.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/requirements_all.txt b/requirements_all.txt index 66c31d0b25d..75ea65b43c4 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2348,7 +2348,7 @@ temperusb==1.5.3 # tensorflow==2.5.0 # homeassistant.components.powerwall -tesla-powerwall==0.3.15 +tesla-powerwall==0.3.17 # homeassistant.components.tesla_wall_connector tesla-wall-connector==1.0.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 3ebda80093b..f14ef8acb71 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1445,7 +1445,7 @@ tailscale==0.2.0 tellduslive==0.10.11 # homeassistant.components.powerwall -tesla-powerwall==0.3.15 +tesla-powerwall==0.3.17 # homeassistant.components.tesla_wall_connector tesla-wall-connector==1.0.1