Merge pull request #86 from gudnimg/generic-types-gudni

Author: JPHutchins

docs/_generate_requests_docstrings.py

@@ -13,7 +13,7 @@
 import importlib.util
 import inspect
 import os
-from typing import Any, List, Optional, Type
+from typing import Any, Optional, Type
 
 from pydantic import BaseModel
 
@@ -71,11 +71,11 @@ def format_type(annotation: Type[Any] | None) -> str:
     if annotation is None:
         raise ValueError("Annotation cannot be None")
     if hasattr(annotation, '__name__'):  # Handles regular types like `int`, `str`, etc.
-        # get the annotations like List[str] for example
+        # get the annotations like list[str] for example
         if hasattr(annotation, '__args__'):
             return f"{annotation.__name__}[{format_type(annotation.__args__[0])}]"
         return f"{annotation.__name__}"
-    elif hasattr(annotation, '__origin__'):  # Handles generic types like List[str], Optional[int]
+    elif hasattr(annotation, '__origin__'):  # Handles generic types like list[str], Optional[int]
         return f"{annotation.__origin__.__module__}.{annotation.__origin__.__name__}"
     return str(annotation)  # Fallback for other types
 
@@ -108,7 +108,7 @@ def get_pydantic_fields(cls: Type[BaseModel]) -> str:
     return args
 
 
-def parse_file(file_path: str) -> List[ClassInfo]:
+def parse_file(file_path: str) -> list[ClassInfo]:
     """Parse the file and extract class definitions."""
     with open(file_path, 'r') as file:
         lines = file.readlines()

examples/usb/upgrade.py

@@ -7,7 +7,7 @@
 import subprocess
 import time
 from pathlib import Path
-from typing import Final, Tuple
+from typing import Final
 
 from serial.tools.list_ports import comports
 from smp import error as smperr
@@ -215,7 +215,7 @@ async def ensure_request(request: SMPRequest[TRep, TEr1, TEr2]) -> TRep:
             raise SystemExit(f"Unknown response: {images}")
 
 
-def get_runner_command(board: str, hex_path: Path) -> Tuple[str, ...]:
+def get_runner_command(board: str, hex_path: Path) -> tuple[str, ...]:
     if "nrf" in board:
         print("Using the nrfjprog runner")
         return ("nrfjprog", "--recover", "--reset", "--verify", "--program", str(hex_path))

smpclient/__init__.py

@@ -30,7 +30,7 @@
 import traceback
 from hashlib import sha256
 from types import TracebackType
-from typing import AsyncIterator, Final, Tuple, Type
+from typing import AsyncIterator, Final, Type
 
 from pydantic import ValidationError
 from smp import header as smpheader
@@ -417,7 +417,7 @@ def _cbor_integer_size(integer: int) -> int:
         # https://datatracker.ietf.org/doc/html/rfc8949#name-core-deterministic-encoding
         return 0 if integer < 24 else 1 if integer <= 0xFF else 2 if integer <= 0xFFFF else 4
 
-    def _get_max_cbor_and_data_size(self, request: smpmsg.WriteRequest) -> Tuple[int, int]:
+    def _get_max_cbor_and_data_size(self, request: smpmsg.WriteRequest) -> tuple[int, int]:
         """Given an `ImageUploadWrite`, return the maximum CBOR size and data size."""
 
         # given empty data in the request, how many bytes are available for the data?

smpclient/mcuboot.py

@@ -11,7 +11,7 @@
 from enum import IntEnum, IntFlag, unique
 from functools import cached_property
 from io import BufferedReader, BytesIO
-from typing import Annotated, Any, Dict, Final, List, Union
+from typing import Annotated, Any, Final, Union
 
 from intelhex import hex2bin  # type: ignore
 from pydantic import Field, GetCoreSchemaHandler
@@ -295,7 +295,7 @@ class ImageInfo:
 
     header: ImageHeader
     tlv_info: ImageTLVInfo
-    tlvs: List[ImageTLVValue]
+    tlvs: list[ImageTLVValue]
     file: str | None = None
 
     def get_tlv(self, tlv: ImageTLVType) -> ImageTLVValue:
@@ -332,15 +332,15 @@ def load_file(path: str) -> 'ImageInfo':
         f.seek(tlv_offset)  # move to the start of the TLV area
         tlv_info = ImageTLVInfo.load_from(f)
 
-        tlvs: List[ImageTLVValue] = []
+        tlvs: list[ImageTLVValue] = []
         while f.tell() < tlv_offset + tlv_info.tlv_tot:
             tlv_header = ImageTLV.load_from(f)
             tlvs.append(ImageTLVValue(header=tlv_header, value=f.read(tlv_header.len)))
 
         return ImageInfo(file=path, header=image_header, tlv_info=tlv_info, tlvs=tlvs)
 
     @cached_property
-    def _map_tlv_type_to_value(self) -> Dict[int, ImageTLVValue]:
+    def _map_tlv_type_to_value(self) -> dict[int, ImageTLVValue]:
         return {tlv.header.type: tlv for tlv in self.tlvs}
 
     def __str__(self) -> str:

smpclient/transport/_udp_client.py

@@ -4,7 +4,7 @@
 
 import asyncio
 import logging
-from typing import Any, Final, NamedTuple, Tuple
+from typing import Any, Final, NamedTuple
 
 from typing_extensions import override
 
@@ -79,7 +79,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         logger.debug(f"Connection made, {transport=}")
 
     @override
-    def datagram_received(self, data: bytes, addr: Tuple[str | Any, int]) -> None:
+    def datagram_received(self, data: bytes, addr: tuple[str | Any, int]) -> None:
         logger.debug(f"{len(data)} B datagram received from {addr}")
         self._receive_queue.put_nowait(data)
 

smpclient/transport/ble.py

@@ -6,7 +6,7 @@
 import logging
 import re
 import sys
-from typing import Final, List, Protocol
+from typing import Final, Protocol
 from uuid import UUID
 
 from bleak import BleakClient, BleakGATTCharacteristic, BleakScanner
@@ -198,7 +198,7 @@ def mtu(self) -> int:
         return self._max_write_without_response_size
 
     @staticmethod
-    async def scan(timeout: int = 5) -> List[BLEDevice]:
+    async def scan(timeout: int = 5) -> list[BLEDevice]:
         """Scan for BLE devices."""
         logger.debug(f"Scanning for BLE devices for {timeout} seconds")
         devices: Final = await BleakScanner(service_uuids=[str(SMP_SERVICE_UUID)]).discover(

tests/extensions/test_intercreate.py

@@ -1,7 +1,6 @@
 """Test the Intercreate extensions."""
 
 from pathlib import Path
-from typing import List
 from unittest.mock import PropertyMock, patch
 
 import pytest
@@ -29,7 +28,7 @@ async def test_upload_hello_world_bin_encoded(mock_mtu: PropertyMock) -> None:
     assert s._transport.mtu == 127
     assert s._transport.max_unencoded_size < 127
 
-    packets: List[bytes] = []
+    packets: list[bytes] = []
 
     def mock_write(data: bytes) -> int:
         """Accumulate the raw packets in the global `packets`."""

tests/test_requests.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Tuple, Type
+from typing import Type
 
 import pytest
 from smp import enumeration_management as smpem
@@ -296,7 +296,7 @@
     ),
 )
 def test_requests(
-    test_tuple: Tuple[
+    test_tuple: tuple[
         smpmsg.Request,
         SMPRequest[TRep, TEr1, TEr2],
         Type[smpmsg.Response],

tests/test_smp_client.py

@@ -5,7 +5,6 @@
 import sys
 from hashlib import sha256
 from pathlib import Path
-from typing import List
 from unittest.mock import AsyncMock, PropertyMock, call, patch
 
 import pytest
@@ -341,7 +340,7 @@ async def test_upload_hello_world_bin_encoded(
     s = SMPClient(m, "address")
     assert s._transport.mtu == max_smp_encoded_frame_size
 
-    packets: List[bytes] = []
+    packets: list[bytes] = []
 
     def mock_write(data: bytes) -> int:
         """Accumulate the raw packets in the global `packets`."""
@@ -572,7 +571,7 @@ async def test_file_upload_test_encoded(max_smp_encoded_frame_size: int, line_bu
     s = SMPClient(m, "address")
     assert s._transport.mtu == max_smp_encoded_frame_size
 
-    packets: List[bytes] = []
+    packets: list[bytes] = []
 
     def mock_write(data: bytes) -> int:
         """Accumulate the raw packets in the global `packets`."""

tests/test_udp_client.py

@@ -1,7 +1,7 @@
 """Test the generic UDP client implementation."""
 
 import asyncio
-from typing import List, Tuple, cast
+from typing import cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -82,14 +82,14 @@ class _ServerProtocol(asyncio.DatagramProtocol):
     """A mock SMP server protocol for unit testing."""
 
     def __init__(self) -> None:
-        self.datagrams_recieved: List[bytes] = []
+        self.datagrams_recieved: list[bytes] = []
 
-    def datagram_received(self, data: bytes, addr: Tuple[str, int]) -> None:
+    def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
         self.datagrams_recieved.append(data)
 
 
 @pytest_asyncio.fixture
-async def udp_server() -> AsyncGenerator[Tuple[asyncio.DatagramTransport, _ServerProtocol], None]:
+async def udp_server() -> AsyncGenerator[tuple[asyncio.DatagramTransport, _ServerProtocol], None]:
     transport, protocol = await asyncio.get_running_loop().create_datagram_endpoint(
         lambda: _ServerProtocol(), local_addr=("127.0.0.1", 1337)
     )
@@ -100,7 +100,7 @@ async def udp_server() -> AsyncGenerator[Tuple[asyncio.DatagramTransport, _Serve
 
 
 @pytest.mark.asyncio
-async def test_send(udp_server: Tuple[asyncio.DatagramTransport, _ServerProtocol]) -> None:
+async def test_send(udp_server: tuple[asyncio.DatagramTransport, _ServerProtocol]) -> None:
     _, p = udp_server
 
     c = UDPClient()
@@ -113,7 +113,7 @@ async def test_send(udp_server: Tuple[asyncio.DatagramTransport, _ServerProtocol
 
 
 @pytest.mark.asyncio
-async def test_receive(udp_server: Tuple[asyncio.DatagramTransport, _ServerProtocol]) -> None:
+async def test_receive(udp_server: tuple[asyncio.DatagramTransport, _ServerProtocol]) -> None:
     t, _ = udp_server
 
     CLIENT_ADDR = Addr("127.0.0.1", 1338)