chore: add pyupgrade to ruff

pyproject.toml

@@ -54,6 +54,7 @@ build_command = "pip install poetry && poetry build"
 [tool.ruff]
 ignore = ["F403", "E741"]
 line-length = 120
+select=["E", "F", "UP"]
 
 [tool.black]
 line-length = 120

roborock/api.py

@@ -13,7 +13,8 @@
 import struct
 import time
 from random import randint
-from typing import Any, Callable, Coroutine, Optional, Type, TypeVar, final
+from typing import Any, TypeVar, final
+from collections.abc import Callable, Coroutine
 
 import aiohttp
 
@@ -89,7 +90,7 @@ def md5hex(message: str) -> str:
 
 
 class PreparedRequest:
-    def __init__(self, base_url: str, base_headers: Optional[dict] = None) -> None:
+    def __init__(self, base_url: str, base_headers: dict | None = None) -> None:
         self.base_url = base_url
         self.base_headers = base_headers or {}
 
@@ -251,7 +252,7 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
                                 data_protocol = RoborockDataProtocol(int(data_point_number))
                                 self._logger.debug(f"Got device update for {data_protocol.name}: {data_point}")
                                 if data_protocol in ROBOROCK_DATA_STATUS_PROTOCOL:
-                                    _cls: Type[Status] = ModelStatus.get(
+                                    _cls: type[Status] = ModelStatus.get(
                                         self.device_info.model, S7MaxVStatus
                                     )  # Default to S7 MAXV if we don't have the data
                                     if self.cache[CacheableAttribute.status].value is None:
@@ -301,7 +302,7 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
         except Exception as ex:
             self._logger.exception(ex)
 
-    def on_connection_lost(self, exc: Optional[Exception]) -> None:
+    def on_connection_lost(self, exc: Exception | None) -> None:
         self._last_disconnection = self.time_func()
         self._logger.info("Roborock client disconnected")
         if exc is not None:
@@ -340,7 +341,7 @@ def _async_response(
     def _get_payload(
         self,
         method: RoborockCommand,
-        params: Optional[list | dict] = None,
+        params: list | dict | None = None,
         secured=False,
     ):
         timestamp = math.floor(time.time())
@@ -372,16 +373,16 @@ async def send_message(self, roborock_message: RoborockMessage):
     async def _send_command(
         self,
         method: RoborockCommand,
-        params: Optional[list | dict] = None,
+        params: list | dict | None = None,
     ):
         raise NotImplementedError
 
     @final
     async def send_command(
         self,
         method: RoborockCommand,
-        params: Optional[list | dict] = None,
-        return_type: Optional[Type[RT]] = None,
+        params: list | dict | None = None,
+        return_type: type[RT] | None = None,
     ) -> RT:
         cacheable_attribute_result = find_cacheable_attribute(method)
 
@@ -404,7 +405,7 @@ async def send_command(
         return response
 
     async def get_status(self) -> Status | None:
-        _cls: Type[Status] = ModelStatus.get(
+        _cls: type[Status] = ModelStatus.get(
             self.device_info.model, S7MaxVStatus
         )  # Default to S7 MAXV if we don't have the data
         return _cls.from_dict(await self.cache[CacheableAttribute.status].async_value())

roborock/cli.py

@@ -3,7 +3,7 @@
 import json
 import logging
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
 
 import click
 from pyshark import FileCapture  # type: ignore
@@ -29,7 +29,7 @@ def __init__(self):
 
     def reload(self):
         if self.roborock_file.is_file():
-            with open(self.roborock_file, "r") as f:
+            with open(self.roborock_file) as f:
                 data = json.load(f)
                 if data:
                     self._login_data = LoginData.from_dict(data)
@@ -54,7 +54,7 @@ def login_data(self):
 @click.group()
 @click.pass_context
 def cli(ctx, debug: int):
-    logging_config: Dict[str, Any] = {"level": logging.DEBUG if debug > 0 else logging.INFO}
+    logging_config: dict[str, Any] = {"level": logging.DEBUG if debug > 0 else logging.INFO}
     logging.basicConfig(**logging_config)  # type: ignore
     ctx.obj = RoborockContext()
 
@@ -153,7 +153,7 @@ async def parser(_, local_key, device_ip, file):
     else:
         _LOGGER.info("Listen for interface rvi0 since no file was provided")
         capture = LiveCapture(interface="rvi0")
-    buffer = {"data": bytes()}
+    buffer = {"data": b""}
 
     def on_package(packet: Packet):
         if hasattr(packet, "ip"):

roborock/cloud_api.py

@@ -6,7 +6,7 @@
 import threading
 import uuid
 from asyncio import Lock, Task
-from typing import Any, Optional
+from typing import Any
 from urllib.parse import urlparse
 
 import paho.mqtt.client as mqtt
@@ -199,7 +199,7 @@ async def send_message(self, roborock_message: RoborockMessage):
     async def _send_command(
         self,
         method: RoborockCommand,
-        params: Optional[list | dict] = None,
+        params: list | dict | None = None,
     ):
         request_id, timestamp, payload = super()._get_payload(method, params, True)
         request_protocol = RoborockMessageProtocol.RPC_REQUEST

roborock/code_mappings.py

@@ -2,7 +2,6 @@
 
 import logging
 from enum import IntEnum
-from typing import Type
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -15,7 +14,7 @@ def name(self) -> str:
         return super().name.lower()
 
     @classmethod
-    def _missing_(cls: Type[RoborockEnum], key) -> RoborockEnum:
+    def _missing_(cls: type[RoborockEnum], key) -> RoborockEnum:
         if hasattr(cls, "unknown"):
             _LOGGER.warning(f"Missing {cls.__name__} code: {key} - defaulting to 'unknown'")
             return cls.unknown  # type: ignore
@@ -24,23 +23,23 @@ def _missing_(cls: Type[RoborockEnum], key) -> RoborockEnum:
         return default_value
 
     @classmethod
-    def as_dict(cls: Type[RoborockEnum]):
+    def as_dict(cls: type[RoborockEnum]):
         return {i.name: i.value for i in cls if i.name != "missing"}
 
     @classmethod
-    def as_enum_dict(cls: Type[RoborockEnum]):
+    def as_enum_dict(cls: type[RoborockEnum]):
         return {i.value: i for i in cls if i.name != "missing"}
 
     @classmethod
-    def values(cls: Type[RoborockEnum]) -> list[int]:
+    def values(cls: type[RoborockEnum]) -> list[int]:
         return list(cls.as_dict().values())
 
     @classmethod
-    def keys(cls: Type[RoborockEnum]) -> list[str]:
+    def keys(cls: type[RoborockEnum]) -> list[str]:
         return list(cls.as_dict().keys())
 
     @classmethod
-    def items(cls: Type[RoborockEnum]):
+    def items(cls: type[RoborockEnum]):
         return cls.as_dict().items()
 
 

roborock/command_cache.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Mapping, Optional
+from collections.abc import Mapping
 
 from roborock import RoborockCommand
 
@@ -38,9 +38,9 @@ class CacheableAttribute(str, Enum):
 class RoborockAttribute:
     attribute: str
     get_command: RoborockCommand
-    add_command: Optional[RoborockCommand] = None
-    set_command: Optional[RoborockCommand] = None
-    close_command: Optional[RoborockCommand] = None
+    add_command: RoborockCommand | None = None
+    set_command: RoborockCommand | None = None
+    close_command: RoborockCommand | None = None
     additional_change_commands: list[RoborockCommand] = field(default_factory=list)