fix(ble): support Python 3.9 type annotations

This commit is contained in:
Zhou Xiao
2026-04-28 12:15:27 +08:00
parent 11447dd9ad
commit 4ceb950dc1
4 changed files with 23 additions and 9 deletions
+5 -2
View File
@@ -3,7 +3,10 @@
from __future__ import annotations
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
from typing import Union
from .console import BLEUARTConsole
from .console import ConsoleEncoding
@@ -12,8 +15,8 @@ from .console import ConsoleTerminator
def run_console(
device_id: str,
terminator: ConsoleTerminator | str = ConsoleTerminator.lf,
encoding: ConsoleEncoding | str = ConsoleEncoding.text,
terminator: Union[ConsoleTerminator, str] = ConsoleTerminator.lf,
encoding: Union[ConsoleEncoding, str] = ConsoleEncoding.text,
with_response: bool = False,
) -> None:
# Initialize BLE UART Console
@@ -3,9 +3,12 @@
from __future__ import annotations
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
import codecs
from enum import Enum
from typing import Union
from loguru import logger
from rich.highlighter import Highlighter
@@ -93,8 +96,8 @@ class BLEUARTConsole(App):
def __init__(
self,
device_id: str,
terminator: ConsoleTerminator | str = ConsoleTerminator.lf,
encoding: ConsoleEncoding | str = ConsoleEncoding.text,
terminator: Union[ConsoleTerminator, str] = ConsoleTerminator.lf,
encoding: Union[ConsoleEncoding, str] = ConsoleEncoding.text,
with_response: bool = False,
) -> None:
super().__init__()
@@ -109,7 +112,7 @@ class BLEUARTConsole(App):
self._ui_ready = False
@staticmethod
def _parse_terminator(terminator: ConsoleTerminator | str) -> bytes:
def _parse_terminator(terminator: Union[ConsoleTerminator, str]) -> bytes:
try:
return TERMINATORS[ConsoleTerminator(terminator)]
except ValueError:
@@ -117,7 +120,7 @@ class BLEUARTConsole(App):
raise ValueError(f'Unsupported terminator: {terminator}. Expected one of: {choices}') from None
@staticmethod
def _parse_encoding(encoding: ConsoleEncoding | str) -> ConsoleEncoding:
def _parse_encoding(encoding: Union[ConsoleEncoding, str]) -> ConsoleEncoding:
try:
return ConsoleEncoding(encoding)
except ValueError:
+5 -1
View File
@@ -1,8 +1,12 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import json
from typing import Any
from typing import Optional
from urllib.error import HTTPError
from urllib.error import URLError
from urllib.request import Request
@@ -20,7 +24,7 @@ def _daemon_url(host: str, port: int, path: str) -> str:
def _request_json(
method: str,
url: str,
payload: dict[str, Any] | None = None,
payload: Optional[dict[str, Any]] = None,
timeout: float = 10.0,
) -> dict[str, Any]:
data = json.dumps(payload).encode() if payload is not None else None
@@ -1,10 +1,14 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Optional
from uuid import uuid4
from fastapi import FastAPI
@@ -66,8 +70,8 @@ def _request_data_size(data: object) -> int:
@app.get('/status')
async def status() -> dict:
bridge: BLEUARTBridge | None = getattr(app.state, 'bridge', None)
pending_requests: dict | None = getattr(app.state, 'pending_requests', None)
bridge: Optional[BLEUARTBridge] = getattr(app.state, 'bridge', None)
pending_requests: Optional[dict] = getattr(app.state, 'pending_requests', None)
return {
'device_id': getattr(app.state, 'device_id', None),