Add ruff

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
Danielhiversen · Dec 28, 2023 · 526f00a · 526f00a
1 parent d777f47
commit 526f00a
Show file tree

Hide file tree

Showing 5 changed files with 29 additions and 83 deletions.
diff --git a/tibber/__init__.py b/tibber/__init__.py
@@ -155,10 +155,7 @@ async def update_info(self) -> None:
             self._all_home_ids += [home_id]
             if not (subs := _home.get("subscriptions")):
                 continue
-            if (
-                subs[0].get("status") is not None
-                and subs[0]["status"].lower() == "running"
-            ):
+            if subs[0].get("status") is not None and subs[0]["status"].lower() == "running":
                 self._active_home_ids += [home_id]
 
     def get_home_ids(self, only_active: bool = True) -> list[str]:
@@ -169,11 +166,7 @@ def get_home_ids(self, only_active: bool = True) -> list[str]:
 
     def get_homes(self, only_active: bool = True) -> list[TibberHome]:
         """Return list of Tibber homes."""
-        return [
-            home
-            for home_id in self.get_home_ids(only_active)
-            if (home := self.get_home(home_id))
-        ]
+        return [home for home_id in self.get_home_ids(only_active) if (home := self.get_home(home_id))]
 
     def get_home(self, home_id: str) -> TibberHome | None:
         """Return an instance of TibberHome for given home id."""
@@ -211,8 +204,9 @@ async def send_notification(self, title: str, message: str) -> bool:
 
     async def fetch_consumption_data_active_homes(self) -> None:
         """Fetch consumption data for active homes."""
-        await asyncio.gather(*[tibber_home.fetch_consumption_data()
-                               for tibber_home in self.get_homes(only_active=True)])
+        await asyncio.gather(
+            *[tibber_home.fetch_consumption_data() for tibber_home in self.get_homes(only_active=True)]
+        )
 
     async def fetch_production_data_active_homes(self) -> None:
         """Fetch production data for active homes."""

diff --git a/tibber/home.py b/tibber/home.py
@@ -95,8 +95,7 @@ async def _fetch_data(self, hourly_data: HourlyData) -> None:
         if (
             not hourly_data.data
             or hourly_data.last_data_timestamp is None
-            or parse(hourly_data.data[0]["from"])
-            < now - dt.timedelta(hours=n_hours + 24)
+            or parse(hourly_data.data[0]["from"]) < now - dt.timedelta(hours=n_hours + 24)
         ):
             hourly_data.data = []
         else:
@@ -120,9 +119,7 @@ async def _fetch_data(self, hourly_data: HourlyData) -> None:
         if not hourly_data.data:
             hourly_data.data = data
         else:
-            hourly_data.data = [
-                entry for entry in hourly_data.data if entry not in data
-            ]
+            hourly_data.data = [entry for entry in hourly_data.data if entry not in data]
             hourly_data.data.extend(data)
 
         _month_energy = 0
@@ -205,9 +202,7 @@ async def update_info(self) -> None:
 
     async def update_info_and_price_info(self) -> None:
         """Update home info and all price info asynchronously."""
-        if data := await self._tibber_control.execute(
-            UPDATE_INFO_PRICE % self._home_id
-        ):
+        if data := await self._tibber_control.execute(UPDATE_INFO_PRICE % self._home_id):
             self.info = data
             self._process_price_info(self.info)
 
@@ -248,9 +243,7 @@ def _process_price_info(self, price_info: dict[str, dict[str, Any]]) -> None:
         self._level_info = {}
         for key in ["current", "today", "tomorrow"]:
             try:
-                price_info_k = price_info["viewer"]["home"]["currentSubscription"][
-                    "priceInfo"
-                ][key]
+                price_info_k = price_info["viewer"]["home"]["currentSubscription"]["priceInfo"][key]
             except (KeyError, TypeError):
                 _LOGGER.error("Could not find price info for %s.", key)
                 continue
@@ -260,10 +253,7 @@ def _process_price_info(self, price_info: dict[str, dict[str, Any]]) -> None:
             for data in price_info_k:
                 self._price_info[data.get("startsAt")] = data.get("total")
                 self._level_info[data.get("startsAt")] = data.get("level")
-                if (
-                    not self.last_data_timestamp
-                    or parse(data.get("startsAt")) > self.last_data_timestamp
-                ):
+                if not self.last_data_timestamp or parse(data.get("startsAt")) > self.last_data_timestamp:
                     self.last_data_timestamp = parse(data.get("startsAt"))
 
     @property
@@ -319,9 +309,7 @@ def has_real_time_consumption(self) -> None | bool:
     def has_production(self) -> bool:
         """Return true if the home has a production metering point."""
         try:
-            return bool(
-                self.info["viewer"]["home"]["meteringPointData"]["productionEan"]
-            )
+            return bool(self.info["viewer"]["home"]["meteringPointData"]["productionEan"])
         except (KeyError, TypeError):
             return False
 
@@ -343,9 +331,7 @@ def consumption_unit(self) -> str:
     def currency(self) -> str:
         """Return the currency."""
         try:
-            return self.info["viewer"]["home"]["currentSubscription"]["priceInfo"][
-                "current"
-            ]["currency"]
+            return self.info["viewer"]["home"]["currentSubscription"]["priceInfo"]["current"]["currency"]
         except (KeyError, TypeError, IndexError):
             _LOGGER.error("Could not find currency.")
         return ""
@@ -393,29 +379,19 @@ async def rt_subscribe(self, callback: Callable[..., Any]) -> None:
 
         def _add_extra_data(data: dict[str, Any]) -> dict[str, Any]:
             live_data = data["data"]["liveMeasurement"]
-            _timestamp = parse(live_data["timestamp"]).astimezone(
-                self._tibber_control.time_zone
-            )
-            while self._rt_power and self._rt_power[0][0] < _timestamp - dt.timedelta(
-                minutes=5
-            ):
+            _timestamp = parse(live_data["timestamp"]).astimezone(self._tibber_control.time_zone)
+            while self._rt_power and self._rt_power[0][0] < _timestamp - dt.timedelta(minutes=5):
                 self._rt_power.pop(0)
 
             self._rt_power.append((_timestamp, live_data["power"] / 1000))
             current_hour = live_data["accumulatedConsumptionLastHour"]
             if current_hour is not None:
                 power = sum(p[1] for p in self._rt_power) / len(self._rt_power)
                 live_data["estimatedHourConsumption"] = round(
-                    current_hour
-                    + power
-                    * (3600 - (_timestamp.minute * 60 + _timestamp.second))
-                    / 3600,
+                    current_hour + power * (3600 - (_timestamp.minute * 60 + _timestamp.second)) / 3600,
                     3,
                 )
-                if (
-                    self._hourly_consumption_data.peak_hour
-                    and current_hour > self._hourly_consumption_data.peak_hour
-                ):
+                if self._hourly_consumption_data.peak_hour and current_hour > self._hourly_consumption_data.peak_hour:
                     self._hourly_consumption_data.peak_hour = round(current_hour, 2)
                     self._hourly_consumption_data.peak_hour_time = _timestamp
             return data
@@ -452,10 +428,7 @@ async def _start() -> None:
                         self.home_id,
                         data,
                     )
-                    if (
-                        self._rt_stopped
-                        or not self._tibber_control.realtime.subscription_running
-                    ):
+                    if self._rt_stopped or not self._tibber_control.realtime.subscription_running:
                         _LOGGER.debug("Stopping rt_subscribe loop")
                         return
             except Exception:  # pylint: disable=broad-except
@@ -547,9 +520,7 @@ async def get_historic_price_data(
         if not (data := await self._tibber_control.execute(query)):
             _LOGGER.error("Could not get the price data.")
             return None
-        return data["viewer"]["home"]["currentSubscription"]["priceRating"][resolution][
-            "entries"
-        ]
+        return data["viewer"]["home"]["currentSubscription"]["priceRating"][resolution]["entries"]
 
     def current_attributes(self) -> dict[str, float]:
         """Get current attributes."""

diff --git a/tibber/realtime.py b/tibber/realtime.py
@@ -103,9 +103,8 @@ async def _watchdog(self) -> None:  # noqa: PLR0912
         next_test_all_homes_running = dt.datetime.now(tz=dt.UTC)
         while self._watchdog_running:
             await asyncio.sleep(5)
-            if (
-                self.sub_manager.transport.running
-                and self.sub_manager.transport.reconnect_at > dt.datetime.now(tz=dt.UTC)
+            if self.sub_manager.transport.running and self.sub_manager.transport.reconnect_at > dt.datetime.now(
+                tz=dt.UTC
             ):
                 if dt.datetime.now(tz=dt.UTC) > next_test_all_homes_running:
                     is_running = True
@@ -120,9 +119,7 @@ async def _watchdog(self) -> None:  # noqa: PLR0912
                             continue
                         if not home.rt_subscription_running:
                             is_running = False
-                            next_test_all_homes_running = (
-                                dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=60)
-                            )
+                            next_test_all_homes_running = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=60)
                             break
                         _LOGGER.debug(
                             "Watchdog: Home %s is alive",
@@ -133,9 +130,7 @@ async def _watchdog(self) -> None:  # noqa: PLR0912
                         _LOGGER.debug("Watchdog: Connection is alive")
                         continue
 
-            self.sub_manager.transport.reconnect_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(
-                seconds=self._timeout
-            )
+            self.sub_manager.transport.reconnect_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=self._timeout)
             _LOGGER.error(
                 "Watchdog: Connection is down, %s",
                 self.sub_manager.transport.reconnect_at,
@@ -204,7 +199,5 @@ def sub_endpoint(self) -> str | None:
     def sub_endpoint(self, sub_endpoint: str) -> None:
         """Set subscription endpoint."""
         self._sub_endpoint = sub_endpoint
-        if self.sub_manager is not None and isinstance(
-            self.sub_manager.transport, TibberWebsocketsTransport
-        ):
+        if self.sub_manager is not None and isinstance(self.sub_manager.transport, TibberWebsocketsTransport):
             self.sub_manager.transport.url = sub_endpoint
diff --git a/tibber/response_handler.py b/tibber/response_handler.py
@@ -44,25 +44,17 @@ async def extract_response_data(response: ClientResponse) -> dict[Any, Any]:
         return result
 
     if response.status in HTTP_CODES_RETRIABLE:
-        error_code, error_message = extract_error_details(
-            result.get("errors", []), str(response.content)
-        )
+        error_code, error_message = extract_error_details(result.get("errors", []), str(response.content))
 
-        raise RetryableHttpExceptionError(
-            response.status, message=error_message, extension_code=error_code
-        )
+        raise RetryableHttpExceptionError(response.status, message=error_message, extension_code=error_code)
 
     if response.status in HTTP_CODES_FATAL:
-        error_code, error_message = extract_error_details(
-            result.get("errors", []), "request failed"
-        )
+        error_code, error_message = extract_error_details(result.get("errors", []), "request failed")
         if error_code == API_ERR_CODE_UNAUTH:
             raise InvalidLoginError(response.status, error_message, error_code)
 
         raise FatalHttpExceptionError(response.status, error_message, error_code)
 
     error_code, error_message = extract_error_details(result.get("errors", []), "N/A")
     # if reached here the HTTP response code is not currently handled
-    raise FatalHttpExceptionError(
-        response.status, f"Unhandled error: {error_message}", error_code
-    )
+    raise FatalHttpExceptionError(response.status, f"Unhandled error: {error_message}", error_code)
diff --git a/tibber/websocker_transport.py b/tibber/websocker_transport.py
@@ -22,9 +22,7 @@ def __init__(self, url: str, access_token: str, user_agent: str) -> None:
         )
         self._user_agent: str = user_agent
         self._timeout: int = 90
-        self.reconnect_at: dt.datetime = dt.datetime.now(tz=dt.UTC) + dt.timedelta(
-            seconds=self._timeout
-        )
+        self.reconnect_at: dt.datetime = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=self._timeout)
 
     @property
     def running(self) -> bool:
@@ -43,7 +41,5 @@ async def _receive(self) -> str:
 
     async def close(self) -> None:
         """Close the websocket connection."""
-        await self._fail(
-            TransportClosed(f"Tibber websocket closed by {self._user_agent}")
-        )
+        await self._fail(TransportClosed(f"Tibber websocket closed by {self._user_agent}"))
         await self.wait_closed()