diff --git a/tools/bt/ble_log_console/conftest.py b/tools/bt/ble_log_console/conftest.py index 0ad3a62cc77..272a2fee7b7 100644 --- a/tools/bt/ble_log_console/conftest.py +++ b/tools/bt/ble_log_console/conftest.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - import sys from pathlib import Path diff --git a/tools/bt/ble_log_console/src/__init__.py b/tools/bt/ble_log_console/src/__init__.py index 3609c82fad9..01d90b52da3 100644 --- a/tools/bt/ble_log_console/src/__init__.py +++ b/tools/bt/ble_log_console/src/__init__.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - import sys if sys.version_info < (3, 10): # noqa: UP036 — runtime guard for users on old Python diff --git a/tools/bt/ble_log_console/src/app.py b/tools/bt/ble_log_console/src/app.py index 92ec1291b96..9772dc7e104 100644 --- a/tools/bt/ble_log_console/src/app.py +++ b/tools/bt/ble_log_console/src/app.py @@ -1,59 +1,57 @@ # 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. """ - import struct import threading import time from datetime import datetime from pathlib import Path from typing import cast +from typing import Optional 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 @@ -85,21 +83,21 @@ class BLELogApp(App): def __init__( self, - port: str | None = None, + port: Optional[str] = None, baudrate: int = 3_000_000, - log_dir: Path | None = None, + log_dir: Optional[Path] = None, ) -> None: super().__init__() self._port = port self._baudrate = baudrate self._log_dir = log_dir or Path.cwd() - self._output_path: Path | None = None - self._serial: serial.Serial | None = None + self._output_path: Optional[Path] = None + self._serial: Optional[serial.Serial] = None # All-time per-source chip write peak (updated from StatsUpdated messages) - self._max_per_source_peak: dict[int, SourcePeakWrite] | None = None - self._ll_max_per_source_peak: dict[int, SourcePeakWrite] | None = None + self._max_per_source_peak: Optional[dict[int, SourcePeakWrite]] = None + self._ll_max_per_source_peak: Optional[dict[int, SourcePeakWrite]] = None # Console-side per-source received bytes (from StatsUpdated snapshots) - self._per_source_rx_bytes: dict[int, int] | None = None + self._per_source_rx_bytes: Optional[dict[int, int]] = None self._funnel_snapshots: list[FunnelSnapshot] = [] self._buf_util_snapshots: list[BufUtilEntry] = [] # Wall-clock capture start (set when backend loop begins) @@ -115,7 +113,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]: @@ -125,7 +126,7 @@ class BLELogApp(App): def buf_util_snapshots(self) -> list[BufUtilEntry]: return self._buf_util_snapshots - def _on_launch_result(self, config: LaunchConfig | None) -> None: + def _on_launch_result(self, config: Optional[LaunchConfig]) -> None: """Handle Launch Screen dismissal.""" if config is None: self.exit() @@ -146,7 +147,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 +193,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 +230,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 +315,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 +383,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 +408,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/checksum.py b/tools/bt/ble_log_console/src/backend/checksum.py index 227193cef71..634076634c2 100644 --- a/tools/bt/ble_log_console/src/backend/checksum.py +++ b/tools/bt/ble_log_console/src/backend/checksum.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Checksum implementations matching BLE Log firmware (ble_log_util.c). Two algorithms: @@ -11,7 +10,6 @@ The firmware's ror32 alignment compensation makes the XOR checksum alignment-independent — simple word-by-word XOR produces the same result regardless of the original buffer alignment. """ - import struct 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..7aae9bbf41d 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 collections.abc import Callable +from typing import Optional +from typing import Union 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 @@ -42,7 +42,7 @@ class FrameParser: def __init__(self) -> None: self._remained = b'' self._sync_state = SyncState.SEARCHING - self._checksum_mode: ChecksumMode | None = None + self._checksum_mode: Optional[ChecksumMode] = None self._confirm_count = 0 self._loss_count = 0 self._ascii_buffer = '' @@ -53,10 +53,10 @@ class FrameParser: return self._sync_state @property - def checksum_mode(self) -> ChecksumMode | None: + def checksum_mode(self) -> Optional[ChecksumMode]: return self._checksum_mode - def feed(self, data: bytes) -> list[ParsedFrame | str]: + def feed(self, data: bytes) -> list[Union[ParsedFrame, str]]: """Feed raw bytes into the parser. Returns a list of: @@ -64,7 +64,7 @@ class FrameParser: - str for ASCII log lines or warning messages """ self._remained += data - results: list[ParsedFrame | str] = [] + results: list[Union[ParsedFrame, str]] = [] # Bounded buffer check (Review Correction #2) if len(self._remained) > MAX_REMAINDER_SIZE: @@ -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 @@ -132,7 +135,7 @@ class FrameParser: offset: int, checksum_fn: Callable[[bytes], int], scope: ChecksumScope, - ) -> tuple[ParsedFrame, int] | None: + ) -> Optional[tuple[ParsedFrame, int]]: """Try to parse a frame at the given offset with specific checksum params.""" if offset + FRAME_HEADER_SIZE > len(buf): return None @@ -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 + ) -> Optional[tuple[ParsedFrame, int, ChecksumMode]]: """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 + ) -> Optional[tuple[ParsedFrame, int]]: """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[Union[ParsedFrame, str]] + ) -> None: """Collect bytes for ASCII line assembly. Only printable ASCII (0x20-0x7E) and newline (0x0A) are collected. @@ -265,7 +283,7 @@ class FrameParser: results.append(self._ascii_buffer) self._ascii_buffer = '' - def _flush_ascii(self, results: list[ParsedFrame | str]) -> None: + def _flush_ascii(self, results: list[Union[ParsedFrame, str]]) -> None: """Flush any pending ASCII buffer.""" if self._ascii_buffer: results.append(self._ascii_buffer) 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..cc9cb6cec29 100644 --- a/tools/bt/ble_log_console/src/backend/internal_decoder.py +++ b/tools/bt/ble_log_console/src/backend/internal_decoder.py @@ -1,13 +1,12 @@ # 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. """ - import struct +from typing import Optional from src.backend.models import BufUtilResult from src.backend.models import EnhStatResult @@ -28,7 +27,7 @@ _ENH_STAT_STRUCT = struct.Struct(' InternalDecoderResult | None: +def decode_internal_frame(payload: bytes) -> Optional[InternalDecoderResult]: """Decode an INTERNAL frame payload. Args: @@ -65,9 +64,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 +85,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..7297dd591f0 100644 --- a/tools/bt/ble_log_console/src/backend/models.py +++ b/tools/bt/ble_log_console/src/backend/models.py @@ -1,12 +1,13 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - import struct from dataclasses import dataclass from dataclasses import field from enum import Enum from pathlib import Path +from typing import Optional from typing import TypedDict +from typing import Union from textual.message import Message @@ -87,7 +88,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 +133,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 +147,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 +155,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 +205,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 +215,7 @@ class TransportSnapshot: fps: float = 0.0 -@dataclass(slots=True) +@dataclass class LossSnapshot: """Snapshot of firmware-reported cumulative loss.""" @@ -220,12 +223,12 @@ 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).""" - per_source: dict[SourceCode, SourcePeakWrite] | None = None - max_per_source: dict[SourceCode, SourcePeakWrite] | None = None + per_source: Optional[dict[SourceCode, SourcePeakWrite]] = None + max_per_source: Optional[dict[SourceCode, SourcePeakWrite]] = None class LossType(str, Enum): @@ -271,7 +274,7 @@ class FunnelSnapshot: throughput: ThroughputInfo -@dataclass(slots=True) +@dataclass class LaunchConfig: """Configuration returned by the Launch Screen.""" @@ -280,7 +283,7 @@ class LaunchConfig: log_dir: Path -@dataclass(slots=True) +@dataclass class FrameStats: """Periodic stats snapshot with metrics grouped by dimension.""" @@ -288,10 +291,10 @@ class FrameStats: loss: LossSnapshot = field(default_factory=LossSnapshot) os_peak: PeakBurstSnapshot = field(default_factory=PeakBurstSnapshot) ll_peak: PeakBurstSnapshot = field(default_factory=PeakBurstSnapshot) - per_source_rx_bytes: dict[SourceCode, int] | None = None + per_source_rx_bytes: Optional[dict[SourceCode, int]] = None sync_state: SyncState = SyncState.SEARCHING - checksum_algorithm: ChecksumAlgorithm | None = None - checksum_scope: ChecksumScope | None = None + checksum_algorithm: Optional[ChecksumAlgorithm] = None + checksum_scope: Optional[ChecksumScope] = None # --- TypedDicts for internal decoder results --- @@ -323,7 +326,7 @@ class BufUtilResult(TypedDict): os_ts_ms: int -InternalDecoderResult = InfoResult | EnhStatResult | BufUtilResult +InternalDecoderResult = Union[InfoResult, EnhStatResult, BufUtilResult] # --- Textual Messages (backend -> frontend) --- @@ -339,8 +342,8 @@ class StatsUpdated(Message): def __init__( self, stats: FrameStats, - funnel_snapshots: list[FunnelSnapshot] | None = None, - buf_util_snapshots: list[BufUtilEntry] | None = None, + funnel_snapshots: Optional[list[FunnelSnapshot]] = None, + buf_util_snapshots: Optional[list[BufUtilEntry]] = None, ) -> None: super().__init__() self.stats = stats @@ -368,7 +371,7 @@ class FrameLossDetected(Message): loss_type: LossType, lost_frames: int, lost_bytes: int, - sn_range: tuple[int, int] | None = None, + sn_range: Optional[tuple[int, int]] = None, ) -> None: super().__init__() self.source_name = source_name diff --git a/tools/bt/ble_log_console/src/backend/stats/__init__.py b/tools/bt/ble_log_console/src/backend/stats/__init__.py index e162df47761..ded0f8cffbb 100644 --- a/tools/bt/ble_log_console/src/backend/stats/__init__.py +++ b/tools/bt/ble_log_console/src/backend/stats/__init__.py @@ -1,13 +1,11 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Stats package -- re-exports for backward-compatible imports.""" - from src.backend.stats.accumulator import StatsAccumulator from src.backend.stats.firmware_loss import FirmwareLossTracker from src.backend.stats.firmware_written import FirmwareWrittenTracker -from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS from src.backend.stats.peak_burst import PeakBurstTracker +from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS from src.backend.stats.sn_gap import REORDER_WINDOW from src.backend.stats.sn_gap import SN_MAX from src.backend.stats.sn_gap import SNGapTracker @@ -16,8 +14,8 @@ from src.backend.stats.traffic_spike import TRAFFIC_THRESHOLD_PCT from src.backend.stats.traffic_spike import TRAFFIC_WINDOW_SEC from src.backend.stats.traffic_spike import TrafficSpikeDetector from src.backend.stats.traffic_spike import TrafficSpikeResult -from src.backend.stats.transport import UART_BITS_PER_BYTE from src.backend.stats.transport import TransportMetrics +from src.backend.stats.transport import UART_BITS_PER_BYTE __all__ = [ 'FirmwareLossTracker', diff --git a/tools/bt/ble_log_console/src/backend/stats/buf_util.py b/tools/bt/ble_log_console/src/backend/stats/buf_util.py index 7b2d952b140..9449b077736 100644 --- a/tools/bt/ble_log_console/src/backend/stats/buf_util.py +++ b/tools/bt/ble_log_console/src/backend/stats/buf_util.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from __future__ import annotations from src.backend.models import BufUtilEntry diff --git a/tools/bt/ble_log_console/src/backend/stats/firmware_loss.py b/tools/bt/ble_log_console/src/backend/stats/firmware_loss.py index c6f4ca898e8..2b8b36fe3c6 100644 --- a/tools/bt/ble_log_console/src/backend/stats/firmware_loss.py +++ b/tools/bt/ble_log_console/src/backend/stats/firmware_loss.py @@ -1,11 +1,9 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Firmware ENH_STAT loss tracking with incremental delta accumulation. Resilient to firmware counter resets from ``ble_log_bench_reset_stat``. """ - from src.backend.models import LossSnapshot from src.backend.models import SourceCode diff --git a/tools/bt/ble_log_console/src/backend/stats/firmware_written.py b/tools/bt/ble_log_console/src/backend/stats/firmware_written.py index 08a23b45dfa..88fe73a421a 100644 --- a/tools/bt/ble_log_console/src/backend/stats/firmware_written.py +++ b/tools/bt/ble_log_console/src/backend/stats/firmware_written.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.models import SourceCode diff --git a/tools/bt/ble_log_console/src/backend/stats/peak_burst.py b/tools/bt/ble_log_console/src/backend/stats/peak_burst.py index 142253e5bf7..e019de14021 100644 --- a/tools/bt/ble_log_console/src/backend/stats/peak_burst.py +++ b/tools/bt/ble_log_console/src/backend/stats/peak_burst.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Sliding-window peak write burst tracker. Tracks the densest burst of log writes within a configurable time window @@ -12,7 +11,6 @@ happen at microsecond frequency, many frames share the same ms timestamp. Instead of computing an inaccurate fps, we count frames and bytes within the densest window. """ - from collections import deque from src.backend.models import PeakBurstSnapshot 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..95bc142976f 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 typing import Optional from src.backend.models import SourceCode @@ -63,7 +63,7 @@ class SNGapTracker: """Return cumulative confirmed gap count per source.""" return dict(self._gap_accum) - def reset(self, src_code: SourceCode | None = None) -> None: + def reset(self, src_code: Optional[SourceCode] = None) -> None: """Reset tracker state. If src_code is None, resets all sources. 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..1b3728cf672 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,11 +1,10 @@ # 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.""" - import time from collections import deque from dataclasses import dataclass +from typing import Optional from src.backend.models import SourceCode @@ -14,7 +13,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,9 +40,11 @@ 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: + def check(self) -> Optional[TrafficSpikeResult]: now = time.perf_counter() window = self._window cutoff = now - TRAFFIC_WINDOW_SEC @@ -83,7 +84,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/stats/transport.py b/tools/bt/ble_log_console/src/backend/stats/transport.py index 45507db7827..47e0e437d38 100644 --- a/tools/bt/ble_log_console/src/backend/stats/transport.py +++ b/tools/bt/ble_log_console/src/backend/stats/transport.py @@ -1,8 +1,6 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Transport-layer metrics: RX bytes, throughput (bps), frame rate (fps).""" - from src.backend.models import TransportSnapshot UART_BITS_PER_BYTE = 10 # 8 data + 1 start + 1 stop 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..e27c7000da5 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 typing import Optional -import serial import serial.tools.list_ports UART_READ_TIMEOUT = 0.1 @@ -18,7 +17,7 @@ def list_serial_ports() -> list[str]: return [port.device for port in ports] -def validate_uart_port(port: str) -> str | None: +def validate_uart_port(port: str) -> Optional[str]: """Validate port exists and is accessible. Returns error message or None if valid.""" available = list_serial_ports() if port not in available: @@ -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..8129c7adc1c 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,15 @@ # 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 pathlib import Path +from typing import Optional +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 +23,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[Optional[LaunchConfig]]): """Interactive setup screen for BLE Log Console.""" DEFAULT_CSS = """ @@ -93,7 +91,7 @@ class LaunchScreen(Screen[LaunchConfig | None]): Binding('ctrl+c', 'quit', show=False, priority=True), ] - def __init__(self, default_log_dir: Path | None = None) -> None: + def __init__(self, default_log_dir: Optional[Path] = None) -> None: super().__init__() self._default_log_dir = default_log_dir or Path.cwd() diff --git a/tools/bt/ble_log_console/src/frontend/log_view.py b/tools/bt/ble_log_console/src/frontend/log_view.py index 9bd7309f1cc..c0322be221b 100644 --- a/tools/bt/ble_log_console/src/frontend/log_view.py +++ b/tools/bt/ble_log_console/src/frontend/log_view.py @@ -1,11 +1,9 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Scrollable log view widget. See Spec Section 11. """ - from rich.text import Text from textual.widgets import RichLog diff --git a/tools/bt/ble_log_console/src/frontend/shortcut_screen.py b/tools/bt/ble_log_console/src/frontend/shortcut_screen.py index db7dda2c5c9..615ebd95b82 100644 --- a/tools/bt/ble_log_console/src/frontend/shortcut_screen.py +++ b/tools/bt/ble_log_console/src/frontend/shortcut_screen.py @@ -1,11 +1,9 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Modal screen showing available keyboard shortcuts. Pushed by the 'h' keybinding; dismissed by Escape or 'h' again. """ - from rich.table import Table from textual.app import ComposeResult from textual.binding import Binding diff --git a/tools/bt/ble_log_console/src/frontend/stats_screen.py b/tools/bt/ble_log_console/src/frontend/stats_screen.py index 8bfbdd1947e..2d42e4d259d 100644 --- a/tools/bt/ble_log_console/src/frontend/stats_screen.py +++ b/tools/bt/ble_log_console/src/frontend/stats_screen.py @@ -1,32 +1,29 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - """Modal screen for per-source frame statistics display. Pushed by the 'd' keybinding; dismissed by Escape or 'd' again. Refreshes every second to show live throughput data. """ - from __future__ import annotations from typing import TYPE_CHECKING from rich.table import Table from rich.text import Text +from src.backend.models import BufUtilEntry +from src.backend.models import format_bytes +from src.backend.models import format_throughput +from src.backend.models import FunnelSnapshot +from src.backend.models import resolve_lbm_name +from src.backend.models import resolve_pool_name +from src.backend.models import resolve_source_name from textual.app import ComposeResult from textual.binding import Binding from textual.containers import Vertical from textual.screen import ModalScreen from textual.widgets import Static -from src.backend.models import BufUtilEntry -from src.backend.models import FunnelSnapshot -from src.backend.models import format_bytes -from src.backend.models import format_throughput -from src.backend.models import resolve_lbm_name -from src.backend.models import resolve_pool_name -from src.backend.models import resolve_source_name - if TYPE_CHECKING: from src.app import BLELogApp diff --git a/tools/bt/ble_log_console/tests/helpers.py b/tools/bt/ble_log_console/tests/helpers.py index 8c8ea283c6a..6059720dc8b 100644 --- a/tools/bt/ble_log_console/tests/helpers.py +++ b/tools/bt/ble_log_console/tests/helpers.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - import struct from collections.abc import Callable diff --git a/tools/bt/ble_log_console/tests/test_buf_util.py b/tools/bt/ble_log_console/tests/test_buf_util.py index c6e211a4f3b..d6f9ba5660a 100644 --- a/tools/bt/ble_log_console/tests/test_buf_util.py +++ b/tools/bt/ble_log_console/tests/test_buf_util.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.stats.buf_util import BufUtilTracker diff --git a/tools/bt/ble_log_console/tests/test_checksum.py b/tools/bt/ble_log_console/tests/test_checksum.py index b67f229da40..c0a01e17d99 100644 --- a/tools/bt/ble_log_console/tests/test_checksum.py +++ b/tools/bt/ble_log_console/tests/test_checksum.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.checksum import sum_checksum from src.backend.checksum import xor_checksum diff --git a/tools/bt/ble_log_console/tests/test_firmware_loss.py b/tools/bt/ble_log_console/tests/test_firmware_loss.py index e346d22dfe9..7117e3cda1e 100644 --- a/tools/bt/ble_log_console/tests/test_firmware_loss.py +++ b/tools/bt/ble_log_console/tests/test_firmware_loss.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.stats.firmware_loss import FirmwareLossTracker diff --git a/tools/bt/ble_log_console/tests/test_firmware_written.py b/tools/bt/ble_log_console/tests/test_firmware_written.py index 2112b0a4815..fe6431e13be 100644 --- a/tools/bt/ble_log_console/tests/test_firmware_written.py +++ b/tools/bt/ble_log_console/tests/test_firmware_written.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.stats.firmware_written import FirmwareWrittenTracker diff --git a/tools/bt/ble_log_console/tests/test_internal_decoder.py b/tools/bt/ble_log_console/tests/test_internal_decoder.py index e48892ef177..07b31ab1ac4 100644 --- a/tools/bt/ble_log_console/tests/test_internal_decoder.py +++ b/tools/bt/ble_log_console/tests/test_internal_decoder.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - import struct from src.backend.internal_decoder import decode_internal_frame diff --git a/tools/bt/ble_log_console/tests/test_launch_screen.py b/tools/bt/ble_log_console/tests/test_launch_screen.py index cffd4442aa9..b3e91cc7234 100644 --- a/tools/bt/ble_log_console/tests/test_launch_screen.py +++ b/tools/bt/ble_log_console/tests/test_launch_screen.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from pathlib import Path from unittest.mock import MagicMock from unittest.mock import patch diff --git a/tools/bt/ble_log_console/tests/test_peak_burst.py b/tools/bt/ble_log_console/tests/test_peak_burst.py index bc650b69299..fa1907dd2df 100644 --- a/tools/bt/ble_log_console/tests/test_peak_burst.py +++ b/tools/bt/ble_log_console/tests/test_peak_burst.py @@ -1,9 +1,8 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - -from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS -from src.backend.stats.peak_burst import PeakBurstTracker from src.backend.stats.peak_burst import _ts_delta_ms +from src.backend.stats.peak_burst import PeakBurstTracker +from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS _SRC = 1 _SRC_B = 2 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..f23503db0e8 100644 --- a/tools/bt/ble_log_console/tests/test_traffic_spike.py +++ b/tools/bt/ble_log_console/tests/test_traffic_spike.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - +from typing import Optional from unittest.mock import patch from src.backend.stats.traffic_spike import TRAFFIC_ALERT_COOLDOWN_SEC @@ -17,7 +17,7 @@ def _make_detector(baudrate: int = 3_000_000) -> TrafficSpikeDetector: def _trigger_spike( d: TrafficSpikeDetector, mock_time: object, t: float, hot_bytes: int, src: int = 1 -) -> TrafficSpikeResult | None: +) -> Optional[TrafficSpikeResult]: mock_time.perf_counter.return_value = t # type: ignore[attr-defined] d.record(hot_bytes, src) mock_time.perf_counter.return_value = t + 0.05 # type: ignore[attr-defined] diff --git a/tools/bt/ble_log_console/tests/test_transport.py b/tools/bt/ble_log_console/tests/test_transport.py index 22a354abafe..bae8ec4887a 100644 --- a/tools/bt/ble_log_console/tests/test_transport.py +++ b/tools/bt/ble_log_console/tests/test_transport.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from src.backend.stats.transport import TransportMetrics diff --git a/tools/bt/ble_log_console/tests/test_uart_transport.py b/tools/bt/ble_log_console/tests/test_uart_transport.py index c395e97150e..80b148d1682 100644 --- a/tools/bt/ble_log_console/tests/test_uart_transport.py +++ b/tools/bt/ble_log_console/tests/test_uart_transport.py @@ -1,6 +1,5 @@ # SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 - from unittest.mock import patch from src.backend.uart_transport import validate_uart_port