mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Add new Solaredge sensors (#34525)
* Changes to be committed: modified: homeassistant/components/solaredge/const.py modified: homeassistant/components/solaredge/sensor.py Solaredge as recently changed its policy about local api access, so solaredge-local doesn't work with last firmware update for almost users. Please check https://github.com/home-assistant/core.git Anyway the solardge remote api is still working, but doesn't got some usefull sensor information as Power SelfConsumption, Power Exported, Power Imported. With my update, I'll fetching API energy details where we got these new sensors. * Grammar/syntax fix * Indentation fix * Black formatting fix * isort fix * To force re-check * Fix too-many-nested-blocks * Fix indentation * Fix Black formatting :D * Fix Redefining built-in var * Removed comment to force check
This commit is contained in:
terminet85
GitHub
parent
6610bbe7bb
commit
a004e6aa68
@ -10,12 +10,13 @@ CONF_SITE_ID = "site_id"
|
||||
|
||||
DEFAULT_NAME = "SolarEdge"
|
||||
|
||||
OVERVIEW_UPDATE_DELAY = timedelta(minutes=10)
|
||||
OVERVIEW_UPDATE_DELAY = timedelta(minutes=15)
|
||||
DETAILS_UPDATE_DELAY = timedelta(hours=12)
|
||||
INVENTORY_UPDATE_DELAY = timedelta(hours=12)
|
||||
POWER_FLOW_UPDATE_DELAY = timedelta(minutes=10)
|
||||
POWER_FLOW_UPDATE_DELAY = timedelta(minutes=15)
|
||||
ENERGY_DETAILS_DELAY = timedelta(minutes=15)
|
||||
|
||||
SCAN_INTERVAL = timedelta(minutes=10)
|
||||
SCAN_INTERVAL = timedelta(minutes=15)
|
||||
|
||||
# Supported overview sensor types:
|
||||
# Key: ['json_key', 'name', unit, icon, default]
|
||||
@ -65,4 +66,15 @@ SENSOR_TYPES = {
|
||||
"solar_power": ["PV", "Solar Power", None, "mdi:solar-power", False],
|
||||
"grid_power": ["GRID", "Grid Power", None, "mdi:power-plug", False],
|
||||
"storage_power": ["STORAGE", "Storage Power", None, "mdi:car-battery", False],
|
||||
"purchased_power": ["Purchased", "Imported Power", None, "mdi:flash", False],
|
||||
"production_power": ["Production", "Production Power", None, "mdi:flash", False],
|
||||
"consumption_power": ["Consumption", "Cosumption Power", None, "mdi:flash", False],
|
||||
"selfconsumption_power": [
|
||||
"SelfConsumption",
|
||||
"SelfConsumption Power",
|
||||
None,
|
||||
"mdi:flash",
|
||||
False,
|
||||
],
|
||||
"feedin_power": ["FeedIn", "Exported Power", None, "mdi:flash", False],
|
||||
}
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
"""Support for SolarEdge Monitoring API."""
|
||||
from datetime import date, datetime
|
||||
import logging
|
||||
|
||||
from requests.exceptions import ConnectTimeout, HTTPError
|
||||
@ -12,6 +13,7 @@ from homeassistant.util import Throttle
|
||||
from .const import (
|
||||
CONF_SITE_ID,
|
||||
DETAILS_UPDATE_DELAY,
|
||||
ENERGY_DETAILS_DELAY,
|
||||
INVENTORY_UPDATE_DELAY,
|
||||
OVERVIEW_UPDATE_DELAY,
|
||||
POWER_FLOW_UPDATE_DELAY,
|
||||
@ -62,6 +64,7 @@ class SolarEdgeSensorFactory:
|
||||
overview = SolarEdgeOverviewDataService(api, site_id)
|
||||
inventory = SolarEdgeInventoryDataService(api, site_id)
|
||||
flow = SolarEdgePowerFlowDataService(api, site_id)
|
||||
energy = SolarEdgeEnergyDetailsService(api, site_id)
|
||||
|
||||
self.services = {"site_details": (SolarEdgeDetailsSensor, details)}
|
||||
|
||||
@ -80,6 +83,15 @@ class SolarEdgeSensorFactory:
|
||||
for key in ["power_consumption", "solar_power", "grid_power", "storage_power"]:
|
||||
self.services[key] = (SolarEdgePowerFlowSensor, flow)
|
||||
|
||||
for key in [
|
||||
"purchased_power",
|
||||
"production_power",
|
||||
"feedin_power",
|
||||
"consumption_power",
|
||||
"selfconsumption_power",
|
||||
]:
|
||||
self.services[key] = (SolarEdgeEnergyDetailsSensor, energy)
|
||||
|
||||
def create_sensor(self, sensor_key):
|
||||
"""Create and return a sensor based on the sensor_key."""
|
||||
sensor_class, service = self.services[sensor_key]
|
||||
@ -181,6 +193,30 @@ class SolarEdgeInventorySensor(SolarEdgeSensor):
|
||||
self._attributes = self.data_service.attributes[self._json_key]
|
||||
|
||||
|
||||
class SolarEdgeEnergyDetailsSensor(SolarEdgeSensor):
|
||||
"""Representation of an SolarEdge Monitoring API power flow sensor."""
|
||||
|
||||
def __init__(self, platform_name, sensor_key, data_service):
|
||||
"""Initialize the power flow sensor."""
|
||||
super().__init__(platform_name, sensor_key, data_service)
|
||||
|
||||
self._json_key = SENSOR_TYPES[self.sensor_key][0]
|
||||
|
||||
self._attributes = {}
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
return self._attributes
|
||||
|
||||
def update(self):
|
||||
"""Get the latest inventory data and update state and attributes."""
|
||||
self.data_service.update()
|
||||
self._state = self.data_service.data.get(self._json_key)
|
||||
self._attributes = self.data_service.attributes.get(self._json_key)
|
||||
self._unit_of_measurement = self.data_service.unit
|
||||
|
||||
|
||||
class SolarEdgePowerFlowSensor(SolarEdgeSensor):
|
||||
"""Representation of an SolarEdge Monitoring API power flow sensor."""
|
||||
|
||||
@ -319,6 +355,68 @@ class SolarEdgeInventoryDataService(SolarEdgeDataService):
|
||||
_LOGGER.debug("Updated SolarEdge inventory: %s, %s", self.data, self.attributes)
|
||||
|
||||
|
||||
class SolarEdgeEnergyDetailsService(SolarEdgeDataService):
|
||||
"""Get and update the latest power flow data."""
|
||||
|
||||
def __init__(self, api, site_id):
|
||||
"""Initialize the power flow data service."""
|
||||
super().__init__(api, site_id)
|
||||
|
||||
self.unit = None
|
||||
|
||||
@Throttle(ENERGY_DETAILS_DELAY)
|
||||
def update(self):
|
||||
"""Update the data from the SolarEdge Monitoring API."""
|
||||
try:
|
||||
now = datetime.now()
|
||||
today = date.today()
|
||||
midnight = datetime.combine(today, datetime.min.time())
|
||||
data = self.api.get_energy_details(
|
||||
self.site_id,
|
||||
midnight,
|
||||
now.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
meters=None,
|
||||
time_unit="DAY",
|
||||
)
|
||||
energy_details = data["energyDetails"]
|
||||
except KeyError:
|
||||
_LOGGER.error("Missing power flow data, skipping update")
|
||||
return
|
||||
except (ConnectTimeout, HTTPError):
|
||||
_LOGGER.error("Could not retrieve data, skipping update")
|
||||
return
|
||||
|
||||
if "meters" not in energy_details:
|
||||
_LOGGER.debug(
|
||||
"Missing meters in energy details data. Assuming site does not have any"
|
||||
)
|
||||
return
|
||||
|
||||
self.data = {}
|
||||
self.attributes = {}
|
||||
self.unit = energy_details["unit"]
|
||||
meters = energy_details["meters"]
|
||||
|
||||
for entity in meters:
|
||||
for key, data in entity.items():
|
||||
if key == "type" and data in [
|
||||
"Production",
|
||||
"SelfConsumption",
|
||||
"FeedIn",
|
||||
"Purchased",
|
||||
"Consumption",
|
||||
]:
|
||||
energy_type = data
|
||||
if key == "values":
|
||||
for row in data:
|
||||
self.data[energy_type] = row["value"]
|
||||
self.attributes[energy_type] = {"date": row["date"]}
|
||||
|
||||
_LOGGER.debug(
|
||||
"Updated SolarEdge energy details: %s, %s", self.data, self.attributes
|
||||
)
|
||||
|
||||
|
||||
class SolarEdgePowerFlowDataService(SolarEdgeDataService):
|
||||
"""Get and update the latest power flow data."""
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user