diff --git a/sunweg/api.py b/sunweg/api.py index 4faaa2d..38c56cf 100644 --- a/sunweg/api.py +++ b/sunweg/api.py @@ -46,7 +46,9 @@ def convert_situation_status(situation: int) -> Status: return Status.WARN -def separate_value_metric(value_with_metric: str | None, default_metric: str = "") -> tuple[float, str]: +def separate_value_metric( + value_with_metric: str | None, default_metric: str = "" +) -> tuple[float, str]: """ Separate the value from the metric. @@ -62,12 +64,11 @@ def separate_value_metric(value_with_metric: str | None, default_metric: str = " split = value_with_metric.split(" ") return ( float(split[0].replace(",", ".")), - default_metric if len(split) < 2 else split[1] + default_metric if len(split) < 2 else split[1], ) - -class APIHelper(): +class APIHelper: """Class to call sunweg.net api.""" SERVER_URI = SUNWEG_URL @@ -120,7 +121,9 @@ def authenticate(self) -> bool: default=lambda o: o.__dict__, ) - result = self._post(SUNWEG_LOGIN_PATH, user_data) + result = self._post(SUNWEG_LOGIN_PATH, user_data, False) + if not result["success"]: + return False self._token = result["token"] return result["success"] @@ -168,7 +171,9 @@ def plant(self, id: int, retry=True) -> Plant | None: try: result = self._get(SUNWEG_PLANT_DETAIL_PATH + str(id)) - (today_energy, today_energy_metric) = separate_value_metric(result["energiaGeradaHoje"], "kWh") + (today_energy, today_energy_metric) = separate_value_metric( + result["energiaGeradaHoje"], "kWh" + ) total_power = separate_value_metric(result["AcumuladoPotencia"])[0] plant = Plant( id=id, @@ -220,8 +225,12 @@ def inverter(self, id: int, retry=True) -> Inverter | None: """ try: result = self._get(SUNWEG_INVERTER_DETAIL_PATH + str(id)) - (total_energy, total_energy_metric) = separate_value_metric(result["energiaacumulada"], "kWh") - (today_energy, today_energy_metric) = separate_value_metric(result["energiadodia"], "kWh") + (total_energy, total_energy_metric) = separate_value_metric( + result["energiaacumulada"], "kWh" + ) + (today_energy, today_energy_metric) = separate_value_metric( + result["energiadodia"], "kWh" + ) (power, power_metric) = separate_value_metric(result["potenciaativa"], "kW") inverter = Inverter( id=id, @@ -259,9 +268,17 @@ def complete_inverter(self, inverter: Inverter, retry=True) -> None: """ try: result = self._get(SUNWEG_INVERTER_DETAIL_PATH + str(inverter.id)) - (inverter.total_energy, inverter.total_energy_metric) = separate_value_metric(result["energiaacumulada"], "kWh") - (inverter.today_energy, inverter.today_energy_metric) = separate_value_metric(result["energiadodia"], "kWh") - (inverter.power, inverter.power_metric) = separate_value_metric(result["potenciaativa"], "kW") + ( + inverter.total_energy, + inverter.total_energy_metric, + ) = separate_value_metric(result["energiaacumulada"], "kWh") + ( + inverter.today_energy, + inverter.today_energy_metric, + ) = separate_value_metric(result["energiadodia"], "kWh") + (inverter.power, inverter.power_metric) = separate_value_metric( + result["potenciaativa"], "kW" + ) inverter.power_factor = float(result["fatorpotencia"].replace(",", ".")) inverter.frequency = float(result["frequencia"].replace(",", ".")) @@ -271,7 +288,14 @@ def complete_inverter(self, inverter: Inverter, retry=True) -> None: self.authenticate() self.complete_inverter(inverter, False) - def month_stats_production(self, year: int, month: int, plant: Plant, inverter: Inverter | None = None, retry: bool = True) -> list[ProductionStats]: + def month_stats_production( + self, + year: int, + month: int, + plant: Plant, + inverter: Inverter | None = None, + retry: bool = True, + ) -> list[ProductionStats]: """ Retrieve month energy production statistics. @@ -288,9 +312,18 @@ def month_stats_production(self, year: int, month: int, plant: Plant, inverter: :return: list of daily energy production statistics :rtype: list[ProductionStats] """ - return self.month_stats_production_by_id(year, month, plant.id, inverter.id if inverter is not None else None, retry) + return self.month_stats_production_by_id( + year, month, plant.id, inverter.id if inverter is not None else None, retry + ) - def month_stats_production_by_id(self, year: int, month: int, plant_id: int, inverter_id: int | None = None, retry: bool = True) -> list[ProductionStats]: + def month_stats_production_by_id( + self, + year: int, + month: int, + plant_id: int, + inverter_id: int | None = None, + retry: bool = True, + ) -> list[ProductionStats]: """ Retrieve month energy production statistics. @@ -307,14 +340,26 @@ def month_stats_production_by_id(self, year: int, month: int, plant_id: int, inv :return: list of daily energy production statistics :rtype: list[ProductionStats] """ - inverter_str:str = str(inverter_id) if inverter_id is not None else "" + inverter_str: str = str(inverter_id) if inverter_id is not None else "" try: - result = self._get(SUNWEG_MONTH_STATS_PATH + f"idusina={plant_id}&idinversor={inverter_str}&date={format(month,'02')}/{year}") - return [ProductionStats(datetime.strptime(item["tempoatual"],"%Y-%m-%d").date(), float(item["energiapordia"]), float(item["prognostico"])) for item in result["graficomes"]] + result = self._get( + SUNWEG_MONTH_STATS_PATH + + f"idusina={plant_id}&idinversor={inverter_str}&date={format(month,'02')}/{year}" + ) + return [ + ProductionStats( + datetime.strptime(item["tempoatual"], "%Y-%m-%d").date(), + float(item["energiapordia"]), + float(item["prognostico"]), + ) + for item in result["graficomes"] + ] except LoginError: if retry: self.authenticate() - return self.month_stats_production_by_id(year, month, plant_id, inverter_id, False) + return self.month_stats_production_by_id( + year, month, plant_id, inverter_id, False + ) return [] def _populate_MPPT(self, result: dict, inverter: Inverter) -> None: @@ -326,7 +371,9 @@ def _populate_MPPT(self, result: dict, inverter: Inverter) -> None: string = String( str_string["nome"], float(result["inversor"]["leitura"][str_string["variaveltensao"]]), - float(result["inversor"]["leitura"][str_string["variavelcorrente"]]), + float( + result["inversor"]["leitura"][str_string["variavelcorrente"]] + ), convert_situation_status(int(str_string["situacao"])), ) mppt.strings.append(string) @@ -346,27 +393,31 @@ def _populate_MPPT(self, result: dict, inverter: Inverter) -> None: ) ) - def _get(self, path: str) -> dict: + def _get(self, path: str, launch_exception_on_error: bool = True) -> dict: """Do a get request returning a treated response.""" res = self.session.get(self.SERVER_URI + path, headers=self._headers()) - result = self._treat_response(res) + result = self._treat_response(res, launch_exception_on_error) return result - def _post(self, path: str, data: Any | None) -> dict: + def _post( + self, path: str, data: Any | None, launch_exception_on_error: bool = True + ) -> dict: """Do a post request returning a treated response.""" res = self.session.post( self.SERVER_URI + path, data=data, headers=self._headers() ) - result = self._treat_response(res) + result = self._treat_response(res, launch_exception_on_error) return result - def _treat_response(self, response: Response) -> dict: + def _treat_response( + self, response: Response, launch_exception_on_error: bool = True + ) -> dict: """Treat the response from requests.""" if response.status_code == 401: raise LoginError("Request failed: %s" % response) if response.status_code != 200: raise SunWegApiError("Request failed: %s" % response) result = response.json() - if not result["success"]: + if launch_exception_on_error and not result["success"]: raise SunWegApiError(result["message"]) return result diff --git a/tests/test_api.py b/tests/test_api.py index 5658b8c..387d263 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -8,7 +8,12 @@ from requests import Response -from sunweg.api import APIHelper, convert_situation_status, SunWegApiError, separate_value_metric +from sunweg.api import ( + APIHelper, + convert_situation_status, + SunWegApiError, + separate_value_metric, +) from sunweg.device import Inverter, String from sunweg.util import Status @@ -34,7 +39,6 @@ def setUp(self) -> None: response._content = "".join(f.readlines()).encode() self.responses[filename] = response - def test_convert_situation_status(self) -> None: """Test the conversion from situation to status.""" status_ok: Status = convert_situation_status(1) @@ -45,46 +49,42 @@ def test_convert_situation_status(self) -> None: assert status_err == Status.ERROR assert status_wrn == Status.WARN - def test_separate_value_metric_comma(self) -> None: """Test the separation from value and metric of string with comma.""" - (value,metric) = separate_value_metric("0,0") + (value, metric) = separate_value_metric("0,0") assert value == 0 assert metric == "" - (value,metric) = separate_value_metric("1,0", "W") + (value, metric) = separate_value_metric("1,0", "W") assert value == 1.0 assert metric == "W" - (value,metric) = separate_value_metric("0,2 kW", "W") + (value, metric) = separate_value_metric("0,2 kW", "W") assert value == 0.2 assert metric == "kW" - def test_separate_value_metric_dot(self) -> None: """Test the separation from value and metric of string with dot.""" - (value,metric) = separate_value_metric("0.0") + (value, metric) = separate_value_metric("0.0") assert value == 0 assert metric == "" - (value,metric) = separate_value_metric("1.0", "W") + (value, metric) = separate_value_metric("1.0", "W") assert value == 1.0 assert metric == "W" - (value,metric) = separate_value_metric("0.2 kW", "W") + (value, metric) = separate_value_metric("0.2 kW", "W") assert value == 0.2 assert metric == "kW" - def test_separate_value_metric_none_int(self) -> None: """Test the separation from value and metric of string with dot.""" - (value,metric) = separate_value_metric(None) + (value, metric) = separate_value_metric(None) assert value == 0 assert metric == "" - (value,metric) = separate_value_metric("1", "W") + (value, metric) = separate_value_metric("1", "W") assert value == 1.0 assert metric == "W" - (value,metric) = separate_value_metric("2 kW", "W") + (value, metric) = separate_value_metric("2 kW", "W") assert value == 2.0 assert metric == "kW" - def test_error500(self) -> None: """Test error 500.""" with patch( @@ -112,9 +112,7 @@ def test_authenticate_failed(self) -> None: return_value=self.responses["auth_fail_response.json"], ): api = APIHelper("user@acme.com", "password") - with pytest.raises(SunWegApiError) as e_info: - api.authenticate() - assert e_info.value.__str__() == "Error message" + assert not api.authenticate() def test_list_plants_none_success(self) -> None: """Test list plants with empty plant list.""" @@ -369,7 +367,7 @@ def test_month_stats_401(self) -> None: assert len(stats) == 0 def test_month_stats_success(self) -> None: - """Test month stats with data from server.""" + """Test month stats with data from server.""" with patch( "requests.Session.get", return_value=self.responses["month_stats_success_response.json"], @@ -384,5 +382,7 @@ def test_month_stats_success(self) -> None: assert stat.date == date(2023, 12, i) assert isinstance(stat.production, float) assert stat.prognostic == 98.774193548387 - assert stat.__str__().startswith("") + assert stat.__str__().startswith( + "" + ) i += 1