Ruff format

Author: ziima

recuair_cli/__init__.py

@@ -1,3 +1,3 @@
 """Command line interface for manipulation of Recuair devices."""
 
-__version__ = '0.0.0'
+__version__ = "0.0.0"

recuair_cli/main.py

@@ -23,11 +23,13 @@
   --version             show program's version number and exit
   --debug               print debug logs
 """
+
 import asyncio
 import logging
 import sys
+from collections.abc import Coroutine
 from http import HTTPStatus
-from typing import Any, Callable, Coroutine, Dict, List, NamedTuple, Optional, TypeVar, cast
+from typing import Any, Callable, NamedTuple, Optional, TypeVar, cast
 
 import httpx
 from bs4 import BeautifulSoup, PageElement, Tag
@@ -73,13 +75,13 @@ class Status(NamedTuple):
 
 def _strip_unit(value: str) -> str:
     """Strip unit from quantity and return only a quantity value."""
-    return value.strip().partition(' ')[0]
+    return value.strip().partition(" ")[0]
 
 
 async def get_status(client: httpx.AsyncClient, device: str) -> Status:
     """Return device status."""
     try:
-        response = await client.get(f'http://{device}/', timeout=3)
+        response = await client.get(f"http://{device}/", timeout=3)
         response.raise_for_status()
     except httpx.HTTPError as error:
         _LOGGER.debug("Error encountered: %s", error)
@@ -88,26 +90,35 @@ async def get_status(client: httpx.AsyncClient, device: str) -> Status:
 
     try:
         content = BeautifulSoup(response.text, features="html.parser")
-        container = cast(Tag, content.find(class_='container'))
-        temperature_raw = cast(PageElement,
-                               cast(Tag, container.find_all(class_='col-12')[1]).find(class_='bigText')).text
-        in_data, _, temp_out = temperature_raw.strip().partition('%')
-        temp_in, _, humi_in = in_data.strip().partition('/')
-        mode_raw = cast(PageElement, cast(Tag, container.find_all(class_='col-12')[3]).find('span')).text
-        co2_raw = cast(PageElement, cast(Tag, container.find_all(class_='col-12')[4]).find('b')).text
-        filter_raw = cast(
-            str,
-            cast(Tag, cast(Tag, container.find_all(class_='filterBox')[1]).div)['style'],
-        ).partition(':')[2].partition('%')[0]
-        fan_raw = cast(
-            str,
-            cast(Tag, cast(Tag, container.find_all(class_='filterBox')[2]).div)['style'],
-        ).partition(':')[2].partition('%')[0]
-        light_raw = cast(str, cast(Tag, container.find(id='myRange'))['value'])
+        container = cast(Tag, content.find(class_="container"))
+        temperature_raw = cast(
+            PageElement, cast(Tag, container.find_all(class_="col-12")[1]).find(class_="bigText")
+        ).text
+        in_data, _, temp_out = temperature_raw.strip().partition("%")
+        temp_in, _, humi_in = in_data.strip().partition("/")
+        mode_raw = cast(PageElement, cast(Tag, container.find_all(class_="col-12")[3]).find("span")).text
+        co2_raw = cast(PageElement, cast(Tag, container.find_all(class_="col-12")[4]).find("b")).text
+        filter_raw = (
+            cast(
+                str,
+                cast(Tag, cast(Tag, container.find_all(class_="filterBox")[1]).div)["style"],
+            )
+            .partition(":")[2]
+            .partition("%")[0]
+        )
+        fan_raw = (
+            cast(
+                str,
+                cast(Tag, cast(Tag, container.find_all(class_="filterBox")[2]).div)["style"],
+            )
+            .partition(":")[2]
+            .partition("%")[0]
+        )
+        light_raw = cast(str, cast(Tag, container.find(id="myRange"))["value"])
 
         return Status(
             device=device,
-            name=cast(Tag, content.find(class_='deviceName')).text,
+            name=cast(Tag, content.find(class_="deviceName")).text,
             temperature_in=int(_strip_unit(temp_in)),
             humidity_in=int(_strip_unit(humi_in)),
             temperature_out=int(_strip_unit(temp_out)),
@@ -122,11 +133,11 @@ async def get_status(client: httpx.AsyncClient, device: str) -> Status:
         raise RecuairError(f"Invalid response returned from device {device}") from error
 
 
-async def post_request(client: httpx.AsyncClient, device: str, data: Dict[str, Any]) -> None:
+async def post_request(client: httpx.AsyncClient, device: str, data: dict[str, Any]) -> None:
     """Send a POST request to the device."""
     try:
         # XXX: Disable redirects. Recuair returns 301 for POST requests.
-        response = await client.post(f'http://{device}/', data=data, timeout=5, follow_redirects=False)
+        response = await client.post(f"http://{device}/", data=data, timeout=5, follow_redirects=False)
     except httpx.HTTPError as error:
         _LOGGER.debug("Error encountered: %s", error)
         raise RecuairError(f"Error from device {device}: {error}") from error
@@ -136,38 +147,43 @@ async def post_request(client: httpx.AsyncClient, device: str, data: Dict[str, A
     _LOGGER.debug("Response [%s]: %s", response, response.text)
 
 
-X = TypeVar('X')
+X = TypeVar("X")
 
 
 # XXX: Add retry, recuair devices are often irresponsive.
 def _wrap_retry(func: Callable[..., X]) -> Callable[..., X]:
     return retry(reraise=True, stop=stop_after_attempt(10), wait=wait_exponential(max=30))(func)
 
 
-async def _run(options: Dict[str, str]) -> None:
+async def _run(options: dict[str, str]) -> None:  # noqa: C901
     """Actually run the command."""
     error_found = False
     async with httpx.AsyncClient() as client:
-        coros: List[Coroutine] = []
-        for device in options['<device>']:
-            if options['start']:
+        coros: list[Coroutine] = []
+        for device in options["<device>"]:
+            if options["start"]:
                 # XXX: Start in auto mode. Recuair GUI starts on mode 1.
-                coros.append(_wrap_retry(post_request)(client, device, {'mode': 'auto'}))
-            elif options['stop']:
-                coros.append(_wrap_retry(post_request)(client, device, {'mode': 'off'}))
-            elif options['holiday']:
-                coros.append(_wrap_retry(post_request)(client, device, {'mode': 'holiday'}))
-            elif options['bypass']:
-                coros.append(_wrap_retry(post_request)(client, device, {'mode': 'bypass'}))
-            elif options['light']:
+                coros.append(_wrap_retry(post_request)(client, device, {"mode": "auto"}))
+            elif options["stop"]:
+                coros.append(_wrap_retry(post_request)(client, device, {"mode": "off"}))
+            elif options["holiday"]:
+                coros.append(_wrap_retry(post_request)(client, device, {"mode": "holiday"}))
+            elif options["bypass"]:
+                coros.append(_wrap_retry(post_request)(client, device, {"mode": "bypass"}))
+            elif options["light"]:
                 # XXX: Recuair doesn't accept only change in light intensity.
                 # Whole light setting has to be provided.
-                if options['off']:
+                if options["off"]:
                     coros.append(
-                        _wrap_retry(post_request)(client, device, {'r': '0', 'g': '0', 'b': '0', 'intensity': '0'}))
+                        _wrap_retry(post_request)(client, device, {"r": "0", "g": "0", "b": "0", "intensity": "0"})
+                    )
                 else:
-                    data = {'r': options['<red>'], 'g': options['<green>'], 'b': options['<blue>'],
-                            'intensity': options['<intensity>']}
+                    data = {
+                        "r": options["<red>"],
+                        "g": options["<green>"],
+                        "b": options["<blue>"],
+                        "intensity": options["<intensity>"],
+                    }
                     coros.append(_wrap_retry(post_request)(client, device, data))
             else:
                 coros.append(_wrap_retry(get_status)(client, device))
@@ -184,16 +200,17 @@ async def _run(options: Dict[str, str]) -> None:
         sys.exit(1)
 
 
-def main(argv: Optional[List[str]] = None) -> None:
+def main(argv: Optional[list[str]] = None) -> None:
     """Run the CLI."""
     options = docopt(__doc__, version=__version__, argv=argv)
 
-    if options['--debug']:  # pragma: no cover
-        logging.basicConfig(level=logging.DEBUG,
-                            format='%(asctime)s %(levelname)-8s %(name)s:%(funcName)s: %(message)s')
+    if options["--debug"]:  # pragma: no cover
+        logging.basicConfig(
+            level=logging.DEBUG, format="%(asctime)s %(levelname)-8s %(name)s:%(funcName)s: %(message)s"
+        )
 
     asyncio.run(_run(options))
 
 
-if __name__ == '__main__':  # pragma: no cover
+if __name__ == "__main__":  # pragma: no cover
     main()