Coverage for pygoodwe/__init__.py: 45%
262 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-15 13:36 +1000
1""" pygoodwe: a (terrible) interface to the goodwe solar API """
4from datetime import date, datetime
5import json
6import logging
7import os
8from pathlib import Path
9import sys
10import time
11from typing import Any, Dict, List, Optional, Union
13import requests
14from requests.sessions import Session
16__version__ = "0.0.17"
18POWERFLOW_STATUS_TEXT = {
19 -1: "Outward",
20}
21DEFAULT_UA = "PVMaster/2.0.4 (iPhone; iOS 11.4.1; Scale/2.00)"
22API_URL = "https://semsportal.com/api/"
25class API:
26 """API implementation"""
28 # pylint: disable=too-many-instance-attributes,too-many-arguments
29 def __init__(
30 self,
31 system_id: str,
32 account: str,
33 password: str,
34 api_url: str = API_URL,
35 log_level: Optional[str] = None,
36 user_agent: str = DEFAULT_UA,
37 skipload: bool = False,
38 ) -> None:
39 """
40 Options:
42 skipload: don't run self.getCurrentReadings() on init
43 api_url: you can change the API endpoint it hits
44 """
45 # TODO: lang: Real Soon Now it'll filter out any responses without that language
47 if log_level is None:
48 if "LOG_LEVEL" in os.environ:
49 log_level = os.environ["LOG_LEVEL"]
50 else:
51 log_level = "INFO"
53 if log_level in ("DEBUG", "INFO", "WARNING"):
54 log_level = getattr(logging, os.getenv("LOG_LEVEL", "INFO"))
55 logging.basicConfig(
56 level=log_level,
57 )
58 self.session = Session()
59 self.system_id = system_id
60 self.account = account
61 self.password = password
62 self.token = '{"version":"v2.0.4","client":"ios","language":"en"}'
63 self.global_url = api_url
64 self.base_url = self.global_url
66 logging.debug("API URL: %s", self.base_url)
68 self.user_agent = user_agent
70 if skipload:
71 logging.debug("Skipping initial load of data")
72 self.data: Dict[str, Any] = {}
73 else:
74 logging.debug("Doing load of data")
75 self.getCurrentReadings(raw=True)
77 def loaddata(self, filename: str) -> None:
78 """loads a json file of existing data"""
79 with open(filename, "r", encoding="utf8") as filehandle:
80 self.data = json.loads(filehandle.read())
82 _loaddata = loaddata
84 def get_current_readings(
85 self,
86 raw: bool = True,
87 retry: int = 1,
88 maxretries: int = 5,
89 delay: int = 30,
90 ) -> Dict[str, Any]: # pylint: disable=invalid-name
91 """gets readings at the current point in time"""
92 payload = {"powerStationId": self.system_id}
94 # GOODWE server
95 self.data = self.call(
96 "v2/PowerStation/GetMonitorDetailByPowerstationId", payload
97 )
99 retval = self.data
101 if not self.data.get("inverter"):
102 if retry < maxretries:
103 logging.error(
104 "no inverter data, try %s, trying again in %s seconds", retry, delay
105 )
106 time.sleep(delay)
107 return self.get_current_readings(
108 raw=raw, retry=retry + 1, maxretries=maxretries, delay=delay
109 )
110 logging.error("No inverter data after %s retries, quitting.", retry)
111 sys.exit(f"No inverter data after {retry} retries, quitting.")
112 return retval
114 # stub function names to old names
115 getCurrentReadings = get_current_readings
117 # def getDayReadings(self, date):
118 # date_s = date.strftime('%Y-%m-%d')
119 # payload = {
120 # 'powerStationId' : self.system_id
121 # }
122 # data = self.call("v2/PowerStation/GetMonitorDetailByPowerstationId", payload)
123 # if 'info' not in data:
124 # logging.warning(date_s + " - Received bad data " + str(data))
125 # return result
126 # result = {
127 # 'latitude' : data['info'].get('latitude'),
128 # 'longitude' : data['info'].get('longitude'),
129 # 'entries' : []
130 # }
131 # payload = {
132 # 'powerstation_id' : self.system_id,
133 # 'count' : 1,
134 # 'date' : date_s
135 # }
136 # data = self.call("PowerStationMonitor/GetPowerStationPowerAndIncomeByDay", payload)
137 # if len(data) == 0:
138 # logging.warning(date_s + " - Received bad data " + str(data))
139 # return result
140 # eday_kwh = data[0]['p']
141 # payload = {
142 # 'id' : self.system_id,
143 # 'date' : date_s
144 # }
145 # data = self.call("PowerStationMonitor/GetPowerStationPacByDayForApp", payload)
146 # if 'pacs' not in data:
147 # logging.warning(date_s + " - Received bad data " + str(data))
148 # return result
149 # minutes = 0
150 # eday_from_power = 0
151 # for sample in data['pacs']:
152 # parsed_date = datetime.strptime(sample['date'], "%m/%d/%Y %H:%M:%S")
153 # next_minutes = parsed_date.hour * 60 + parsed_date.minute
154 # sample['minutes'] = next_minutes - minutes
155 # minutes = next_minutes
156 # eday_from_power += sample['pac'] * sample['minutes']
157 # factor = eday_kwh / eday_from_power if eday_from_power > 0 else 1
158 # eday_kwh = 0
159 # for sample in data['pacs']:
160 # date += timedelta(minutes=sample['minutes'])
161 # pgrid_w = sample['pac']
162 # increase = pgrid_w * sample['minutes'] * factor
163 # if increase > 0:
164 # eday_kwh += increase
165 # result['entries'].append({
166 # 'dt' : date,
167 # 'pgrid_w': pgrid_w,
168 # 'eday_kwh': round(eday_kwh, 3)
169 # })
170 # return result
172 @property
173 def headers(self) -> Dict[str, str]:
174 """request headers"""
175 return {
176 "User-Agent": self.user_agent,
177 "Token": self.token,
178 }
180 # pylint: disable=invalid-name
181 def getDayDetailedReadingsExcel(
182 self,
183 export_date: date,
184 timeout: int = 10,
185 filename: Optional[str] = None,
186 ) -> bool:
187 """retrieves the detailed daily results of the given date as an Excel sheet,
188 processing the Excel sheet is outside the scope of the current module,
189 possible args:
190 - filename: the path where to write the output file, default "./Plant_Power_{datestr}.xls
191 """
192 datestr = datetime.strftime(export_date, "%Y-%m-%d")
193 if filename is None:
194 filename = f"Plant_Power_{datestr}.xls"
195 logging.debug("Will write data for %s to file: %s", datestr, filename)
197 uri = "v1/PowerStation/ExportPowerstationPac"
198 # {"api":"v2/PowerStation/ExportPowerstationPac","param":{"date":"2021-12-20","pw_id":"<my-pw-id>"
199 payload_export = {
200 "date": datestr,
201 "pw_id": self.system_id,
202 }
204 data = self.call(uri, payload=payload_export)
206 payload_get_url = {"id": data}
207 get_url_uri = "v1/ReportData/GetStationPowerDataFilePath"
208 data = self.call(get_url_uri, payload=payload_get_url)
210 file_url = data.get("file_path")
211 if file_url is None:
212 logging.error("Failed to get file path from ")
213 return False
215 response = requests.get(file_url, timeout=timeout)
216 response.raise_for_status()
218 try:
219 file_download_path = Path(filename)
220 file_download_path.write_bytes(response.content)
221 except Exception as error_message: # pylint: disable=broad-except
222 logging.error("Failed to write file %s! Error: %s", filename, error_message)
223 return False
224 return True
226 def do_login(self, timeout: int = 10) -> bool:
227 """does the login and token saving thing"""
228 login_payload = {
229 "account": self.account,
230 "pwd": self.password,
231 }
232 try:
233 response = self.session.post(
234 self.global_url + "v2/Common/CrossLogin",
235 headers=self.headers,
236 data=login_payload,
237 timeout=timeout,
238 )
239 response.raise_for_status()
240 except requests.exceptions.RequestException as exp:
241 logging.error("RequestException during do_login(): %s", exp)
242 print(f"{exp=}")
243 return False
245 data = response.json()
246 if data.get("code") != 0:
247 logging.error("Failed to log in: %s", data.get("msg"))
248 print(f"{data=}")
249 return False
251 if data.get("api"):
252 logging.debug("Setting base url to %s", data.get("api"))
253 self.base_url = data.get("api")
254 self.token = json.dumps(data.get("data"))
255 logging.debug("Done login, token: %s", self.token)
256 return True
258 def call(
259 self,
260 url: str,
261 payload: Any,
262 max_tries: int = 3,
263 timeout: int = 10,
264 ) -> Dict[str, Any]: # pylint: disable=unused-argument
265 """makes a call to the API"""
266 for i in range(1, max_tries):
267 try:
268 logging.debug(
269 "Pulling the following URL: base_url='%s', url='%s'",
270 self.base_url,
271 url,
272 )
273 response = self.session.post(
274 self.base_url + url,
275 headers=self.headers,
276 data=payload,
277 timeout=timeout,
278 )
279 response.raise_for_status()
280 data = response.json()
281 logging.debug("call response.json(): %s", json.dumps(data))
283 # APIs return "success", "Success", "Successful" in the 'msg'
284 # seen "Successful" in ExportPowerStationPac
285 # logging.error("Msg result %s - %s", self.base_url + url, data.get('msg', ''))
286 if (
287 data.get("msg", "").lower()
288 in (
289 "success",
290 "successful",
291 )
292 and "data" in data
293 ): # pylint: disable=no-else-return
294 logging.debug(
295 "Returning data: %s", json.dumps(data["data"], default=str)
296 )
297 result: Dict[str, Any] = data.get("data")
298 return result
299 logging.debug(json.dumps(data))
301 logging.debug("Logging in again...")
302 if not self.do_login():
303 logging.error("Failed to log in, bailing")
304 return {}
305 except requests.exceptions.RequestException as exp:
306 logging.error("RequestException: %s", exp)
307 logging.debug("Sleeping for %s seconds...", i)
308 time.sleep(i)
310 logging.error("Failed to call GoodWe API url='%s'", self.base_url + url)
311 return {}
313 @classmethod
314 def parseValue(cls, value: str, unit: str) -> float: # pylint: disable=invalid-name
315 """takes a string value and reutrns it as a float (if possible)"""
316 try:
317 return float(value.rstrip(unit))
318 except ValueError as exp:
319 logging.warning("ValueError: %s", exp)
320 return 0.0
322 def are_batteries_full(self, fullstate: float = 100.0) -> bool:
323 """boolean result for if the batteries are full. you can set your given 'full'
324 percentage in float if you want to lower this a little
325 are_batteries_full(fullstate=90.0): returns bool
326 """
327 for battery in self.get_batteries_soc():
328 if battery < fullstate:
329 return False
330 return True
332 def _get_batteries_soc(self) -> List[float]:
333 """returns a list of the state of charge for the batteries
334 returns: list[float,]
335 """
336 if not self.data:
337 self.getCurrentReadings()
338 if "inverter" not in self.data:
339 raise ValueError("Couldn't get data...")
340 return [
341 float(inverter.get("invert_full", {}).get("soc"))
342 for inverter in self.data["inverter"]
343 ]
345 def get_batteries_soc(self) -> List[float]:
346 """return the battery state of charge"""
347 return self._get_batteries_soc()
349 def getPVFlow(self) -> float: # pylint: disable=invalid-name
350 """PV flow data"""
351 raise NotImplementedError("SingleInverter has this, multi does not")
353 def getVoltage(self) -> List[float]: # pylint: disable=invalid-name
354 """returns the a list of the first AC channel voltages"""
355 if not self.data:
356 self.getCurrentReadings(True)
357 if "inverter" not in self.data:
358 raise ValueError("Couldn't get data...")
359 return [
360 float(inverter.get("invert_full", {}).get("vac1"))
361 for inverter in self.data["inverter"]
362 ]
364 def getPmeter(self) -> float: # pylint: disable=invalid-name
365 """gets the current line pmeter"""
366 if not self.data:
367 self.getCurrentReadings()
368 return float(self.data.get("inverter", {}).get("invert_full", {}).get("pmeter"))
370 def getLoadFlow(self) -> List[float]: # pylint: disable=invalid-name
371 """returns the list of inverter multi-unit load watts"""
372 raise NotImplementedError("multi-unit load watts isn't implemented yet")
374 def get_inverter_temperature(self) -> List[float]:
375 """returns the list of inverter temperatures"""
376 if not self.data:
377 self.get_current_readings(True)
378 if "inverter" not in self.data:
379 raise ValueError("Couldn't get data...")
380 return [
381 float(inverter.get("invert_full", {}).get("tempperature"))
382 for inverter in self.data["inverter"]
383 ]
385 def getDataPvoutput(
386 self,
387 ) -> Dict[str, Union[str, float]]: # pylint: disable=invalid-name
388 """updates and returns the data necessary for a one-shot pvoutput upload
389 'd' : testdate.strftime("%Y%m%d"),
390 't' : testtime.strftime("%H:%M"),
391 'v2' : 500, # power generation
392 'v4' : 450,
393 'v5' : 23.5, # temperature
394 'v6' : 234.0, # voltage
395 """
396 if not self.data:
397 self.getCurrentReadings()
398 # "time": "10/04/2019 14:37:29"
399 timestamp = datetime.strptime(
400 self.data.get("info", {}).get("time"), "%m/%d/%Y %H:%M:%S"
401 )
402 data: Dict[str, Union[str, float]] = {}
403 data["d"] = timestamp.strftime("%Y%m%d") # date
404 data["t"] = timestamp.strftime("%H:%M") # time
405 data["v2"] = self.getPVFlow() # PV Generation
406 data["v4"] = self.getLoadFlow()[0] # power consumption
407 data["v5"] = self.get_inverter_temperature()[0] # inverter temperature
408 data["v6"] = self.getVoltage()[0] # voltage
409 return data
412class SingleInverter(API):
413 """API implementation for an account with a single inverter"""
415 # pylint: disable=too-many-arguments
416 def __init__(
417 self,
418 system_id: str,
419 account: str,
420 password: str,
421 api_url: str = API_URL,
422 log_level: Optional[str] = None,
423 user_agent: str = DEFAULT_UA,
424 skipload: bool = False,
425 ) -> None:
426 self.loadflow = 0.0
427 self.loadflow_direction = ""
429 self.data: Dict[str, Any]
431 # instantiate the base class
432 super().__init__(
433 system_id, account, password, api_url, log_level, user_agent, skipload
434 )
436 def loaddata(self, filename: str) -> None:
437 """loads the ata from a given file"""
438 self._loaddata(filename)
439 if self.data.get("inverter"):
440 self.data["inverter"] = self.data["inverter"][0]
442 def get_current_readings(
443 self,
444 raw: bool = True,
445 retry: int = 1,
446 maxretries: int = 5,
447 delay: int = 30,
448 ) -> Any:
449 """grabs the data and makes sure self.data only has a single inverter"""
451 # update the data
452 super().get_current_readings(
453 raw=raw, retry=retry, maxretries=maxretries, delay=delay
454 )
456 # reduce self.data['inverter'] to a single dict from a list
457 self.data["inverter"] = self.data["inverter"][0]
459 return self.data
461 getCurrentReadings = get_current_readings
463 def _get_station_location(self) -> Dict[str, Union[str, int]]:
464 """gets the identified lat and long from the station data"""
465 return self.get_station_location()
467 def get_station_location(self) -> Dict[str, Union[str, int]]:
468 """gets the identified lat and long from the station data"""
469 if not self.data:
470 self.getCurrentReadings()
471 return {
472 "latitude": self.data.get("info", {}).get("latitude"),
473 "longitude": self.data.get("info", {}).get("longitude"),
474 }
476 def getPVFlow(self) -> float:
477 """returns the current flow amount of the PV panels"""
478 if not self.data:
479 self.getCurrentReadings()
480 if self.data["powerflow"]["pv"].endswith("(W)"):
481 pvflow = self.data["powerflow"]["pv"][:-3]
482 else:
483 pvflow = self.data["powerflow"]["pv"]
484 return float(pvflow)
486 def getVoltage(self) -> float: # type: ignore
487 """gets the current line voltage"""
488 if not self.data:
489 self.getCurrentReadings()
490 return float(self.data["inverter"]["invert_full"]["vac1"])
492 def get_day_income(self) -> float:
493 """gets the current daily income"""
494 if not self.data:
495 self.getCurrentReadings()
496 return float(self.data["kpi"]["day_income"])
498 def get_total_income(self) -> float:
499 """gets the total income"""
500 if not self.data:
501 self.getCurrentReadings()
502 return float(self.data["kpi"]["total_income"])
504 def get_total_power(self) -> float:
505 """gets the total power generated"""
506 if not self.data:
507 self.getCurrentReadings()
508 return float(self.data["kpi"]["total_power"])
510 def get_day_power(self) -> float:
511 """gets the total power generated"""
512 if not self.data:
513 self.getCurrentReadings()
514 return float(self.data["kpi"]["power"])
516 def getLoadFlow(self) -> float: # type: ignore
517 if not self.data:
518 self.getCurrentReadings()
519 if self.data["powerflow"]["bettery"].endswith("(W)"):
520 loadflow = float(self.data["powerflow"]["load"][:-3])
521 else:
522 loadflow = float(self.data["powerflow"]["load"])
523 # I'd love to see the *house* generate power
524 if self.data["powerflow"]["loadStatus"] == -1:
525 loadflow_direction = "Importing"
526 elif self.data["powerflow"]["loadStatus"] == 1:
527 loadflow_direction = "Using Battery"
528 else:
529 raise ValueError(
530 f"Your 'load' is doing something odd - status is '{self.data['powerflow']['loadStatus']}''."
531 ) # pylint: disable=line-too-long
532 self.loadflow = loadflow
533 self.loadflow_direction = loadflow_direction
534 return loadflow
536 def _get_batteries_soc(self) -> float: # type: ignore
537 """returns the state of charge of the battery"""
538 if not self.data:
539 self.getCurrentReadings()
540 if not self.data.get("soc", False):
541 raise ValueError("No state of charge available from data")
542 return float(self.data["soc"].get("power"))
544 def get_battery_soc(self) -> float:
545 """returns the single value state of charge for the batteries in the plant
546 returns : float
547 """
548 return self._get_batteries_soc()
550 def get_inverter_temperature(self) -> float: # type: ignore
551 if not self.data:
552 self.get_current_readings(True)
553 return float(self.data["inverter"]["tempperature"])
555 def getDataPvoutput(
556 self,
557 ) -> Dict[str, Union[str, float]]: # pylint: disable=invalid-name
558 """updates and returns the data necessary for a one-shot pvoutput upload
559 'd' : testdate.strftime("%Y%m%d"),
560 't' : testtime.strftime("%H:%M"),
561 'v2' : 500, # power generation
562 'v4' : 450,
563 'v5' : 23.5, # temperature
564 'v6' : 234.0, # voltage
565 """
566 if not self.data:
567 self.getCurrentReadings()
568 # "time": "10/04/2019 14:37:29"
569 timestamp = datetime.strptime(
570 self.data.get("info", {}).get("time"), "%m/%d/%Y %H:%M:%S"
571 )
572 data: Dict[str, Union[str, float]] = {}
573 data["d"] = timestamp.strftime("%Y%m%d") # date
574 data["t"] = timestamp.strftime("%H:%M") # time
575 data["v2"] = self.getPVFlow() # PV Generation
576 data["v4"] = self.getLoadFlow() # power consumption
577 data["v5"] = self.get_inverter_temperature() # inverter temperature
578 data["v6"] = self.getVoltage() # voltage
579 return data