diff --git a/homeassistant/configuration.yaml b/homeassistant/configuration.yaml index ac7569c..5cd822d 100644 --- a/homeassistant/configuration.yaml +++ b/homeassistant/configuration.yaml @@ -54,11 +54,10 @@ google_assistant: - BRIGHT_LIGHTS - ENTRY_LIGHTS -# auth_header: -# allow_bypass_login: true -# # username_header: X-Forwarded-Preferred-Username -# username_header: X-Homeassistant-User -# debug: true +auth_header: + allow_bypass_login: true + username_header: X-Homeassistant-User + debug: true openid: client_id: !secret openid_client_id diff --git a/homeassistant/custom_components/sems/__init__.py b/homeassistant/custom_components/sems/__init__.py index 4363647..8db3ac5 100644 --- a/homeassistant/custom_components/sems/__init__.py +++ b/homeassistant/custom_components/sems/__init__.py @@ -2,128 +2,83 @@ from __future__ import annotations -import asyncio -from datetime import timedelta import logging +from dataclasses import dataclass +from datetime import timedelta +from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant -from homeassistant.core import callback +from homeassistant.helpers import config_validation as cv from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed -from homeassistant.helpers.entity_registry import async_migrate_entries from .const import ( CONF_SCAN_INTERVAL, CONF_STATION_ID, DEFAULT_SCAN_INTERVAL, DOMAIN, + GOODWE_SPELLING, PLATFORMS, ) from .sems_api import SemsApi _LOGGER: logging.Logger = logging.getLogger(__package__) +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + + +@dataclass(slots=True) +class SemsRuntimeData: + """Runtime data stored on the config entry.""" + + api: SemsApi + coordinator: SemsDataUpdateCoordinator + + +type SemsConfigEntry = ConfigEntry[SemsRuntimeData] + + +@dataclass(slots=True) +class SemsData: + """Runtime SEMS data returned by the coordinator.""" + + inverters: dict[str, dict[str, Any]] + homekit: dict[str, Any] | None = None + currency: str | None = None + async def async_setup(hass: HomeAssistant, config: dict): """Set up the sems component.""" - # Ensure our name space for storing objects is a known type. A dict is - # common/preferred as it allows a separate instance of your class for each - # instance that has been created in the UI. - hass.data.setdefault(DOMAIN, {}) - return True -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: +async def async_setup_entry(hass: HomeAssistant, entry: SemsConfigEntry) -> bool: """Set up sems from a config entry.""" - semsApi = SemsApi(hass, entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD]) - coordinator = SemsDataUpdateCoordinator(hass, semsApi, entry) - await coordinator.async_config_entry_first_refresh() - hass.data[DOMAIN][entry.entry_id] = coordinator + sems_api = SemsApi(hass, entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD]) + coordinator = SemsDataUpdateCoordinator(hass, sems_api, entry) + entry.runtime_data = SemsRuntimeData(api=sems_api, coordinator=coordinator) + await coordinator.async_config_entry_first_refresh() await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True -# async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry): -# """Set up this integration using UI.""" -# if hass.data.get(DOMAIN) is None: -# hass.data.setdefault(DOMAIN, {}) -# _LOGGER.info(STARTUP_MESSAGE) - -# username = entry.data.get(CONF_USERNAME) -# password = entry.data.get(CONF_PASSWORD) - -# conn = Connection(username, password) -# client = MyenergiClient(conn) - -# coordinator = MyenergiDataUpdateCoordinator(hass, client=client, entry=entry) -# await coordinator.async_config_entry_first_refresh() - -# hass.data[DOMAIN][entry.entry_id] = coordinator - -# for platform in PLATFORMS: -# if entry.options.get(platform, True): -# coordinator.platforms.append(platform) -# hass.async_add_job( -# hass.config_entries.async_forward_entry_setup(entry, platform) -# ) - -# entry.add_update_listener(async_reload_entry) -# return True - - -async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: +async def async_unload_entry(hass: HomeAssistant, entry: SemsConfigEntry) -> bool: """Unload a config entry.""" - unload_ok = all( - await asyncio.gather( - *[ - hass.config_entries.async_forward_entry_unload(entry, platform) - for platform in PLATFORMS - ] - ) - ) - if unload_ok: - hass.data[DOMAIN].pop(entry.entry_id) - - return unload_ok + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) -# async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: -# """Handle removal of an entry.""" -# coordinator = hass.data[DOMAIN][entry.entry_id] -# unloaded = all( -# await asyncio.gather( -# *[ -# hass.config_entries.async_forward_entry_unload(entry, platform) -# for platform in PLATFORMS -# if platform in coordinator.platforms -# ] -# ) -# ) -# if unloaded: -# hass.data[DOMAIN].pop(entry.entry_id) - -# return unloaded - - -# async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: -# """Reload config entry.""" -# await async_unload_entry(hass, entry) -# await async_setup_entry(hass, entry) - - -class SemsDataUpdateCoordinator(DataUpdateCoordinator): +class SemsDataUpdateCoordinator(DataUpdateCoordinator[SemsData]): """Class to manage fetching data from the API.""" - def __init__(self, hass: HomeAssistant, semsApi: SemsApi, entry) -> None: + def __init__( + self, hass: HomeAssistant, sems_api: SemsApi, entry: ConfigEntry + ) -> None: """Initialize.""" - self.semsApi = semsApi - self.platforms = [] - self.stationId = entry.data[CONF_STATION_ID] - self.hass = hass + self.sems_api = sems_api + self.station_id = entry.data[CONF_STATION_ID] update_interval = timedelta( seconds=entry.data.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) @@ -136,70 +91,82 @@ class SemsDataUpdateCoordinator(DataUpdateCoordinator): update_interval=update_interval, ) - async def _async_update_data(self): + async def _async_update_data(self) -> SemsData: """Fetch data from API endpoint. This is the place to pre-process the data to lookup tables so entities can quickly look up their data. """ + # Note: asyncio.TimeoutError and aiohttp.ClientError are already + # handled by the data update coordinator. + # async with async_timeout.timeout(10): try: - # Note: asyncio.TimeoutError and aiohttp.ClientError are already - # handled by the data update coordinator. - # async with async_timeout.timeout(10): result = await self.hass.async_add_executor_job( - self.semsApi.getData, self.stationId + self.sems_api.getData, self.station_id ) + except Exception as err: + raise UpdateFailed(f"Error communicating with API: {err}") from err + else: _LOGGER.debug("semsApi.getData result: %s", result) - # _LOGGER.warning("SEMS - Try get getPowerStationIds") - # powerStationIds = await self.hass.async_add_executor_job( - # self.semsApi.getPowerStationIds - # ) - # _LOGGER.warning( - # "SEMS - getPowerStationIds: Found power station IDs: %s", - # powerStationIds, - # ) - - inverters = result["inverter"] - - # found = [] - # _LOGGER.debug("Found inverters: %s", inverters) - data = {} - if inverters is None: - # something went wrong, probably token could not be fetched + inverters = result.get("inverter") + inverters_by_sn: dict[str, dict[str, Any]] = {} + if not inverters or not isinstance(inverters, list): raise UpdateFailed( - "Error communicating with API, probably token could not be fetched, see debug logs" + "Error communicating with API: invalid or missing inverter data. See debug logs." ) + + # Get Inverter Data for inverter in inverters: - name = inverter["invert_full"]["name"] - # powerstation_id = inverter["invert_full"]["powerstation_id"] - sn = inverter["invert_full"]["sn"] + inverter_full = inverter.get("invert_full") + if not isinstance(inverter_full, dict): + continue + + name = inverter_full.get("name") + sn = inverter_full.get("sn") + if not isinstance(sn, str): + continue + _LOGGER.debug("Found inverter attribute %s %s", name, sn) - data[sn] = inverter["invert_full"] + inverters_by_sn[sn] = inverter_full - hasPowerflow = result["hasPowerflow"] - hasEnergeStatisticsCharts = result["hasEnergeStatisticsCharts"] + # Add currency + kpi = result.get("kpi") + if not isinstance(kpi, dict): + kpi = {} + currency = kpi.get("currency") - if hasPowerflow: + has_powerflow = bool(result.get("hasPowerflow")) + has_energy_statistics_charts = bool( + result.get(GOODWE_SPELLING.hasEnergyStatisticsCharts) + ) + + homekit: dict[str, Any] | None = None + + if has_powerflow: _LOGGER.debug("Found powerflow data") - if hasEnergeStatisticsCharts: - StatisticsCharts = { - f"Charts_{key}": val - for key, val in result["energeStatisticsCharts"].items() - } - StatisticsTotals = { - f"Totals_{key}": val - for key, val in result["energeStatisticsTotals"].items() - } - powerflow = { - **result["powerflow"], - **StatisticsCharts, - **StatisticsTotals, - } - else: - powerflow = result["powerflow"] + powerflow = result.get("powerflow") + if not isinstance(powerflow, dict): + powerflow = {} - powerflow["sn"] = result["homKit"]["sn"] + if has_energy_statistics_charts: + charts = result.get(GOODWE_SPELLING.energyStatisticsCharts) + if not isinstance(charts, dict): + charts = {} + totals = result.get(GOODWE_SPELLING.energyStatisticsTotals) + if not isinstance(totals, dict): + totals = {} + + powerflow = { + **powerflow, + **{f"Charts_{key}": val for key, val in charts.items()}, + **{f"Totals_{key}": val for key, val in totals.items()}, + } + + homekit_data = result.get(GOODWE_SPELLING.homeKit) + if not isinstance(homekit_data, dict): + homekit_data = {} + powerflow["sn"] = homekit_data.get("sn") # Goodwe 'Power Meter' (not HomeKit) doesn't have a sn # Let's put something in, otherwise we can't see the data. @@ -208,72 +175,16 @@ class SemsDataUpdateCoordinator(DataUpdateCoordinator): # _LOGGER.debug("homeKit sn: %s", result["homKit"]["sn"]) # This seems more accurate than the Chart_sum - powerflow["all_time_generation"] = result["kpi"]["total_power"] + powerflow["all_time_generation"] = kpi.get("total_power") - data["homeKit"] = powerflow + homekit = powerflow + data = SemsData( + inverters=inverters_by_sn, homekit=homekit, currency=currency + ) _LOGGER.debug("Resulting data: %s", data) return data - # except ApiError as err: - except Exception as err: - # logging.exception("Something awful happened!") - raise UpdateFailed(f"Error communicating with API: {err}") from err -# # migrate to _power ids for inverter entry -# async def async_migrate_entry(hass, config_entry): -# """Migrate old entry.""" -# _LOGGER.debug( -# "Migrating configuration from version %s.%s", -# config_entry.version, -# config_entry.minor_version, -# ) - -# if config_entry.version < 7: -# # get existing entities for device -# semsApi = SemsApi( -# hass, config_entry.data[CONF_USERNAME], config_entry.data[CONF_PASSWORD] -# ) -# coordinator = SemsDataUpdateCoordinator(hass, semsApi, config_entry) -# await coordinator.async_config_entry_first_refresh() - -# _LOGGER.debug(f"found inverter {coordinator.data}") - -# for idx, ent in enumerate(coordinator.data): -# _LOGGER.debug("Found inverter: %s", ent) - -# old_unique_id = f"{ent}" -# new_unique_id = f"{old_unique_id}-XXX" -# _LOGGER.debug( -# "Old unique id: %s; new unique id: %s", old_unique_id, new_unique_id -# ) - -# @callback -# def update_unique_id(entity_entry): -# """Update unique ID of entity entry.""" -# return { -# "new_unique_id": entity_entry.unique_id.replace( -# old_unique_id, new_unique_id -# ) -# } - -# if old_unique_id != new_unique_id: -# await async_migrate_entries( -# hass, config_entry.entry_id, update_unique_id -# ) - -# hass.config_entries.async_update_entry( -# config_entry, unique_id=new_unique_id -# ) -# # version = 7 -# _LOGGER.info( -# "Migrated unique id from %s to %s", old_unique_id, new_unique_id -# ) - -# _LOGGER.info( -# "Migration from version %s.%s successful", -# config_entry.version, -# config_entry.minor_version, -# ) - -# return True +# Type alias to make type inference working for pylance +type SemsCoordinator = SemsDataUpdateCoordinator diff --git a/homeassistant/custom_components/sems/config_flow.py b/homeassistant/custom_components/sems/config_flow.py index 66e93ea..a219af8 100644 --- a/homeassistant/custom_components/sems/config_flow.py +++ b/homeassistant/custom_components/sems/config_flow.py @@ -6,17 +6,13 @@ import logging from typing import Any from homeassistant import config_entries +from homeassistant.const import CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError -from .const import DOMAIN, SEMS_CONFIG_SCHEMA, CONF_STATION_ID +from .const import CONF_STATION_ID, DOMAIN, SEMS_CONFIG_SCHEMA from .sems_api import SemsApi -from homeassistant.const import ( - CONF_PASSWORD, - CONF_USERNAME, -) - _LOGGER = logging.getLogger(__name__) @@ -73,7 +69,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): async def async_step_user( self, user_input: dict[str, Any] | None = None - ) -> dict[str, Any]: + ) -> config_entries.ConfigFlowResult: """Handle the initial step.""" if user_input is None: return self.async_show_form(step_id="user", data_schema=SEMS_CONFIG_SCHEMA) diff --git a/homeassistant/custom_components/sems/const.py b/homeassistant/custom_components/sems/const.py index 43b4daa..b8a4ad6 100644 --- a/homeassistant/custom_components/sems/const.py +++ b/homeassistant/custom_components/sems/const.py @@ -1,7 +1,6 @@ """Constants for the SEMS integration.""" import voluptuous as vol - from homeassistant.const import CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_USERNAME DOMAIN = "sems" @@ -23,3 +22,24 @@ SEMS_CONFIG_SCHEMA = vol.Schema( ): int, # , default=DEFAULT_SCAN_INTERVAL } ) + +AC_EMPTY = 6553.5 +AC_CURRENT_EMPTY = 6553.5 +AC_FEQ_EMPTY = 655.35 + + +STATUS_LABELS = {-1: "Offline", 0: "Waiting", 1: "Normal", 2: "Fault"} + + +class GOODWE_SPELLING: + """Constants for correcting GoodWe API spelling errors.""" + + battery = "bettery" + batteryStatus = "betteryStatus" + homeKit = "homKit" + temperature = "tempperature" + hasEnergyStatisticsCharts = "hasEnergeStatisticsCharts" + energyStatisticsCharts = "energeStatisticsCharts" + energyStatisticsTotals = "energeStatisticsTotals" + thisMonthTotalE = "thismonthetotle" + lastMonthTotalE = "lastmonthetotle" diff --git a/homeassistant/custom_components/sems/manifest.json b/homeassistant/custom_components/sems/manifest.json index 28f6431..981e282 100644 --- a/homeassistant/custom_components/sems/manifest.json +++ b/homeassistant/custom_components/sems/manifest.json @@ -9,6 +9,7 @@ "documentation": "https://github.com/TimSoethout/goodwe-sems-home-assistant", "iot_class": "cloud_polling", "issue_tracker": "https://github.com/TimSoethout/goodwe-sems-home-assistant/issues", + "loggers": ["custom_components.sems"], "requirements": [], - "version": "7.4.1" + "version": "8.2.0" } \ No newline at end of file diff --git a/homeassistant/custom_components/sems/sems_api.py b/homeassistant/custom_components/sems/sems_api.py index b86b9c8..d0edc6e 100644 --- a/homeassistant/custom_components/sems/sems_api.py +++ b/homeassistant/custom_components/sems/sems_api.py @@ -1,9 +1,12 @@ +from __future__ import annotations + import json import logging +from typing import Any import requests - from homeassistant import exceptions +from homeassistant.core import HomeAssistant _LOGGER = logging.getLogger(__name__) @@ -26,233 +29,280 @@ _DefaultHeaders = { class SemsApi: """Interface to the SEMS API.""" - def __init__(self, hass, username, password): + def __init__(self, hass: HomeAssistant, username: str, password: str) -> None: """Init dummy hub.""" self._hass = hass self._username = username self._password = password - self._token = None + self._token: dict[str, Any] | None = None def test_authentication(self) -> bool: """Test if we can authenticate with the host.""" try: self._token = self.getLoginToken(self._username, self._password) - return self._token is not None except Exception as exception: _LOGGER.exception("SEMS Authentication exception: %s", exception) return False + else: + return self._token is not None - def getLoginToken(self, userName, password): + def _make_http_request( + self, + url: str, + headers: dict[str, str], + data: str | None = None, + json_data: dict[str, Any] | None = None, + operation_name: str = "HTTP request", + validate_code: bool = True, + ) -> dict[str, Any] | None: + """Make a generic HTTP request with error handling and optional code validation.""" + try: + _LOGGER.debug("SEMS - Making %s to %s", operation_name, url) + + response = requests.post( + url, + headers=headers, + data=data, + json=json_data, + timeout=_RequestTimeout, + ) + + _LOGGER.debug("%s Response: %s", operation_name, response) + # _LOGGER.debug("%s Response text: %s", operation_name, response.text) + + response.raise_for_status() + jsonResponse: dict[str, Any] = response.json() + + # Validate response code if requested + if validate_code: + if jsonResponse.get("code") not in (0, "0"): + _LOGGER.error( + "%s failed with code: %s, message: %s", + operation_name, + jsonResponse.get("code"), + jsonResponse.get("msg", "Unknown error"), + ) + return None + + if jsonResponse.get("data") is None: + _LOGGER.error("%s response missing data field", operation_name) + return None + + return jsonResponse + + except (requests.RequestException, ValueError, KeyError) as exception: + _LOGGER.error("Unable to complete %s: %s", operation_name, exception) + raise + + def getLoginToken(self, userName: str, password: str) -> dict[str, Any] | None: """Get the login token for the SEMS API.""" try: - # Get our Authentication Token from SEMS Portal API - _LOGGER.debug("SEMS - Getting API token") - # Prepare Login Data to retrieve Authentication Token # Dict won't work here somehow, so this magic string creation must do. login_data = '{"account":"' + userName + '","pwd":"' + password + '"}' - # login_data = {"account": userName, "pwd": password} - # Make POST request to retrieve Authentication Token from SEMS API - login_response = requests.post( + jsonResponse = self._make_http_request( _LoginURL, - headers=_DefaultHeaders, + _DefaultHeaders, data=login_data, - timeout=_RequestTimeout, + operation_name="login API call", + validate_code=True, ) - _LOGGER.debug("Login Response: %s", login_response) - # _LOGGER.debug("Login Response text: %s", login_response.text) - login_response.raise_for_status() + if jsonResponse is None: + return None - # Process response as JSON - jsonResponse = login_response.json() # json.loads(login_response.text) - # _LOGGER.debug("Login JSON response %s", jsonResponse) # Get all the details from our response, needed to make the next POST request (the one that really fetches the data) # Also store the api url send with the authentication request for later use tokenDict = jsonResponse["data"] + if not isinstance(tokenDict, dict): + _LOGGER.error("Login response data was not a dict") + return None tokenDict["api"] = jsonResponse["api"] _LOGGER.debug("SEMS - API Token received: %s", tokenDict) return tokenDict - except Exception as exception: - _LOGGER.error("Unable to fetch login token from SEMS API. %s", exception) + + except (requests.RequestException, ValueError, KeyError) as exception: + _LOGGER.error("Unable to fetch login token from SEMS API: %s", exception) return None - def getPowerStationIds(self, renewToken=False, maxTokenRetries=2) -> str: + def _make_api_call( + self, + url_part: str, + data: str | None = None, + renewToken: bool = False, + maxTokenRetries: int = 2, + operation_name: str = "API call", + ) -> dict[str, Any] | None: + """Make a generic API call with token management and retry logic.""" + _LOGGER.debug("SEMS - Making %s", operation_name) + if maxTokenRetries <= 0: + _LOGGER.info("SEMS - Maximum token fetch tries reached, aborting for now") + raise OutOfRetries + + if self._token is None or renewToken: + _LOGGER.debug( + "API token not set (%s) or new token requested (%s), fetching", + self._token, + renewToken, + ) + self._token = self.getLoginToken(self._username, self._password) + + if self._token is None: + _LOGGER.error("Failed to obtain API token") + return None + + # Prepare headers + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "token": json.dumps(self._token), + } + + api_url = self._token["api"] + url_part + + try: + jsonResponse: dict[str, Any] | None = self._make_http_request( + api_url, + headers, + data=data, + operation_name=operation_name, + validate_code=True, + ) + + # _make_http_request already validated the response, so if we get here, it's successful + if jsonResponse is None: + # Response validation failed in _make_http_request + _LOGGER.debug( + "%s not successful, retrying with new token, %s retries remaining", + operation_name, + maxTokenRetries, + ) + return self._make_api_call( + url_part, data, True, maxTokenRetries - 1, operation_name + ) + + # Response is valid, return the data + return jsonResponse["data"] + + except (requests.RequestException, ValueError, KeyError) as exception: + _LOGGER.error("Unable to complete %s: %s", operation_name, exception) + return None + + def getPowerStationIds( + self, renewToken: bool = False, maxTokenRetries: int = 2 + ) -> dict[str, Any] | None: """Get the power station ids from the SEMS API.""" - try: - # Get the status of our SEMS Power Station - _LOGGER.debug( - "SEMS - getPowerStationIds Making Power Station Status API Call" - ) - if maxTokenRetries <= 0: - _LOGGER.info( - "SEMS - Maximum token fetch tries reached, aborting for now" - ) - raise OutOfRetries - if self._token is None or renewToken: - _LOGGER.debug( - "API token not set (%s) or new token requested (%s), fetching", - self._token, - renewToken, - ) - self._token = self.getLoginToken(self._username, self._password) + return self._make_api_call( + _GetPowerStationIdByOwnerURLPart, + data=None, + renewToken=renewToken, + maxTokenRetries=maxTokenRetries, + operation_name="getPowerStationIds API call", + ) - # Prepare Power Station status Headers - headers = { - "Content-Type": "application/json", - "Accept": "application/json", - "token": json.dumps(self._token), - } - - getPowerStationIdUrl = self._token["api"] + _GetPowerStationIdByOwnerURLPart - _LOGGER.debug( - "Querying SEMS API (%s) for power station ids by owner", - getPowerStationIdUrl, - ) - - response = requests.post( - getPowerStationIdUrl, - headers=headers, - # data=data, - timeout=_RequestTimeout, - ) - jsonResponse = response.json() - _LOGGER.debug("Response: %s", jsonResponse) - # try again and renew token is unsuccessful - if ( - jsonResponse["msg"] not in ["Successful", "操作成功"] - or jsonResponse["data"] is None - ): - _LOGGER.debug( - "GetPowerStationIdByOwner Query not successful (%s), retrying with new token, %s retries remaining", - jsonResponse["msg"], - maxTokenRetries, - ) - return self.getPowerStationIds( - True, maxTokenRetries=maxTokenRetries - 1 - ) - - return jsonResponse["data"] - except Exception as exception: - _LOGGER.error( - "Unable to fetch power station Ids from SEMS Api. %s", exception - ) - - def getData(self, powerStationId, renewToken=False, maxTokenRetries=2) -> dict: + def getData( + self, powerStationId: str, renewToken: bool = False, maxTokenRetries: int = 2 + ) -> dict[str, Any]: """Get the latest data from the SEMS API and updates the state.""" - try: - # Get the status of our SEMS Power Station - _LOGGER.debug("SEMS - Making Power Station Status API Call") - if maxTokenRetries <= 0: - _LOGGER.info( - "SEMS - Maximum token fetch tries reached, aborting for now" - ) - raise OutOfRetries - if self._token is None or renewToken: - _LOGGER.debug( - "API token not set (%s) or new token requested (%s), fetching", - self._token, - renewToken, - ) - self._token = self.getLoginToken(self._username, self._password) + data = '{"powerStationId":"' + powerStationId + '"}' + result = self._make_api_call( + _PowerStationURLPart, + data=data, + renewToken=renewToken, + maxTokenRetries=maxTokenRetries, + operation_name="getData API call", + ) + return result if isinstance(result, dict) else {} - # Prepare Power Station status Headers - headers = { - "Content-Type": "application/json", - "Accept": "application/json", - "token": json.dumps(self._token), - } + def _make_control_api_call( + self, + data: dict[str, Any], + renewToken: bool = False, + maxTokenRetries: int = 2, + operation_name: str = "Control API call", + ) -> bool: + """Make a control API call with different response handling.""" + _LOGGER.debug("SEMS - Making %s", operation_name) + if maxTokenRetries <= 0: + _LOGGER.info("SEMS - Maximum token fetch tries reached, aborting for now") + raise OutOfRetries - powerStationURL = self._token["api"] + _PowerStationURLPart + if self._token is None or renewToken: _LOGGER.debug( - "Querying SEMS API (%s) for power station id: %s", - powerStationURL, - powerStationId, + "API token not set (%s) or new token requested (%s), fetching", + self._token, + renewToken, ) + self._token = self.getLoginToken(self._username, self._password) - data = '{"powerStationId":"' + powerStationId + '"}' + if self._token is None: + _LOGGER.error("Failed to obtain API token") + return False - response = requests.post( - powerStationURL, headers=headers, data=data, timeout=_RequestTimeout - ) - jsonResponse = response.json() - _LOGGER.debug("Response: %s", jsonResponse) - # try again and renew token is unsuccessful - if ( - jsonResponse["msg"] not in ["success", "操作成功"] - or jsonResponse["data"] is None - ): - _LOGGER.debug( - "Query not successful (%s), retrying with new token, %s retries remaining", - jsonResponse["msg"], - maxTokenRetries, - ) - return self.getData( - powerStationId, True, maxTokenRetries=maxTokenRetries - 1 - ) + # Prepare headers + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "token": json.dumps(self._token), + } - return jsonResponse["data"] - except Exception as exception: - _LOGGER.error("Unable to fetch data from SEMS. %s", exception) - return {} + api_url = self._token["api"] + _PowerControlURLPart - def change_status(self, inverterSn, status, renewToken=False, maxTokenRetries=2): - """Schedule the downtime of the station""" try: - # Get the status of our SEMS Power Station - _LOGGER.debug("SEMS - Making Power Station Status API Call") - if maxTokenRetries <= 0: - _LOGGER.info( - "SEMS - Maximum token fetch tries reached, aborting for now" - ) - raise OutOfRetries - if self._token is None or renewToken: - _LOGGER.debug( - "API token not set (%s) or new token requested (%s), fetching", - self._token, - renewToken, - ) - self._token = self.getLoginToken(self._username, self._password) - - # Prepare Power Station status Headers - headers = { - "Content-Type": "application/json", - "Accept": "application/json", - "token": json.dumps(self._token), - } - - powerControlURL = self._token["api"] + _PowerControlURLPart - # powerControlURL = _PowerControlURL - _LOGGER.debug( - "Sending power control command (%s) for power station id: %s", - powerControlURL, - inverterSn, + # Control API uses different validation (HTTP status code), so don't validate JSON response code + self._make_http_request( + api_url, + headers, + json_data=data, + operation_name=operation_name, + validate_code=False, ) - data = { - "InverterSN": inverterSn, - "InverterStatusSettingMark": "1", - "InverterStatus": str(status), - } + # For control API, any successful HTTP response (status 200) means success + # The _make_http_request already validated HTTP status via raise_for_status() + return True - response = requests.post( - powerControlURL, headers=headers, json=data, timeout=_RequestTimeout - ) - if response.status_code != 200: - # try again and renew token is unsuccessful + except requests.HTTPError as e: + if hasattr(e.response, "status_code") and e.response.status_code != 200: _LOGGER.warning( - "Power control command not successful, retrying with new token, %s retries remaining", + "%s not successful, retrying with new token, %s retries remaining", + operation_name, maxTokenRetries, ) - return self.change_status( - inverterSn, status, True, maxTokenRetries=maxTokenRetries - 1 + return self._make_control_api_call( + data, True, maxTokenRetries - 1, operation_name ) + _LOGGER.error("Unable to execute %s: %s", operation_name, e) + return False + except (requests.RequestException, ValueError, KeyError) as exception: + _LOGGER.error("Unable to execute %s: %s", operation_name, exception) + return False - return - except Exception as exception: - _LOGGER.error("Unable to execute Power control command. %s", exception) + def change_status( + self, + inverterSn: str, + status: str | int, + renewToken: bool = False, + maxTokenRetries: int = 2, + ) -> None: + """Schedule the downtime of the station.""" + data = { + "InverterSN": inverterSn, + "InverterStatusSettingMark": "1", + "InverterStatus": str(status), + } + + success = self._make_control_api_call( + data, + renewToken=renewToken, + maxTokenRetries=maxTokenRetries, + operation_name=f"power control command for inverter {inverterSn}", + ) + + if not success: + _LOGGER.error("Power control command failed after all retries") class OutOfRetries(exceptions.HomeAssistantError): diff --git a/homeassistant/custom_components/sems/sensor.py b/homeassistant/custom_components/sems/sensor.py index 6155f5a..a446a49 100644 --- a/homeassistant/custom_components/sems/sensor.py +++ b/homeassistant/custom_components/sems/sensor.py @@ -4,174 +4,610 @@ For more details about this platform, please refer to the documentation at https://github.com/TimSoethout/goodwe-sems-home-assistant """ -from datetime import timedelta import logging -# from typing import Coroutine +import re +from collections.abc import Callable +from dataclasses import dataclass +from decimal import Decimal +from typing import Any -import homeassistant from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorStateClass, ) -from homeassistant.components.switch import SwitchEntity -from homeassistant.const import CONF_SCAN_INTERVAL, UnitOfEnergy, UnitOfPower, Platform -from homeassistant.core import HomeAssistant -from homeassistant.helpers import entity_registry -from homeassistant.helpers.device_registry import DeviceInfo -from homeassistant.helpers.update_coordinator import ( - CoordinatorEntity, - DataUpdateCoordinator, - UpdateFailed, +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + PERCENTAGE, + Platform, + UnitOfElectricCurrent, + UnitOfElectricPotential, + UnitOfEnergy, + UnitOfFrequency, + UnitOfPower, + UnitOfTemperature, + UnitOfTime, ) +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity -from .const import CONF_STATION_ID, DEFAULT_SCAN_INTERVAL, DOMAIN +from . import SemsConfigEntry, SemsCoordinator, SemsData +from .const import ( + AC_CURRENT_EMPTY, + AC_EMPTY, + AC_FEQ_EMPTY, + DOMAIN, + GOODWE_SPELLING, + STATUS_LABELS, +) +from .device import device_info_for_inverter _LOGGER = logging.getLogger(__name__) +type SemsValuePath = list[str | int] -async def async_setup_entry(hass, config_entry, async_add_entities): + +def convert_status_to_label(status: Any) -> str: + """Convert numeric status code to human-readable label.""" + return STATUS_LABELS.get(int(status), "Unknown") + + +@dataclass(slots=True) +class SemsSensorType: + """SEMS sensor definition.""" + + device_info: DeviceInfo + unique_id: str + value_path: SemsValuePath + name: str | None = None # Name is None when it is determined by device class / UOM. + device_class: SensorDeviceClass | None = None + native_unit_of_measurement: str | None = None + state_class: SensorStateClass | None = None + empty_value: Any = None + data_type_converter: Callable = Decimal + custom_value_handler: Callable[[Any, dict[str, Any]], Any] | None = None + + +@dataclass(slots=True) +class SemsHomekitSensorType(SemsSensorType): + """SEMS HomeKit/powerflow sensor definition.""" + + +@dataclass(slots=True) +class SemsInverterSensorType(SemsSensorType): + """SEMS inverter sensor definition.""" + + +def get_homekit_sn(homekit_data: dict[str, Any] | None) -> str | None: + """Return the HomeKit serial number from coordinator data, if available.""" + + if homekit_data is None: + return None + value = homekit_data.get("sn") + return value if isinstance(value, str) else None + + +def get_has_existing_homekit_entity( + homekit_data: dict[str, Any] | None, hass: HomeAssistant, config_entry: ConfigEntry +) -> bool: + """Return whether a HomeKit entity already exists for this config entry.""" + + home_kit_sn = get_homekit_sn(homekit_data) + if home_kit_sn is not None: + ent_reg = er.async_get(hass) + entities = er.async_entries_for_config_entry(ent_reg, config_entry.entry_id) + for entity in entities: + if entity.unique_id == home_kit_sn: + return True + return False + + +def sensor_options_for_data( + data: SemsData, has_existing_homekit_entity: bool = False +) -> list[SemsSensorType]: + """Build a list of sensor definitions for the given coordinator data.""" + + sensors: list[SemsSensorType] = [] + currency = data.currency + _LOGGER.debug("Detected currency: %s", currency) + + for serial_number, inverter_data in data.inverters.items(): + # serial_number = inverter["sn"] + path_to_inverter: SemsValuePath = [serial_number] + # device_data = get_value_from_path(data, path_to_inverter) + + device_info = device_info_for_inverter(serial_number, inverter_data) + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-status", + [*path_to_inverter, "status"], + "Status", + data_type_converter=convert_status_to_label, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-capacity", + [*path_to_inverter, "capacity"], + "Capacity", + SensorDeviceClass.POWER, + UnitOfPower.KILO_WATT, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-power", + # "Power", + [*path_to_inverter, "pac"], + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.WATT, + state_class=SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-energy", + [*path_to_inverter, "etotal"], + device_class=SensorDeviceClass.ENERGY, + native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, + state_class=SensorStateClass.TOTAL_INCREASING, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-hour-total", + [*path_to_inverter, "hour_total"], + "Total Hours", + native_unit_of_measurement=UnitOfTime.HOURS, + state_class=SensorStateClass.TOTAL_INCREASING, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-temperature", + [*path_to_inverter, GOODWE_SPELLING.temperature], + device_class=SensorDeviceClass.TEMPERATURE, + native_unit_of_measurement=UnitOfTemperature.CELSIUS, + state_class=SensorStateClass.MEASUREMENT, + empty_value=0, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-eday", + [*path_to_inverter, "eday"], + "Energy Today", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{GOODWE_SPELLING.thisMonthTotalE}", + [*path_to_inverter, GOODWE_SPELLING.thisMonthTotalE], + "Energy This Month", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{GOODWE_SPELLING.lastMonthTotalE}", + [*path_to_inverter, GOODWE_SPELLING.lastMonthTotalE], + "Energy Last Month", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-iday", + [*path_to_inverter, "iday"], + "Income Today", + SensorDeviceClass.MONETARY, + currency, + SensorStateClass.TOTAL, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-itotal", + [*path_to_inverter, "itotal"], + "Income Total", + SensorDeviceClass.MONETARY, + currency, + SensorStateClass.TOTAL, + ), + ] + # Multiple strings + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-vpv{idx}", + [*path_to_inverter, f"vpv{idx}"], + f"PV String {idx} Voltage", + SensorDeviceClass.VOLTAGE, + UnitOfElectricPotential.VOLT, + SensorStateClass.MEASUREMENT, + 0, + ) + for idx in range(1, 5) + if get_value_from_path(data.inverters, [*path_to_inverter, f"vpv{idx}"]) + is not None + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-ipv{idx}", + [*path_to_inverter, f"ipv{idx}"], + f"PV String {idx} Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + 0, + ) + for idx in range(1, 5) + if get_value_from_path(data.inverters, [*path_to_inverter, f"ipv{idx}"]) + is not None + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-vac{idx}", + [*path_to_inverter, f"vac{idx}"], + f"Grid {idx} AC Voltage", + SensorDeviceClass.VOLTAGE, + UnitOfElectricPotential.VOLT, + SensorStateClass.MEASUREMENT, + AC_EMPTY, + ) + for idx in range(1, 4) + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-iac{idx}", + [*path_to_inverter, f"iac{idx}"], + f"Grid {idx} AC Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + AC_CURRENT_EMPTY, + ) + for idx in range(1, 4) + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-fac{idx}", + [*path_to_inverter, f"fac{idx}"], + f"Grid {idx} AC Frequency", + SensorDeviceClass.FREQUENCY, + UnitOfFrequency.HERTZ, + SensorStateClass.MEASUREMENT, + AC_FEQ_EMPTY, + ) + for idx in range(1, 4) + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-vbattery1", + [*path_to_inverter, "vbattery1"], + "Battery Voltage", + SensorDeviceClass.VOLTAGE, + UnitOfElectricPotential.VOLT, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-ibattery1", + [*path_to_inverter, "ibattery1"], + "Battery Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + ), + ] + battery_count = get_value_from_path( + data.inverters, [*path_to_inverter, "battery_count"] + ) + if isinstance(battery_count, int): + for idx in range(battery_count): + path_to_battery: SemsValuePath = [ + *path_to_inverter, + "more_batterys", + idx, + ] + sensors += [ + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-pbattery", + [*path_to_battery, "pbattery"], + f"Battery {idx} Power", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-vbattery", + [*path_to_battery, "vbattery"], + f"Battery {idx} Voltage", + SensorDeviceClass.VOLTAGE, + UnitOfElectricPotential.VOLT, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-ibattery", + [*path_to_battery, "ibattery"], + f"Battery {idx} Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-soc", + [*path_to_battery, "soc"], + f"Battery {idx} State of Charge", + SensorDeviceClass.BATTERY, + PERCENTAGE, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-soh", + [*path_to_battery, "soh"], + f"Battery {idx} State of Health", + SensorDeviceClass.BATTERY, + PERCENTAGE, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-bms_temperature", + [*path_to_battery, "bms_temperature"], + f"Battery {idx} BMS Temperature", + SensorDeviceClass.TEMPERATURE, + UnitOfTemperature.CELSIUS, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-bms_discharge_i_max", + [*path_to_battery, "bms_discharge_i_max"], + f"Battery {idx} BMS Discharge Max Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + ), + SemsInverterSensorType( + device_info, + f"{serial_number}-{idx}-bms_charge_i_max", + [*path_to_battery, "bms_charge_i_max"], + f"Battery {idx} BMS Charge Max Current", + SensorDeviceClass.CURRENT, + UnitOfElectricCurrent.AMPERE, + SensorStateClass.MEASUREMENT, + ), + ] + _LOGGER.debug("Sensors for inverter %s: %s", serial_number, sensors) + + # HomeKit powerflow + SEMS charts live in `SemsData.homekit`. + if data.homekit is not None: + inverter_serial_number = get_homekit_sn(data.homekit) + if not has_existing_homekit_entity or inverter_serial_number is None: + inverter_serial_number = "powerflow" + serial_backwards_compatibility = ( + "homeKit" # the old code uses homeKit for the serial number + ) + device_info = DeviceInfo( + identifiers={ + # Serial numbers are unique identifiers within a specific domain + (DOMAIN, serial_backwards_compatibility) + }, + name="HomeKit", + manufacturer="GoodWe", + ) + + def status_value_handler( + status_path: SemsValuePath, + ) -> Callable[[Any, dict[str, Any]], Any]: + """Return a handler that applies a sign depending on grid status.""" + + def value_status_handler(value: Any, data: dict[str, Any]) -> Any: + """Apply the grid status sign to the given value.""" + if value is None: + return None + grid_status = get_value_from_path(data, status_path) + if grid_status is None: + return value + try: + return Decimal(str(value)) * int(grid_status) + except (TypeError, ValueError): + return value + + return value_status_handler + + sensors += [ + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}", # backwards compatibility otherwise would be f"{serial_number}-load" + ["powerflow", "load"], + "HomeKit Load", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + custom_value_handler=status_value_handler(["powerflow", "loadStatus"]), + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-pv", + ["powerflow", "pv"], + "HomeKit PV", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-grid", + ["powerflow", "grid"], + "HomeKit Grid", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-load-status", + ["powerflow", "loadStatus"], + "HomeKit Load Status", + None, + None, + SensorStateClass.MEASUREMENT, + # Note: for the dedicated "load-status" sensor we intentionally use + # gridStatus here instead of loadStatus. The "HomeKit Load" power + # sensor above uses loadStatus to determine the sign of the load + # power value itself, while this sensor exposes the load state using + # the same import/export (sign) convention as the grid power sensor. + custom_value_handler=status_value_handler(["powerflow", "gridStatus"]), + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-battery", + ["powerflow", GOODWE_SPELLING.battery], + "HomeKit Battery", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + custom_value_handler=status_value_handler( + ["powerflow", GOODWE_SPELLING.batteryStatus] + ), + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-genset", + ["powerflow", "genset"], + "HomeKit generator", + SensorDeviceClass.POWER, + UnitOfPower.WATT, + SensorStateClass.MEASUREMENT, + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-soc", + ["powerflow", "soc"], + "HomeKit State of Charge", + SensorDeviceClass.BATTERY, + PERCENTAGE, + SensorStateClass.MEASUREMENT, + ), + ] + if data.homekit.get(GOODWE_SPELLING.hasEnergyStatisticsCharts): + if data.homekit.get(GOODWE_SPELLING.energyStatisticsCharts): + sensors += [ + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-import-energy", + [GOODWE_SPELLING.energyStatisticsCharts, "buy"], + "SEMS Import", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-export-energy", + [GOODWE_SPELLING.energyStatisticsCharts, "sell"], + "SEMS Export", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + ] + if data.homekit.get(GOODWE_SPELLING.energyStatisticsTotals): + sensors += [ + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-import-energy-total", + [GOODWE_SPELLING.energyStatisticsTotals, "buy"], + "SEMS Total Import", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + SemsHomekitSensorType( + device_info, + f"{inverter_serial_number}-export-energy-total", + [GOODWE_SPELLING.energyStatisticsTotals, "sell"], + "SEMS Total Export", + SensorDeviceClass.ENERGY, + UnitOfEnergy.KILO_WATT_HOUR, + SensorStateClass.TOTAL_INCREASING, + ), + ] + return sensors + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: SemsConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Add sensors for passed config_entry in HA.""" - # _LOGGER.debug("hass.data[DOMAIN] %s", hass.data[DOMAIN]) - # semsApi = hass.data[DOMAIN][config_entry.entry_id] - # stationId = config_entry.data[CONF_STATION_ID] - - # _LOGGER.debug("config_entry %s", config_entry.data) - # update_interval = timedelta( - # seconds=config_entry.data.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) - # ) - - # async def async_update_data(): - # """Fetch data from API endpoint. - - # This is the place to pre-process the data to lookup tables - # so entities can quickly look up their data. - # """ - # try: - # # Note: asyncio.TimeoutError and aiohttp.ClientError are already - # # handled by the data update coordinator. - # # async with async_timeout.timeout(10): - # result = await hass.async_add_executor_job(semsApi.getData, stationId) - # _LOGGER.debug("Resulting result: %s", result) - - # inverters = result["inverter"] - - # # found = [] - # # _LOGGER.debug("Found inverters: %s", inverters) - # data = {} - # if inverters is None: - # # something went wrong, probably token could not be fetched - # raise UpdateFailed( - # "Error communicating with API, probably token could not be fetched, see debug logs" - # ) - # for inverter in inverters: - # name = inverter["invert_full"]["name"] - # # powerstation_id = inverter["invert_full"]["powerstation_id"] - # sn = inverter["invert_full"]["sn"] - # _LOGGER.debug("Found inverter attribute %s %s", name, sn) - # data[sn] = inverter["invert_full"] - - # hasPowerflow = result["hasPowerflow"] - # hasEnergeStatisticsCharts = result["hasEnergeStatisticsCharts"] - - # if hasPowerflow: - # if hasEnergeStatisticsCharts: - # StatisticsCharts = { - # f"Charts_{key}": val - # for key, val in result["energeStatisticsCharts"].items() - # } - # StatisticsTotals = { - # f"Totals_{key}": val - # for key, val in result["energeStatisticsTotals"].items() - # } - # powerflow = { - # **result["powerflow"], - # **StatisticsCharts, - # **StatisticsTotals, - # } - # else: - # powerflow = result["powerflow"] - - # powerflow["sn"] = result["homKit"]["sn"] - - # # Goodwe 'Power Meter' (not HomeKit) doesn't have a sn - # # Let's put something in, otherwise we can't see the data. - # if powerflow["sn"] is None: - # powerflow["sn"] = "GW-HOMEKIT-NO-SERIAL" - - # #_LOGGER.debug("homeKit sn: %s", result["homKit"]["sn"]) - # # This seems more accurate than the Chart_sum - # powerflow["all_time_generation"] = result["kpi"]["total_power"] - - # data["homeKit"] = powerflow - - # # _LOGGER.debug("Resulting data: %s", data) - # return data - # # except ApiError as err: - # except Exception as err: - # # logging.exception("Something awful happened!") - # raise UpdateFailed(f"Error communicating with API: {err}") from err - - # coordinator = DataUpdateCoordinator( - # hass, - # _LOGGER, - # # Name of the data. For logging purposes. - # name="SEMS API", - # update_method=async_update_data, - # # Polling interval. Will only be polled if there are subscribers. - # update_interval=update_interval, - # ) - coordinator = hass.data[DOMAIN][config_entry.entry_id] - - # - # Fetch initial data so we have data when entities subscribe - # - # If the refresh fails, async_config_entry_first_refresh will - # raise ConfigEntryNotReady and setup will try again later - # - # If you do not want to retry setup on failure, use - # coordinator.async_refresh() instead - # - await coordinator.async_config_entry_first_refresh() + coordinator = config_entry.runtime_data.coordinator # _LOGGER.debug("Initial coordinator data: %s", coordinator.data) - for _idx, ent in enumerate(coordinator.data): + # Backwards compatibility note: keep IDs stable for existing entity registry entries. + for _idx, ent in enumerate(coordinator.data.inverters): _migrate_to_new_unique_id(hass, ent) - async_add_entities( - SemsSensor(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - # Don't make SemsSensor for homeKit, since it is not an inverter; unsure how this could work before... - if ent != "homeKit" + has_existing_homekit_entity = get_has_existing_homekit_entity( + coordinator.data.homekit, hass, config_entry ) - async_add_entities( - SemsStatisticsSensor(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - # Don't make SemsStatisticsSensor for homeKit, since it is not an inverter; unsure how this could work before... - if ent != "homeKit" - ) - async_add_entities( - SemsPowerflowSensor(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - if ent == "homeKit" - ) - async_add_entities( - SemsTotalImportSensor(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - if ent == "homeKit" - ) - async_add_entities( - SemsTotalExportSensor(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - if ent == "homeKit" + + sensor_options: list[SemsSensorType] = sensor_options_for_data( + coordinator.data, has_existing_homekit_entity ) + sensors = [ + ( + SemsHomekitSensor + if isinstance(sensor_option, SemsHomekitSensorType) + else SemsInverterSensor + )( + coordinator, + sensor_option.device_info, + sensor_option.unique_id, + sensor_option.name, + sensor_option.value_path, + sensor_option.data_type_converter, + sensor_option.device_class, + sensor_option.native_unit_of_measurement, + sensor_option.state_class, + sensor_option.empty_value, + sensor_option.custom_value_handler, + ) + for sensor_option in sensor_options + ] + async_add_entities(sensors) + + # async_add_entities( + # SemsSensor(coordinator, ent) + # for idx, ent in enumerate(coordinator.data) + # # Don't make SemsSensor for homeKit, since it is not an inverter; unsure how this could work before... + # if ent != "homeKit" + # ) + # async_add_entities( + # SemsStatisticsSensor(coordinator, ent) + # for idx, ent in enumerate(coordinator.data) # Migrate old power sensor unique ids to new unique ids (with `-power`) def _migrate_to_new_unique_id(hass: HomeAssistant, sn: str) -> None: """Migrate old unique ids to new unique ids.""" - ent_reg = entity_registry.async_get(hass) + ent_reg = er.async_get(hass) old_unique_id = sn new_unique_id = f"{old_unique_id}-power" @@ -195,434 +631,172 @@ def _migrate_to_new_unique_id(hass: HomeAssistant, sn: str) -> None: ) -class SemsSensor(CoordinatorEntity, SensorEntity): - """SemsSensor using CoordinatorEntity. +def get_value_from_path(data: dict[str, Any], path: SemsValuePath) -> Any: + """Return the value at a nested path in a dict, or `None` if missing.""" - The CoordinatorEntity class provides: - should_poll - async_update - async_added_to_hass - available - """ - - # Sensor has name determined by device class (e.g. Inverter 123456 Power) - _attr_has_entity_name = True - - def __init__(self, coordinator, sn) -> None: - """Pass coordinator to CoordinatorEntity.""" - super().__init__(coordinator) - self.coordinator = coordinator - self.sn = sn - self._attr_unique_id = f"{self.coordinator.data[self.sn]['sn']}-power" - _LOGGER.debug("Creating SemsSensor with id %s", self.sn) - self._attr_unique_id = f"{self.coordinator.data[self.sn]['sn']}-power" - _LOGGER.debug( - "Creating SemsSensor with id %s and data %s", - self.sn, - self.coordinator.data[self.sn], - ) - - _attr_device_class = SensorDeviceClass.POWER - _attr_native_unit_of_measurement = UnitOfPower.WATT - _attr_should_poll = False - - @property - def native_value(self): - """Return the value reported by the sensor.""" - data = self.coordinator.data[self.sn] - return data["pac"] if data["status"] == 1 else 0 - - def _statusText(self, status) -> str: - labels = {-1: "Offline", 0: "Waiting", 1: "Normal", 2: "Fault"} - return labels.get(status, "Unknown") - - # For backwards compatibility - @property - def extra_state_attributes(self): - """Return the state attributes of the monitored installation.""" - data = self.coordinator.data[self.sn] - attributes = {k: v for k, v in data.items() if k is not None and v is not None} - attributes["statusText"] = self._statusText(data["status"]) - return attributes - - @property - def is_on(self) -> bool: - """Return entity status.""" - return self.coordinator.data[self.sn]["status"] == 1 - - @property - def available(self): - """Return if entity is available.""" - return self.coordinator.last_update_success - - @property - def device_info(self) -> DeviceInfo: - """Return device information.""" - return DeviceInfo( - identifiers={(DOMAIN, self.sn)}, - name=f"Inverter {self.coordinator.data[self.sn]['name']}", - manufacturer="GoodWe", - model=self.extra_state_attributes.get("model_type", "unknown"), - sw_version=self.extra_state_attributes.get("firmwareversion", "unknown"), - configuration_url=f"https://semsportal.com/PowerStation/PowerStatusSnMin/{self.coordinator.data[self.sn]['powerstation_id']}", - ) - - async def async_added_to_hass(self): - """When entity is added to hass.""" - self.async_on_remove( - self.coordinator.async_add_listener(self.async_write_ha_state) - ) - - async def async_update(self): - """Update the entity. - - Only used by the generic entity update service. - """ - await self.coordinator.async_request_refresh() + value: Any = data + try: + for key in path: + value = value[key] + except (KeyError, TypeError): + return None + return value -class SemsStatisticsSensor(CoordinatorEntity, SensorEntity): - """Sensor in kWh to enable HA statistics, in the end usable in the power component.""" +class SemsSensor(CoordinatorEntity[SemsCoordinator], SensorEntity): + """Representation of a GoodWe SEMS sensor backed by the shared coordinator.""" + + str_clean_regex = re.compile(r"(\d+\.?\d*)") _attr_has_entity_name = True - def __init__(self, coordinator, sn) -> None: - """Pass coordinator to CoordinatorEntity.""" + def __init__( + self, + coordinator: SemsCoordinator, + device_info: DeviceInfo, + unique_id: str, + name: str | None, + value_path: SemsValuePath, + data_type_converter: Callable, + device_class: SensorDeviceClass | None = None, + native_unit_of_measurement: str | None = None, + state_class: SensorStateClass | None = None, + empty_value=None, + custom_value_handler=None, + ) -> None: + """Initialize a SEMS sensor.""" + super().__init__(coordinator) - self.coordinator = coordinator - self.sn = sn - _LOGGER.debug("Creating SemsStatisticsSensor with id %s", self.sn) + self._value_path = value_path + self._data_type_converter = data_type_converter + self._empty_value = empty_value + + self._attr_unique_id = unique_id + self._attr_device_info = device_info + self._attr_device_class = device_class + self._attr_native_unit_of_measurement = native_unit_of_measurement + self._attr_state_class = state_class + + # When `name` is None, Home Assistant determines the name from + # device class / unit (using has_entity_name). + if name is not None: + self._attr_name = name + + self._custom_value_handler = custom_value_handler + + raw_value = self._get_native_value_from_coordinator() + + # Disable-by-default must be decided before registry entry is created. + if raw_value is None or ( + self._empty_value is not None and raw_value == self._empty_value + ): + _LOGGER.debug( + "Disabling SemsSensor `%s` by default since initial value is None or empty (`%s`)", + unique_id, + raw_value, + ) + self._attr_entity_registry_enabled_default = False + _LOGGER.debug( - "Creating SemsSensor with id %s and data %s", - self.sn, - self.coordinator.data[self.sn], + "Created SemsSensor with id `%s`, `%s`, value path `%s`", # , data `%s`", + unique_id, + name, + value_path, ) - @property - def device_class(self): - return SensorDeviceClass.ENERGY + def _get_native_value_from_coordinator(self) -> Any: + """Get the raw value from coordinator data.""" + + data = self._get_data_dict() + if data is None: + return None + return get_value_from_path(data, self._value_path) + + def _get_data_dict(self) -> dict[str, Any] | None: + """Return the dict to read values from.""" + + return self.coordinator.data.inverters @property - def unit_of_measurement(self): - return UnitOfEnergy.KILO_WATT_HOUR + def native_value(self) -> Any: + """Return the current value.""" + + value = self._get_native_value_from_coordinator() + + if isinstance(value, str): + if match := self.str_clean_regex.search(value): + value = match.group(1) + + if value is None: + return None + if self._empty_value is not None and value == self._empty_value: + return None + + if self._custom_value_handler is not None: + data = self._get_data_dict() + if data is None: + return None + return self._custom_value_handler(value, data) + + try: + return self._data_type_converter(value) + except (TypeError, ValueError): + return value # @property - # def name(self) -> str: - # """Return the name of the sensor.""" - # return f"Inverter {self.coordinator.data[self.sn]['name']} Energy" + # def suggested_display_precision(self): + # """Return the suggested number of decimal digits for display.""" + # return 2 + + +class SemsInverterSensor(SemsSensor): + """Sensor that reads from inverter data.""" + + def _get_data_dict(self) -> dict[str, Any] | None: + """Return inverter dict.""" + + return self.coordinator.data.inverters @property - def unique_id(self) -> str: - return f"{self.coordinator.data[self.sn]['sn']}-energy" + def extra_state_attributes(self) -> dict[str, Any] | None: + """Return inverter attributes for backwards compatibility.""" - @property - def state(self): - """Return the state of the device.""" - _LOGGER.debug( - "SemsStatisticsSensor state, coordinator data: %s", self.coordinator.data - ) - _LOGGER.debug("SemsStatisticsSensor self.sn: %s", self.sn) - _LOGGER.debug( - "SemsStatisticsSensor state, self data: %s", self.coordinator.data[self.sn] - ) - data = self.coordinator.data[self.sn] - return data["etotal"] + if not (unique_id := self._attr_unique_id) or not unique_id.endswith("-power"): + return None - @property - def should_poll(self) -> bool: - """No need to poll. Coordinator notifies entity of updates.""" - return False + if not self._value_path: + return None - @property - def device_info(self): - # _LOGGER.debug("self.device_state_attributes: %s", self.device_state_attributes) - data = self.coordinator.data[self.sn] - return { - "identifiers": { - # Serial numbers are unique identifiers within a specific domain - (DOMAIN, self.sn) - }, - # "name": self.name, - "manufacturer": "GoodWe", - "model": data.get("model_type", "unknown"), - "sw_version": data.get("firmwareversion", "unknown"), - # "via_device": (DOMAIN, self.api.bridgeid), + inverter_sn = self._value_path[0] + if not isinstance(inverter_sn, str): + return None + + inverter_data = self.coordinator.data.inverters.get(inverter_sn) + if inverter_data is None: + return None + + attributes = { + key: value + for key, value in inverter_data.items() + if key is not None and value is not None } - @property - def state_class(self): - """used by Metered entities / Long Term Statistics""" - return SensorStateClass.TOTAL_INCREASING - - async def async_added_to_hass(self): - """When entity is added to hass.""" - self.async_on_remove( - self.coordinator.async_add_listener(self.async_write_ha_state) - ) - - async def async_update(self): - """Update the entity. - - Only used by the generic entity update service. - """ - await self.coordinator.async_request_refresh() - - -class SemsTotalImportSensor(CoordinatorEntity, SensorEntity): - """Sensor in kWh to enable HA statistics, in the end usable in the power component.""" - - _attr_has_entity_name = True - - def __init__(self, coordinator, sn): - """Pass coordinator to CoordinatorEntity.""" - super().__init__(coordinator) - self.coordinator = coordinator - self.sn = sn - _LOGGER.debug("Creating SemsStatisticsSensor with id %s", self.sn) - - @property - def device_class(self): - return SensorDeviceClass.ENERGY - - @property - def unit_of_measurement(self): - return UnitOfEnergy.KILO_WATT_HOUR - - @property - def name(self) -> str: - """Return the name of the sensor.""" - return "HomeKit Import" - - @property - def unique_id(self) -> str: - return f"{self.coordinator.data[self.sn]['sn']}-import-energy" - - @property - def state(self): - """Return the state of the device.""" - data = self.coordinator.data[self.sn] - return data["Charts_buy"] - - def statusText(self, status) -> str: - labels = {-1: "Offline", 0: "Waiting", 1: "Normal", 2: "Fault"} - return labels[status] if status in labels else "Unknown" - - @property - def should_poll(self) -> bool: - """No need to poll. Coordinator notifies entity of updates.""" - return False - - @property - def device_info(self): - return { - "identifiers": { - # Serial numbers are unique identifiers within a specific domain - (DOMAIN, self.sn) - }, - "name": "Homekit", - "manufacturer": "GoodWe", - } - - @property - def state_class(self): - """used by Metered entities / Long Term Statistics""" - return SensorStateClass.TOTAL_INCREASING - - async def async_added_to_hass(self): - """When entity is added to hass.""" - self.async_on_remove( - self.coordinator.async_add_listener(self.async_write_ha_state) - ) - - async def async_update(self): - """Update the entity. - - Only used by the generic entity update service. - """ - await self.coordinator.async_request_refresh() - - -class SemsTotalExportSensor(CoordinatorEntity, SensorEntity): - """Sensor in kWh to enable HA statistics, in the end usable in the power component.""" - - _attr_has_entity_name = True - - def __init__(self, coordinator, sn): - """Pass coordinator to CoordinatorEntity.""" - super().__init__(coordinator) - self.coordinator = coordinator - self.sn = sn - _LOGGER.debug("Creating SemsStatisticsSensor with id %s", self.sn) - - @property - def device_class(self): - return SensorDeviceClass.ENERGY - - @property - def unit_of_measurement(self): - return UnitOfEnergy.KILO_WATT_HOUR - - @property - def name(self) -> str: - """Return the name of the sensor.""" - return "HomeKit Export" - - @property - def unique_id(self) -> str: - return f"{self.coordinator.data[self.sn]['sn']}-export-energy" - - @property - def state(self): - """Return the state of the device.""" - data = self.coordinator.data[self.sn] - return data["Charts_sell"] - - def statusText(self, status) -> str: - labels = {-1: "Offline", 0: "Waiting", 1: "Normal", 2: "Fault"} - return labels[status] if status in labels else "Unknown" - - @property - def should_poll(self) -> bool: - """No need to poll. Coordinator notifies entity of updates.""" - return False - - @property - def device_info(self): - return { - "identifiers": { - # Serial numbers are unique identifiers within a specific domain - (DOMAIN, self.sn) - }, - "name": "Homekit", - "manufacturer": "GoodWe", - } - - @property - def state_class(self): - """used by Metered entities / Long Term Statistics""" - return SensorStateClass.TOTAL_INCREASING - - async def async_added_to_hass(self): - """When entity is added to hass.""" - self.async_on_remove( - self.coordinator.async_add_listener(self.async_write_ha_state) - ) - - async def async_update(self): - """Update the entity. - - Only used by the generic entity update service. - """ - await self.coordinator.async_request_refresh() - - -class SemsPowerflowSensor(CoordinatorEntity, SensorEntity): - """SemsPowerflowSensor using CoordinatorEntity. - - The CoordinatorEntity class provides: - should_poll - async_update - async_added_to_hass - available - """ - - _attr_has_entity_name = True - - def __init__(self, coordinator, sn): - """Pass coordinator to CoordinatorEntity.""" - super().__init__(coordinator) - self.coordinator = coordinator - self.sn = sn - - @property - def device_class(self): - return SensorDeviceClass.POWER_FACTOR - - @property - def unit_of_measurement(self): - return UnitOfPower.WATT - - @property - def name(self) -> str: - """Return the name of the sensor.""" - return f"HomeKit {self.coordinator.data[self.sn]['sn']}" - - @property - def unique_id(self) -> str: - return f"{self.coordinator.data[self.sn]['sn']}-homekit" - - @property - def state(self): - """Return the state of the device.""" - data = self.coordinator.data[self.sn] - load = data["load"] - - if load: - load = load.replace("(W)", "") - - return load if data["gridStatus"] == 1 else 0 - - def statusText(self, status) -> str: - labels = {-1: "Offline", 0: "Waiting", 1: "Normal", 2: "Fault"} - return labels[status] if status in labels else "Unknown" - - # For backwards compatibility - @property - def extra_state_attributes(self): - """Return the state attributes of the monitored installation.""" - data = self.coordinator.data[self.sn] - - attributes = {k: v for k, v in data.items() if k is not None and v is not None} - - attributes["pv"] = data["pv"].replace("(W)", "") - attributes["bettery"] = data["bettery"].replace("(W)", "") - attributes["load"] = data["load"].replace("(W)", "") - attributes["grid"] = data["grid"].replace("(W)", "") - - attributes["statusText"] = self.statusText(data["gridStatus"]) - - if data["loadStatus"] == -1: - attributes["PowerFlowDirection"] = "Export %s" % data["grid"] - if data["loadStatus"] == 1: - attributes["PowerFlowDirection"] = "Import %s" % data["grid"] + status = inverter_data.get("status") + if status is None: + attributes["statusText"] = "Unknown" + else: + try: + attributes["statusText"] = STATUS_LABELS.get(int(status), "Unknown") + except (TypeError, ValueError): + attributes["statusText"] = "Unknown" return attributes - @property - def is_on(self) -> bool: - """Return entity status.""" - self.coordinator.data[self.sn]["gridStatus"] == 1 - @property - def should_poll(self) -> bool: - """No need to poll. Coordinator notifies entity of updates.""" - return False +class SemsHomekitSensor(SemsSensor): + """Sensor that reads from HomeKit/powerflow data.""" - @property - def available(self): - """Return if entity is available.""" - return self.coordinator.last_update_success + def _get_data_dict(self) -> dict[str, Any] | None: + """Return HomeKit dict.""" - @property - def device_info(self): - return { - "identifiers": { - # Serial numbers are unique identifiers within a specific domain - (DOMAIN, self.sn) - }, - "name": "Homekit", - "manufacturer": "GoodWe", - } - - async def async_added_to_hass(self): - """When entity is added to hass.""" - self.async_on_remove( - self.coordinator.async_add_listener(self.async_write_ha_state) - ) - - async def async_update(self): - """Update the entity. - - Only used by the generic entity update service. - """ - await self.coordinator.async_request_refresh() + return self.coordinator.data.homekit diff --git a/homeassistant/custom_components/sems/switch.py b/homeassistant/custom_components/sems/switch.py index 3177549..85a7291 100644 --- a/homeassistant/custom_components/sems/switch.py +++ b/homeassistant/custom_components/sems/switch.py @@ -1,52 +1,51 @@ -"""Support for switch controlling an output of a GoodWe SEMS inverter. +"""Support for inverter control switches from the GoodWe SEMS API. For more details about this platform, please refer to the documentation at https://github.com/TimSoethout/goodwe-sems-home-assistant """ +from __future__ import annotations + import logging +from typing import Any from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity -from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity -from .const import DOMAIN +from . import SemsCoordinator +from .device import device_info_for_inverter _LOGGER = logging.getLogger(__name__) - -from homeassistant.core import HomeAssistant +_INVERTER_STATUS_ON = 1 +_COMMAND_TURN_OFF = 2 +_COMMAND_TURN_ON = 4 -async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entities): - """Add switches for passed config_entry in HA.""" - coordinator = hass.data[DOMAIN][config_entry.entry_id] - # stationId = config_entry.data[CONF_STATION_ID] +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up SEMS switches from a config entry.""" + coordinator = config_entry.runtime_data.coordinator - # coordinator.data should contain a dictionary of inverters, with as data `invert_full` - # Don't make switches for homekit, since it is not an inverter async_add_entities( - SemsStatusSwitch(coordinator, ent) - for idx, ent in enumerate(coordinator.data) - if ent != "homeKit" + SemsStatusSwitch(coordinator, sn) for sn in coordinator.data.inverters ) -class SemsStatusSwitch(CoordinatorEntity, SwitchEntity): - """SemsStatusSwitch using CoordinatorEntity. - - The CoordinatorEntity class provides: - should_poll - async_update - async_added_to_hass - available - """ +class SemsStatusSwitch(CoordinatorEntity[SemsCoordinator], SwitchEntity): + """Switch to control inverter status, backed by the SEMS coordinator.""" # Sensor has device name (e.g. Inverter 123456 Power) _attr_has_entity_name = True - # _attr_name = None + _attr_device_class = SwitchDeviceClass.SWITCH - def __init__(self, coordinator, sn) -> None: + def __init__(self, coordinator: SemsCoordinator, sn: str) -> None: """Initialize the SemsStatusSwitch. Args: @@ -54,40 +53,38 @@ class SemsStatusSwitch(CoordinatorEntity, SwitchEntity): sn: The serial number of the inverter. """ - _LOGGER.debug("Try create SemsStatusSwitch for Inverter %s", sn) - super().__init__(coordinator, context=sn) - self.coordinator = coordinator - self.sn = sn - self._attr_device_info = DeviceInfo( - identifiers={ - # Serial numbers are unique identifiers within a specific domain - (DOMAIN, self.sn) - }, - # Commented out for now, since not all inverter entries have a name; could be related to creating too much switch devices, also for non-inverters such as homekit. - # name=f"Inverter {self.coordinator.data[self.sn]['name']}", - ) - self._attr_unique_id = f"{self.sn}-switch" + _LOGGER.debug("Try create SemsStatusSwitch for inverter %s", sn) + super().__init__(coordinator) + self._sn = sn + inverter_data = coordinator.data.inverters.get(sn, {}) + self._attr_device_info = device_info_for_inverter(sn, inverter_data) + self._attr_unique_id = f"{self._sn}-switch" # somehow needed, no default naming self._attr_name = "Switch" - self._attr_device_class = SwitchDeviceClass.OUTLET - _LOGGER.debug("Creating SemsStatusSwitch for Inverter %s", self.sn) + _LOGGER.debug("Creating SemsStatusSwitch for Inverter %s", self._sn) @property def is_on(self) -> bool: """Return entity status.""" - _LOGGER.debug("coordinator.data[sn]: %s", self.coordinator.data[self.sn]) - return self.coordinator.data[self.sn]["status"] == 1 + status = self.coordinator.data.inverters.get(self._sn, {}).get("status") + return status == _INVERTER_STATUS_ON - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn off the inverter.""" - _LOGGER.debug("Inverter %s set to Off", self.sn) + _LOGGER.debug("Inverter %s set to off", self._sn) await self.hass.async_add_executor_job( - self.coordinator.semsApi.change_status, self.sn, 2 + self.coordinator.sems_api.change_status, + self._sn, + _COMMAND_TURN_OFF, ) + await self.coordinator.async_request_refresh() - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn on the inverter.""" - _LOGGER.debug("Inverter %s set to On", self.sn) + _LOGGER.debug("Inverter %s set to on", self._sn) await self.hass.async_add_executor_job( - self.coordinator.semsApi.change_status, self.sn, 4 + self.coordinator.sems_api.change_status, + self._sn, + _COMMAND_TURN_ON, ) + await self.coordinator.async_request_refresh() diff --git a/homeassistant/tesla_fleet.key b/homeassistant/tesla_fleet.key new file mode 100644 index 0000000..71ab8a1 --- /dev/null +++ b/homeassistant/tesla_fleet.key @@ -0,0 +1,5 @@ +-----BEGIN EC PRIVATE KEY----- +MHcCAQEEIGgbmLKO9Z+njxoSQWoeLBRyGzWZK0FYYZzk/dM1pDJVoAoGCCqGSM49 +AwEHoUQDQgAEhS5sHVP4YaWgVKAS0e/B/ObkA9lQ6Whx4Z+VYHhZtP3hQQLtVDGG +3e/2ncGTpyStQgLSi9Js3wr1GgoT/DNAsQ== +-----END EC PRIVATE KEY-----