From ffd7a6c649ce7d6904149b5593f4c77a530b4b7a Mon Sep 17 00:00:00 2001 From: Zhou Xiao Date: Tue, 7 Apr 2026 16:07:59 +0800 Subject: [PATCH] fix(ble_log_console): resolve mypy errors for Python <3.10 compatibility Add 'from __future__ import annotations' to defer annotation evaluation, replace @dataclass(slots=True) with @dataclass, use typing.Union for runtime type aliases, and suppress textual Screen metaclass false positive. --- tools/bt/ble_log_console/src/app.py | 90 +++++++++++++------ .../src/backend/frame_parser.py | 44 ++++++--- .../src/backend/internal_decoder.py | 17 ++-- .../bt/ble_log_console/src/backend/models.py | 26 +++--- .../src/backend/stats/sn_gap.py | 2 +- .../src/backend/stats/traffic_spike.py | 12 ++- .../src/backend/uart_transport.py | 7 +- .../src/frontend/launch_screen.py | 9 +- .../tests/test_traffic_spike.py | 1 + 9 files changed, 138 insertions(+), 70 deletions(-) diff --git a/tools/bt/ble_log_console/src/app.py b/tools/bt/ble_log_console/src/app.py index 92ec1291b96..d94c3dd36d3 100644 --- a/tools/bt/ble_log_console/src/app.py +++ b/tools/bt/ble_log_console/src/app.py @@ -1,10 +1,10 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Textual App wiring backend Worker to frontend widgets. See Spec Section 6. """ +from __future__ import annotations import struct import threading @@ -14,46 +14,45 @@ from pathlib import Path from typing import cast import serial -from textual.app import App -from textual.app import ComposeResult -from textual.binding import Binding -from textual.message import Message - from src.backend.frame_parser import FrameParser from src.backend.internal_decoder import decode_internal_frame -from src.backend.models import FRAME_OVERHEAD -from src.backend.models import LL_TS_OFFSET from src.backend.models import BackendStopped from src.backend.models import BleLogSource from src.backend.models import BufUtilEntry from src.backend.models import BufUtilResult from src.backend.models import EnhStatResult +from src.backend.models import FRAME_OVERHEAD from src.backend.models import FrameLossDetected from src.backend.models import FunnelSnapshot +from src.backend.models import has_os_ts from src.backend.models import InfoResult from src.backend.models import InternalFrameDecoded from src.backend.models import InternalSource +from src.backend.models import is_ll_source from src.backend.models import LaunchConfig +from src.backend.models import LL_TS_OFFSET from src.backend.models import LogLine from src.backend.models import LossType from src.backend.models import ParsedFrame +from src.backend.models import resolve_source_name from src.backend.models import SourcePeakWrite from src.backend.models import StatsUpdated from src.backend.models import SyncState from src.backend.models import SyncStateChanged from src.backend.models import TrafficSpikeDetected -from src.backend.models import has_os_ts -from src.backend.models import is_ll_source -from src.backend.models import resolve_source_name from src.backend.stats import StatsAccumulator -from src.backend.uart_transport import UART_BLOCK_SIZE from src.backend.uart_transport import open_serial +from src.backend.uart_transport import UART_BLOCK_SIZE from src.frontend.launch_screen import LaunchScreen from src.frontend.log_view import LogView from src.frontend.shortcut_screen import ShortcutScreen from src.frontend.stats_screen import BufUtilScreen from src.frontend.stats_screen import StatsScreen from src.frontend.status_panel import StatusPanel +from textual.app import App +from textual.app import ComposeResult +from textual.binding import Binding +from textual.message import Message STATS_INTERVAL = 0.25 # seconds @@ -115,7 +114,10 @@ class BLELogApp(App): self._resolve_output_path() self.run_worker(self._backend_loop, thread=True, exclusive=True) else: - self.push_screen(LaunchScreen(default_log_dir=self._log_dir), callback=self._on_launch_result) + self.push_screen( + LaunchScreen(default_log_dir=self._log_dir), + callback=self._on_launch_result, + ) @property def funnel_snapshots(self) -> list[FunnelSnapshot]: @@ -146,7 +148,9 @@ class BLELogApp(App): """Thread-safe message posting from backend worker.""" self.call_from_thread(self.post_message, msg) - def _emit_stats(self, stats: StatsAccumulator, parser: FrameParser, last_time: float) -> float: + def _emit_stats( + self, stats: StatsAccumulator, parser: FrameParser, last_time: float + ) -> float: """Emit a stats snapshot if the interval has elapsed. Returns updated timestamp.""" now = time.perf_counter() if now - last_time < STATS_INTERVAL: @@ -190,14 +194,18 @@ class BLELogApp(App): # Lazy file handles — created on first data arrival output_file = None console_log_file = None - console_log_path = self._output_path.with_name(self._output_path.stem + '_console.log') + console_log_path = self._output_path.with_name( + self._output_path.stem + '_console.log' + ) try: while True: with self._serial_lock: block = ser.read(UART_BLOCK_SIZE) if not block: - last_snapshot_time = self._emit_stats(stats, parser, last_snapshot_time) + last_snapshot_time = self._emit_stats( + stats, parser, last_snapshot_time + ) continue # 1. Save raw binary (lazy-open on first block) @@ -223,18 +231,31 @@ class BLELogApp(App): if isinstance(item, ParsedFrame): frame_size = len(item.payload) + FRAME_OVERHEAD if item.source_code != BleLogSource.INTERNAL: - stats.record_frame(frame_size, item.source_code, item.frame_sn) + stats.record_frame( + frame_size, item.source_code, item.frame_sn + ) stats.record_frame_traffic(frame_size, item.source_code) else: stats.record_frame() # count frame for transport metrics, no SN tracking - if has_os_ts(item.source_code) and item.source_code != BleLogSource.INTERNAL: - stats.record_frame_ts(item.os_ts_ms, frame_size, item.source_code) + if ( + has_os_ts(item.source_code) + and item.source_code != BleLogSource.INTERNAL + ): + stats.record_frame_ts( + item.os_ts_ms, frame_size, item.source_code + ) elif is_ll_source(item.source_code) and len(item.payload) >= 6: - (lc_ts_us,) = struct.unpack_from(' 0: - source_name = resolve_source_name(enh['src_code']) + source_name = resolve_source_name( + enh['src_code'] + ) self._post( FrameLossDetected( source_name, @@ -290,7 +316,9 @@ class BLELogApp(App): # A single log line may span multiple frames due to # batch sealing, so buffer partial lines until '\n'. elif item.source_code == BleLogSource.REDIR: - payload_text = item.payload.decode('ascii', errors='replace') + payload_text = item.payload.decode( + 'ascii', errors='replace' + ) # Write raw payload to console log (independent of line buffering) if console_log_file is None: @@ -356,7 +384,9 @@ class BLELogApp(App): if msg.int_src == InternalSource.INIT_DONE: info = cast(InfoResult, msg.payload) log_view = self.query_one(LogView) - log_view.write_info(f'BLE Log v{info["version"]} initialized - statistics reset') + log_view.write_info( + f"BLE Log v{info['version']} initialized - statistics reset" + ) elif msg.int_src == InternalSource.FLUSH: log_view = self.query_one(LogView) log_view.write_info('Firmware flush - SN counters reset') @@ -379,13 +409,17 @@ class BLELogApp(App): def on_traffic_spike_detected(self, msg: TrafficSpikeDetected) -> None: top_sources = sorted(msg.per_source.items(), key=lambda x: x[1], reverse=True) - src_parts = ', '.join(f'{resolve_source_name(s)} {p:.0f}%' for s, p in top_sources if p >= 1.0) + src_parts = ', '.join( + f'{resolve_source_name(s)} {p:.0f}%' for s, p in top_sources if p >= 1.0 + ) if msg.utilization_pct >= 100.0: util_str = 'saturated' else: util_str = f'{msg.utilization_pct:.0f}% wire' log_view = self.query_one(LogView) - log_view.write_traffic(f'{msg.throughput_kbs:.0f} KB/s ({util_str}) over {msg.duration_ms:.0f}ms | {src_parts}') + log_view.write_traffic( + f'{msg.throughput_kbs:.0f} KB/s ({util_str}) over {msg.duration_ms:.0f}ms | {src_parts}' + ) # --- Actions --- diff --git a/tools/bt/ble_log_console/src/backend/frame_parser.py b/tools/bt/ble_log_console/src/backend/frame_parser.py index e21ba4bcbff..fafee06599a 100644 --- a/tools/bt/ble_log_console/src/backend/frame_parser.py +++ b/tools/bt/ble_log_console/src/backend/frame_parser.py @@ -1,25 +1,25 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Frame sync state machine with checksum auto-detection. Parses BLE Log module frames from a raw byte stream. See Spec Sections 7, 8. """ +from __future__ import annotations from collections.abc import Callable from src.backend.checksum import sum_checksum from src.backend.checksum import xor_checksum from src.backend.models import CHECKSUM_STRUCT +from src.backend.models import ChecksumAlgorithm +from src.backend.models import ChecksumMode +from src.backend.models import ChecksumScope from src.backend.models import FRAME_HEADER_SIZE from src.backend.models import FRAME_OVERHEAD from src.backend.models import HEADER_STRUCT from src.backend.models import MAX_FRAME_SIZE from src.backend.models import MAX_REMAINDER_SIZE -from src.backend.models import ChecksumAlgorithm -from src.backend.models import ChecksumMode -from src.backend.models import ChecksumScope from src.backend.models import ParsedFrame from src.backend.models import SyncState @@ -85,7 +85,10 @@ class FrameParser: results.append(frame) offset = next_offset self._on_frame_found(mode) - elif self._sync_state == SyncState.CONFIRMING_SYNC and self._might_be_incomplete_frame(buf, offset): + elif ( + self._sync_state == SyncState.CONFIRMING_SYNC + and self._might_be_incomplete_frame(buf, offset) + ): break elif ( self._sync_state == SyncState.SEARCHING @@ -95,7 +98,7 @@ class FrameParser: break else: if not self._ever_synced: - self._collect_ascii(buf[offset : offset + 1], results) + self._collect_ascii(buf[offset:offset + 1], results) offset += 1 else: # SYNCED or CONFIRMING_LOSS: use locked checksum mode @@ -145,8 +148,10 @@ class FrameParser: if offset + FRAME_OVERHEAD + payload_len > len(buf): return None - header = buf[offset : offset + FRAME_HEADER_SIZE] - payload = buf[offset + FRAME_HEADER_SIZE : offset + FRAME_HEADER_SIZE + payload_len] + header = buf[offset:offset + FRAME_HEADER_SIZE] + payload = buf[ + offset + FRAME_HEADER_SIZE:offset + FRAME_HEADER_SIZE + payload_len + ] checksum_offset = offset + FRAME_HEADER_SIZE + payload_len stored_checksum = CHECKSUM_STRUCT.unpack_from(buf, checksum_offset)[0] @@ -177,7 +182,9 @@ class FrameParser: next_offset = offset + FRAME_OVERHEAD + payload_len return frame, next_offset - def _try_parse_with_probe(self, buf: bytes, offset: int) -> tuple[ParsedFrame, int, ChecksumMode] | None: + def _try_parse_with_probe( + self, buf: bytes, offset: int + ) -> tuple[ParsedFrame, int, ChecksumMode] | None: """Try all checksum combinations at the given offset (SEARCHING mode).""" for algo, scope, fn in _CHECKSUM_PROBES: result = self._try_parse_at(buf, offset, fn, scope) @@ -187,11 +194,17 @@ class FrameParser: return frame, next_offset, mode return None - def _try_parse_locked(self, buf: bytes, offset: int) -> tuple[ParsedFrame, int] | None: + def _try_parse_locked( + self, buf: bytes, offset: int + ) -> tuple[ParsedFrame, int] | None: """Try to parse with the locked checksum mode.""" if self._checksum_mode is None: return None - fn = xor_checksum if self._checksum_mode.algorithm == ChecksumAlgorithm.XOR else sum_checksum + fn = ( + xor_checksum + if self._checksum_mode.algorithm == ChecksumAlgorithm.XOR + else sum_checksum + ) return self._try_parse_at(buf, offset, fn, self._checksum_mode.scope) def _on_frame_found(self, mode: ChecksumMode) -> None: @@ -241,7 +254,10 @@ class FrameParser: return True if remaining >= FRAME_HEADER_SIZE: payload_len, _ = HEADER_STRUCT.unpack_from(buf, offset) - if payload_len <= MAX_FRAME_SIZE and remaining < FRAME_OVERHEAD + payload_len: + if ( + payload_len <= MAX_FRAME_SIZE + and remaining < FRAME_OVERHEAD + payload_len + ): return True return False @@ -250,7 +266,9 @@ class FrameParser: self._ever_synced = True self._sync_state = new_state - def _collect_ascii(self, byte_data: bytes, results: list[ParsedFrame | str]) -> None: + def _collect_ascii( + self, byte_data: bytes, results: list[ParsedFrame | str] + ) -> None: """Collect bytes for ASCII line assembly. Only printable ASCII (0x20-0x7E) and newline (0x0A) are collected. diff --git a/tools/bt/ble_log_console/src/backend/internal_decoder.py b/tools/bt/ble_log_console/src/backend/internal_decoder.py index 874211df3f1..ee23163f821 100644 --- a/tools/bt/ble_log_console/src/backend/internal_decoder.py +++ b/tools/bt/ble_log_console/src/backend/internal_decoder.py @@ -1,11 +1,11 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Decode BLE_LOG_SRC_INTERNAL(0) frame payloads. Payload format on wire: [4B os_ts][1B int_src_code][variable sub-payload] See Spec Section 9. """ +from __future__ import annotations import struct @@ -65,9 +65,14 @@ def decode_internal_frame(payload: bytes) -> InternalDecoderResult | None: if int_src == InternalSource.ENH_STAT: if len(sub_payload) < _ENH_STAT_STRUCT.size: return None - _, src_code, written_frame_cnt, lost_frame_cnt, written_bytes_cnt, lost_bytes_cnt = ( - _ENH_STAT_STRUCT.unpack_from(sub_payload, 0) - ) + ( + _, + src_code, + written_frame_cnt, + lost_frame_cnt, + written_bytes_cnt, + lost_bytes_cnt, + ) = _ENH_STAT_STRUCT.unpack_from(sub_payload, 0) return EnhStatResult( int_src=int_src, src_code=src_code, @@ -81,7 +86,9 @@ def decode_internal_frame(payload: bytes) -> InternalDecoderResult | None: if int_src == InternalSource.BUF_UTIL: if len(sub_payload) < _BUF_UTIL_STRUCT.size: return None - _, lbm_id, trans_cnt, inflight_peak = _BUF_UTIL_STRUCT.unpack_from(sub_payload, 0) + _, lbm_id, trans_cnt, inflight_peak = _BUF_UTIL_STRUCT.unpack_from( + sub_payload, 0 + ) pool = (lbm_id >> 4) & 0x0F index = lbm_id & 0x0F return BufUtilResult( diff --git a/tools/bt/ble_log_console/src/backend/models.py b/tools/bt/ble_log_console/src/backend/models.py index 6eb46e63738..46c2f8fc2d2 100644 --- a/tools/bt/ble_log_console/src/backend/models.py +++ b/tools/bt/ble_log_console/src/backend/models.py @@ -1,5 +1,6 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations import struct from dataclasses import dataclass @@ -7,6 +8,7 @@ from dataclasses import field from enum import Enum from pathlib import Path from typing import TypedDict +from typing import Union from textual.message import Message @@ -87,7 +89,9 @@ _NO_OS_TS_SOURCES: frozenset[int] = frozenset( ) -_LL_SOURCES: frozenset[int] = frozenset({BleLogSource.LL_TASK, BleLogSource.LL_HCI, BleLogSource.LL_ISR}) +_LL_SOURCES: frozenset[int] = frozenset( + {BleLogSource.LL_TASK, BleLogSource.LL_HCI, BleLogSource.LL_ISR} +) LL_TS_OFFSET = 2 # lc_ts starts at payload[2:6] LL_TS_SIZE = 4 @@ -130,13 +134,13 @@ class BufUtilPool(int, Enum): # --- Data classes --- -@dataclass(slots=True) +@dataclass class ChecksumMode: algorithm: ChecksumAlgorithm scope: ChecksumScope -@dataclass(slots=True) +@dataclass class ParsedFrame: source_code: int frame_sn: int @@ -144,7 +148,7 @@ class ParsedFrame: os_ts_ms: int # extracted from first 4 bytes of payload; only valid when has_os_ts(source_code) is True -@dataclass(slots=True) +@dataclass class SourcePeakWrite: """Peak write burst for a single source within a 1ms window.""" @@ -152,7 +156,7 @@ class SourcePeakWrite: peak_bytes: int = 0 # total bytes in that same window -@dataclass(slots=True) +@dataclass class SourceStats: """Console-side accumulated per-source statistics (resilient to firmware counter resets).""" @@ -202,7 +206,7 @@ def resolve_lbm_name(pool: int, index: int) -> str: return f'lbm_{pool}_{index}' -@dataclass(slots=True) +@dataclass class TransportSnapshot: """Snapshot of transport-layer metrics for the current stats interval.""" @@ -212,7 +216,7 @@ class TransportSnapshot: fps: float = 0.0 -@dataclass(slots=True) +@dataclass class LossSnapshot: """Snapshot of firmware-reported cumulative loss.""" @@ -220,7 +224,7 @@ class LossSnapshot: total_bytes: int = 0 -@dataclass(slots=True) +@dataclass class PeakBurstSnapshot: """Peak write burst metrics for a single clock domain (os_ts or lc_ts).""" @@ -271,7 +275,7 @@ class FunnelSnapshot: throughput: ThroughputInfo -@dataclass(slots=True) +@dataclass class LaunchConfig: """Configuration returned by the Launch Screen.""" @@ -280,7 +284,7 @@ class LaunchConfig: log_dir: Path -@dataclass(slots=True) +@dataclass class FrameStats: """Periodic stats snapshot with metrics grouped by dimension.""" @@ -323,7 +327,7 @@ class BufUtilResult(TypedDict): os_ts_ms: int -InternalDecoderResult = InfoResult | EnhStatResult | BufUtilResult +InternalDecoderResult = Union[InfoResult, EnhStatResult, BufUtilResult] # --- Textual Messages (backend -> frontend) --- diff --git a/tools/bt/ble_log_console/src/backend/stats/sn_gap.py b/tools/bt/ble_log_console/src/backend/stats/sn_gap.py index fb2559301f2..5f9b71dbe6a 100644 --- a/tools/bt/ble_log_console/src/backend/stats/sn_gap.py +++ b/tools/bt/ble_log_console/src/backend/stats/sn_gap.py @@ -1,12 +1,12 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Sliding receive window gap tracker for per-source frame sequence numbers. Frames are only declared lost when the receive window advances past their SN without them being received, tolerating out-of-order delivery up to REORDER_WINDOW frames. """ +from __future__ import annotations from src.backend.models import SourceCode diff --git a/tools/bt/ble_log_console/src/backend/stats/traffic_spike.py b/tools/bt/ble_log_console/src/backend/stats/traffic_spike.py index a11ac0fdd68..2fc6dcea090 100644 --- a/tools/bt/ble_log_console/src/backend/stats/traffic_spike.py +++ b/tools/bt/ble_log_console/src/backend/stats/traffic_spike.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Real-time traffic spike detection using a sliding window over wall-clock time.""" +from __future__ import annotations import time from collections import deque @@ -14,7 +14,7 @@ TRAFFIC_THRESHOLD_PCT = 0.8 # 80% of wire max TRAFFIC_ALERT_COOLDOWN_SEC = 2.0 # minimum interval between alerts -@dataclass(slots=True) +@dataclass class TrafficSpikeResult: throughput_kbs: float wire_max_kbs: float @@ -41,7 +41,9 @@ class TrafficSpikeDetector: def record(self, frame_size: int, src_code: SourceCode) -> None: self._window.append((time.perf_counter(), frame_size, src_code)) if self._spike_active: - self._spike_per_source[src_code] = self._spike_per_source.get(src_code, 0) + frame_size + self._spike_per_source[src_code] = ( + self._spike_per_source.get(src_code, 0) + frame_size + ) def check(self) -> TrafficSpikeResult | None: now = time.perf_counter() @@ -83,7 +85,9 @@ class TrafficSpikeDetector: spike_bps = self._spike_peak_bps src_total = max(sum(self._spike_per_source.values()), 1) - src_pcts = {src: v / src_total * 100.0 for src, v in self._spike_per_source.items()} + src_pcts = { + src: v / src_total * 100.0 for src, v in self._spike_per_source.items() + } return TrafficSpikeResult( throughput_kbs=spike_bps / 1024.0, diff --git a/tools/bt/ble_log_console/src/backend/uart_transport.py b/tools/bt/ble_log_console/src/backend/uart_transport.py index bb40364856b..b2a8d4f21c6 100644 --- a/tools/bt/ble_log_console/src/backend/uart_transport.py +++ b/tools/bt/ble_log_console/src/backend/uart_transport.py @@ -1,12 +1,11 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """UART read loop with raw binary file writing. See Spec Sections 6, 12. """ +from __future__ import annotations -import serial import serial.tools.list_ports UART_READ_TIMEOUT = 0.1 @@ -28,6 +27,8 @@ def validate_uart_port(port: str) -> str | None: def open_serial(port: str, baudrate: int) -> serial.Serial: try: - return serial.Serial(port, baudrate=baudrate, timeout=UART_READ_TIMEOUT, exclusive=True) + return serial.Serial( + port, baudrate=baudrate, timeout=UART_READ_TIMEOUT, exclusive=True + ) except (ValueError, serial.SerialException): return serial.Serial(port, baudrate=baudrate, timeout=UART_READ_TIMEOUT) diff --git a/tools/bt/ble_log_console/src/frontend/launch_screen.py b/tools/bt/ble_log_console/src/frontend/launch_screen.py index 8fffc199ad9..928b93a0881 100644 --- a/tools/bt/ble_log_console/src/frontend/launch_screen.py +++ b/tools/bt/ble_log_console/src/frontend/launch_screen.py @@ -1,14 +1,16 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Launch Screen — interactive setup for port, baud rate, and log directory. Shown on startup when --port is not provided via CLI. Dismissed with a LaunchConfig result on Connect, or None on quit. """ +from __future__ import annotations from pathlib import Path +from src.backend.models import LaunchConfig +from src.backend.uart_transport import list_serial_ports from textual import on from textual import work from textual.app import ComposeResult @@ -22,14 +24,11 @@ from textual.widgets import Label from textual.widgets import Select from textual_fspicker import SelectDirectory -from src.backend.models import LaunchConfig -from src.backend.uart_transport import list_serial_ports - BAUD_RATES: list[int] = [115200, 230400, 460800, 921600, 1500000, 2000000, 3000000] DEFAULT_BAUD_RATE: int = 3000000 -class LaunchScreen(Screen[LaunchConfig | None]): +class LaunchScreen(Screen[LaunchConfig | None]): # type: ignore[misc] """Interactive setup screen for BLE Log Console.""" DEFAULT_CSS = """ diff --git a/tools/bt/ble_log_console/tests/test_traffic_spike.py b/tools/bt/ble_log_console/tests/test_traffic_spike.py index ec321269045..3994599dc1b 100644 --- a/tools/bt/ble_log_console/tests/test_traffic_spike.py +++ b/tools/bt/ble_log_console/tests/test_traffic_spike.py @@ -1,5 +1,6 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations from unittest.mock import patch