-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathcoordinator.py
171 lines (145 loc) · 6.57 KB
/
coordinator.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
"""The Solcast Solar integration."""
from __future__ import annotations
import logging
import traceback
from homeassistant.core import HomeAssistant
from homeassistant.helpers.event import async_track_utc_time_change
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .solcastapi import SolcastApi
_LOGGER = logging.getLogger(__name__)
class SolcastUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching data from Solcast Solar API."""
def __init__(self, hass: HomeAssistant, solcast: SolcastApi, version: str) -> None:
"""Initialize."""
self.solcast = solcast
self._hass = hass
self._previousenergy = None
self._version = version
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
)
async def _async_update_data(self):
"""Update data via library."""
return self.solcast._data
async def setup(self):
d={}
self._previousenergy = d
try:
#4.0.18 - added reset usage call to reset usage sensors at UTC midnight
async_track_utc_time_change(self._hass, self.update_utcmidnight_usage_sensor_data, hour=0,minute=0,second=0)
async_track_utc_time_change(self._hass, self.update_integration_listeners, second=0)
except Exception as error:
_LOGGER.error("SOLCAST - Error coordinator setup: %s", traceback.format_exc())
async def update_integration_listeners(self, *args):
try:
self.async_update_listeners()
except Exception:
#_LOGGER.error("SOLCAST - update_integration_listeners: %s", traceback.format_exc())
pass
async def update_utcmidnight_usage_sensor_data(self, *args):
try:
self.solcast._api_used = 0
self.async_update_listeners()
except Exception:
#_LOGGER.error("SOLCAST - update_utcmidnight_usage_sensor_data: %s", traceback.format_exc())
pass
async def service_event_update(self, *args):
#await self.solcast.sites_weather()
await self.solcast.http_data(dopast=False)
await self.update_integration_listeners()
async def service_event_delete_old_solcast_json_file(self, *args):
await self.solcast.delete_solcast_file()
async def service_query_forecast_data(self, *args) -> tuple:
return await self.solcast.get_forecast_list(*args)
def get_energy_tab_data(self):
return self.solcast.get_energy_data()
def get_sensor_value(self, key=""):
if key == "total_kwh_forecast_today":
return self.solcast.get_total_kwh_forecast_day(0)
elif key == "peak_w_today":
return self.solcast.get_peak_w_day(0)
elif key == "peak_w_time_today":
return self.solcast.get_peak_w_time_day(0)
elif key == "forecast_this_hour":
return self.solcast.get_forecast_n_hour(0)
elif key == "forecast_next_hour":
return self.solcast.get_forecast_n_hour(1)
elif key == "forecast_custom_hour":
return self.solcast.get_forecast_custom_hour(self.solcast._customhoursensor)
elif key == "forecast_next_12hour":
return self.solcast.get_forecast_n_hour(12)
elif key == "forecast_next_24hour":
return self.solcast.get_forecast_n_hour(24)
elif key == "total_kwh_forecast_tomorrow":
return self.solcast.get_total_kwh_forecast_day(1)
elif key == "total_kwh_forecast_d3":
return self.solcast.get_total_kwh_forecast_day(2)
elif key == "total_kwh_forecast_d4":
return self.solcast.get_total_kwh_forecast_day(3)
elif key == "total_kwh_forecast_d5":
return self.solcast.get_total_kwh_forecast_day(4)
elif key == "total_kwh_forecast_d6":
return self.solcast.get_total_kwh_forecast_day(5)
elif key == "total_kwh_forecast_d7":
return self.solcast.get_total_kwh_forecast_day(6)
elif key == "power_now":
return self.solcast.get_power_production_n_mins(0)
elif key == "power_now_30m":
return self.solcast.get_power_production_n_mins(30)
elif key == "power_now_1hr":
return self.solcast.get_power_production_n_mins(60)
elif key == "power_now_12hr":
return self.solcast.get_power_production_n_mins(60*12)
elif key == "power_now_24hr":
return self.solcast.get_power_production_n_mins(60*24)
elif key == "peak_w_tomorrow":
return self.solcast.get_peak_w_day(1)
elif key == "peak_w_time_tomorrow":
return self.solcast.get_peak_w_time_day(1)
elif key == "get_remaining_today":
return self.solcast.get_remaining_today()
elif key == "api_counter":
return self.solcast.get_api_used_count()
elif key == "api_limit":
return self.solcast.get_api_limit()
elif key == "lastupdated":
return self.solcast.get_last_updated_datetime()
elif key == "hard_limit":
#return self.solcast._hardlimit < 100
return False if self.solcast._hardlimit == 100 else f"{round(self.solcast._hardlimit * 1000)}w"
# elif key == "weather_description":
# return self.solcast.get_weather()
#just in case
return None
def get_sensor_extra_attributes(self, key=""):
if key == "total_kwh_forecast_today":
return self.solcast.get_forecast_day(0)
elif key == "total_kwh_forecast_tomorrow":
return self.solcast.get_forecast_day(1)
elif key == "total_kwh_forecast_d3":
return self.solcast.get_forecast_day(2)
elif key == "total_kwh_forecast_d4":
return self.solcast.get_forecast_day(3)
elif key == "total_kwh_forecast_d5":
return self.solcast.get_forecast_day(4)
elif key == "total_kwh_forecast_d6":
return self.solcast.get_forecast_day(5)
elif key == "total_kwh_forecast_d7":
return self.solcast.get_forecast_day(6)
#just in case
return None
def get_site_sensor_value(self, roof_id, key):
match key:
case "site_data":
return self.solcast.get_rooftop_site_total_today(roof_id)
case _:
return None
def get_site_sensor_extra_attributes(self, roof_id, key):
match key:
case "site_data":
return self.solcast.get_rooftop_site_extra_data(roof_id)
case _:
return None