github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Enable strict typing for powerwall (#65577)

This commit is contained in:
J. Nick Koston 2022-02-23 01:15:31 -10:00 committed by GitHub
parent 34bae4dcd4
commit e1989e2858
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 326 additions and 394 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -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.*

303
homeassistant/components/powerwall/__init__.py
View File

@ -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:

140
homeassistant/components/powerwall/binary_sensor.py
View File

@ -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()

29
homeassistant/components/powerwall/config_flow.py
View File

@ -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"]

28
homeassistant/components/powerwall/const.py
View File

@ -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"

43
homeassistant/components/powerwall/entity.py
View File

@ -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

2
homeassistant/components/powerwall/manifest.json
View File

@ -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": [
{

50
homeassistant/components/powerwall/models.py Normal file
View File

@ -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

109
homeassistant/components/powerwall/sensor.py
View File

@ -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()

11
mypy.ini
View File

@ -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

2
requirements_all.txt
View File

@ -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

2
requirements_test_all.txt
View File

@ -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