Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions docs/_generate_requests_docstrings.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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()
Expand Down
4 changes: 2 additions & 2 deletions examples/usb/upgrade.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand Down
4 changes: 2 additions & 2 deletions smpclient/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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?
Expand Down
8 changes: 4 additions & 4 deletions smpclient/mcuboot.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
4 changes: 2 additions & 2 deletions smpclient/transport/_udp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)

Expand Down
4 changes: 2 additions & 2 deletions smpclient/transport/ble.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
3 changes: 1 addition & 2 deletions tests/extensions/test_intercreate.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
"""Test the Intercreate extensions."""

from pathlib import Path
from typing import List
from unittest.mock import PropertyMock, patch

import pytest
Expand Down Expand Up @@ -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`."""
Expand Down
4 changes: 2 additions & 2 deletions tests/test_requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -296,7 +296,7 @@
),
)
def test_requests(
test_tuple: Tuple[
test_tuple: tuple[
smpmsg.Request,
SMPRequest[TRep, TEr1, TEr2],
Type[smpmsg.Response],
Expand Down
5 changes: 2 additions & 3 deletions tests/test_smp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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`."""
Expand Down Expand Up @@ -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`."""
Expand Down
12 changes: 6 additions & 6 deletions tests/test_udp_client.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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)
)
Expand All @@ -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()
Expand All @@ -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)
Expand Down
Loading