Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-10-02 03:00:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(ble_log_console): resolve pre-commit flake8, mypy, and import ordering failures

Browse Source 
Replace Python 3.10+ X | Y union syntax with Optional/Union from typing,
remove unsupported @dataclass(slots=True), fix flake8 E203 slice whitespace,
and apply auto-fixed import ordering and quote style.
This commit is contained in:
Zhou Xiao
2026-04-07 16:25:09 +08:00
parent 37dc0826e6
commit 0e54ef88b3
31 changed files with 178 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/bt/ble_log_console/conftest.py
-1
View File
@@ -1,6 +1,5 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import sys
from pathlib import Path
tools/bt/ble_log_console/src/__init__.py
-1
View File
@@ -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
tools/bt/ble_log_console/src/app.py
+70 -37
View File
@@ -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('<I', item.payload, LL_TS_OFFSET)
stats.record_ll_frame_ts(lc_ts_us, frame_size, item.source_code)
(lc_ts_us,) = struct.unpack_from(
'<I', item.payload, LL_TS_OFFSET
)
stats.record_ll_frame_ts(
lc_ts_us, frame_size, item.source_code
)
elif item.source_code == BleLogSource.REDIR:
wall_ms = int(time.perf_counter() * 1000) & 0xFFFFFFFF
stats.record_frame_wall_ts(wall_ms, frame_size, item.source_code)
stats.record_frame_wall_ts(
wall_ms, frame_size, item.source_code
)
# Decode internal frames
if item.source_code == BleLogSource.INTERNAL:
@@ -251,7 +271,10 @@ class BLELogApp(App):
self._post(InternalFrameDecoded(int_src, decoded))
if int_src in (InternalSource.INIT_DONE, InternalSource.INFO):
if int_src in (
InternalSource.INIT_DONE,
InternalSource.INFO,
):
info = cast(InfoResult, decoded)
stats.set_firmware_version(info['version'])
if int_src == InternalSource.INIT_DONE:
@@ -269,7 +292,9 @@ class BLELogApp(App):
baudrate=self._baudrate,
)
if new_frames > 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 ---
tools/bt/ble_log_console/src/backend/checksum.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/backend/frame_parser.py
+38 -20
View File
@@ -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)
tools/bt/ble_log_console/src/backend/internal_decoder.py
+13 -7
View File
@@ -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('<BBIIII')
_BUF_UTIL_STRUCT = struct.Struct('<BBBB')
def decode_internal_frame(payload: bytes) -> 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(
tools/bt/ble_log_console/src/backend/models.py
+23 -20
View File
@@ -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
tools/bt/ble_log_console/src/backend/stats/__init__.py
+2 -4
View File
@@ -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',
tools/bt/ble_log_console/src/backend/stats/buf_util.py
-1
View File
@@ -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
tools/bt/ble_log_console/src/backend/stats/firmware_loss.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/backend/stats/firmware_written.py
-1
View File
@@ -1,6 +1,5 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src.backend.models import SourceCode
tools/bt/ble_log_console/src/backend/stats/peak_burst.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/backend/stats/sn_gap.py
+2 -2
View File
@@ -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.
tools/bt/ble_log_console/src/backend/stats/traffic_spike.py
+9 -6
View File
@@ -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,
tools/bt/ble_log_console/src/backend/stats/transport.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/backend/uart_transport.py
+5 -4
View File
@@ -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)
tools/bt/ble_log_console/src/frontend/launch_screen.py
+5 -7
View File
@@ -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()
tools/bt/ble_log_console/src/frontend/log_view.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/frontend/shortcut_screen.py
-2
View File
@@ -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
tools/bt/ble_log_console/src/frontend/stats_screen.py
+7 -10
View File
@@ -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
tools/bt/ble_log_console/tests/helpers.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_buf_util.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_checksum.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_firmware_loss.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_firmware_written.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_internal_decoder.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_launch_screen.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_peak_burst.py
+2 -3
View File
@@ -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
tools/bt/ble_log_console/tests/test_traffic_spike.py
+2 -2
View File
@@ -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]
tools/bt/ble_log_console/tests/test_transport.py
-1
View File
@@ -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
tools/bt/ble_log_console/tests/test_uart_transport.py
-1
View File
@@ -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