import asyncio
import os
import sys
import threading
import time
from collections import deque
from dataclasses import dataclass
from queue import Queue
from typing import Deque, Dict, List, Optional
from rich import box
from rich.console import Console
from rich.table import Table
from rich.text import Text
from serial.tools import list_ports
from ..devices.esp_serial import EspPortInfo, detect_port_info_no_cache
MAX_RECENT_DEVICES = 5
MAX_DETECT_RETRY = 2
MAX_DEBUG_LOGS = 20
# Debug mode: set UART_MONITOR_DEBUG=1 to keep screen history and show debug logs.
DEBUG = os.environ.get('UART_MONITOR_DEBUG', '').lower() in ('1', 'true', 'yes', 'on')
[docs]
@dataclass
class Chip:
target: str = ''
mac: str = ''
revision: str = ''
xtal: str = ''
flash: str = ''
description: str = ''
[docs]
def clear(self) -> None:
self.target = ''
self.mac = ''
self.revision = ''
self.xtal = ''
self.flash = ''
self.description = ''
[docs]
@dataclass
class Device:
name: str
sys_device: str
location: str
description: str
connected: bool
last_seen: float
first_seen: float
chip: Chip
console = Console()
devices: Dict[str, Device] = {}
devices_lock = threading.Lock()
detect_queue: 'Queue[Device]' = Queue() # queue.Queue is not subscriptable on Python 3.7
recent_devices: List[Device] = [] # recent connecting devices
debug_logs: Deque[str] = deque(maxlen=MAX_DEBUG_LOGS)
debug_logs_lock = threading.Lock()
[docs]
def debug_print(*args) -> None: # type: ignore
"""Record debug message for on-screen debug log panel."""
if not DEBUG:
return
with debug_logs_lock:
debug_logs.append('[DEBUG] ' + ' '.join(str(arg) for arg in args))
def _update_chip_from_port_info(chip: Chip, esp_port: EspPortInfo) -> bool:
if not esp_port.support_esptool:
chip.clear()
chip.target = 'unknown'
chip.description = esp_port.serial_description
return True
chip.target = esp_port.target
chip.revision = esp_port.chip_version
chip.xtal = f'{esp_port.chip_xtal}MHz' if esp_port.chip_xtal else ''
chip.mac = esp_port.mac
chip.flash = esp_port.flash_size or ''
chip.description = esp_port.chip_description
return True
async def _detect_port_info(device: Device) -> EspPortInfo:
"""Run the (blocking) port detection off the event loop.
Uses ``asyncio.to_thread`` (Python 3.9+) when available; on Python 3.7/3.8
it falls back to calling the detection synchronously.
"""
if hasattr(asyncio, 'to_thread'):
return await asyncio.to_thread(
detect_port_info_no_cache,
device.sys_device,
device.location,
device.description,
)
# Python 3.7/3.8 fallback: asyncio.to_thread is unavailable, call directly.
return detect_port_info_no_cache(device.sys_device, device.location, device.description)
[docs]
async def detect_port_chip(device: Device) -> None:
"""Detect chip info on a serial port with retry."""
retry = MAX_DETECT_RETRY
while True:
esp_port_info = None
try:
esp_port_info = await _detect_port_info(device)
except Exception as e: # pylint: disable=broad-exception-caught
debug_print(f'{device.sys_device} detect info failed {type(e)}: {str(e)}')
debug_print(esp_port_info)
if esp_port_info:
_update_chip_from_port_info(device.chip, esp_port_info)
if esp_port_info.support_esptool:
return
# retry if esptool failed
if retry > 0:
retry -= 1
await asyncio.sleep(1 * (MAX_DETECT_RETRY - retry))
continue
return
[docs]
async def detect_chip(device: Device) -> None:
"""Detect the espressif chip on specified serial port."""
# lsof is Linux-only; on Windows / when missing, skip occupancy check.
try:
proc = await asyncio.create_subprocess_exec(
'lsof', device.sys_device, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
await proc.communicate()
if proc.returncode == 0:
debug_print(f'{device.sys_device} is occupied (lsof)')
device.chip.target = ''
return # lsof got succ, the device is occupied
except FileNotFoundError:
debug_print('lsof command not available, skip occupancy check')
await detect_port_chip(device)
[docs]
async def detect_all_chips() -> None:
"""Detect all devices concurrently."""
tasks = [detect_chip(detect_queue.get()) for _ in range(detect_queue.qsize())]
await asyncio.gather(*tasks, return_exceptions=False)
[docs]
def detect_chip_worker() -> None:
while True:
device = detect_queue.get()
asyncio.run(detect_chip(device))
display_serial_ports()
def _port_identity(port) -> Optional[str]: # type: ignore
"""Stable key for a serial port across platforms.
Linux prefers ``usb_interface_path``; Windows usually lacks it, so fall back
to ``hwid`` / ``device``.
"""
return getattr(port, 'usb_interface_path', None) or getattr(port, 'hwid', None) or port.device or None
[docs]
def refresh_serial_ports(initial: bool = True) -> bool:
global recent_devices # pylint: disable=global-statement
timestamp = time.time()
changed = False
with devices_lock:
ports = list(list_ports.comports())
for port in ports:
iface_path = _port_identity(port)
# Linux: require USB location to filter non-USB tty devices.
# Windows: location may be missing; track by device/hwid instead.
if not iface_path:
continue
if sys.platform != 'win32' and not port.location:
continue
name = port.name if port.name else port.device.split('/')[-1].split('\\')[-1]
location = port.location or port.device
if iface_path in devices:
# update the device info
device = devices[iface_path]
if device.last_seen - device.first_seen < 10 and timestamp - device.first_seen >= 10: # pylint: disable=chained-comparison
changed = True
if not device.connected:
changed = True
device.first_seen = timestamp
device.chip.clear()
device.chip.target = 'Detecting...'
# the device name and location may change, so we update it
device.sys_device = port.device
device.location = location
device.description = port.description
detect_queue.put(device)
if not initial:
recent_devices.append(device)
device.name = name
device.connected = True
device.last_seen = timestamp
else:
# new device
device = Device(
name,
port.device,
location,
port.description,
connected=True,
last_seen=timestamp,
first_seen=timestamp,
chip=Chip(target='Detecting...'),
)
devices[iface_path] = device
detect_queue.put(device)
if not initial:
recent_devices.append(device)
changed = True
# update the status of disconnected devices
disconnect_devices = set()
for iface_path, device in list(devices.items()):
if device.last_seen < timestamp:
if device.connected:
device.connected = False
disconnect_devices.add(device.location)
changed = True
if device.last_seen < timestamp - 10:
del devices[iface_path]
changed = True
recent_devices = [
d
for d in recent_devices[-MAX_RECENT_DEVICES:]
if d.location not in disconnect_devices and (timestamp - d.last_seen) <= 1800
]
return changed
[docs]
def device_event_handler(action, device): # type: ignore # pylint: disable=unused-argument
if device.subsystem != 'tty':
return
if refresh_serial_ports(False):
display_serial_ports()
[docs]
def check_new_devices_status() -> None:
if refresh_serial_ports(False):
display_serial_ports()
def _add_debug_logs_row() -> None:
if not DEBUG:
return
with debug_logs_lock:
logs = list(debug_logs)
log_table = Table(title='Debug Logs', header_style='', box=box.ROUNDED, show_header=False)
log_table.add_column('Message', overflow='fold')
if logs:
for line in logs:
log_table.add_row(Text(line, style='dim'))
else:
log_table.add_row(Text('(no logs yet)', style='dim italic'))
console.print(log_table)
[docs]
def display_serial_ports() -> None:
table = Table(title='Serial Ports', header_style='', box=box.ROUNDED)
table.add_column('Location', justify='left', min_width=15)
table.add_column('Name', justify='left', min_width=10)
table.add_column('Status', justify='center')
table.add_column('Target', justify='left', min_width=15)
table.add_column('Revision', justify='left')
table.add_column('XTAL', justify='left')
table.add_column('MAC', justify='left', min_width=25)
table.add_column('Flash', justify='left')
table.add_column('Description', justify='left')
with devices_lock:
sorted_devices = sorted(devices.values(), key=lambda d: d.location)
current_time = time.time()
for i, device in enumerate(sorted_devices):
is_new_device = (current_time - device.first_seen) <= 10
if device.connected:
status = '●'
style = 'green' if is_new_device else ''
status_text = Text(status, style='green')
else:
status = '○'
style = 'dim'
status_text = Text(status, style=style)
location_text = Text(device.location, style=style)
name_text = Text(device.name, style=style)
target = Text(device.chip.target, style=style)
rev = Text(device.chip.revision, style=style)
xtal = Text(device.chip.xtal, style=style)
mac = Text(device.chip.mac, style=style)
flash = Text(device.chip.flash, style=style)
description = Text(device.chip.description, style=style)
table.add_row(
location_text,
name_text,
status_text,
target,
rev,
xtal,
mac,
flash,
description,
end_section=(i == len(sorted_devices) - 1),
)
if recent_devices:
for device in recent_devices[::-1]:
if (current_time - device.first_seen) <= 1800:
location_text = Text(device.location)
name_text = Text(device.name)
status_text = Text('●', style='green')
target = Text(device.chip.target)
rev = Text(device.chip.revision)
xtal = Text(device.chip.xtal)
mac = Text(device.chip.mac)
flash = Text(device.chip.flash)
description = Text(device.chip.description)
table.add_row(location_text, name_text, status_text, target, rev, xtal, mac, flash, description)
console.clear()
console.print(table)
_add_debug_logs_row()
console.print('Press Ctrl+C to exit')
def _bootstrap_monitoring() -> None:
"""Initial scan, detect existing ports, and start the detect worker."""
refresh_serial_ports()
display_serial_ports()
# detecting all ports first
asyncio.run(detect_all_chips())
display_serial_ports()
# start a detect worker for new coming device
detect_thread = threading.Thread(target=detect_chip_worker, daemon=True)
detect_thread.start()
[docs]
def start_monitoring_linux() -> None:
"""Linux monitor: pyudev netlink for hotplug + 1s polling fallback."""
import pyudev # Linux-only; must not be imported at module top level
_bootstrap_monitoring()
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by(subsystem='tty')
observer = pyudev.MonitorObserver(monitor, device_event_handler)
observer.daemon = True
observer.start()
try:
while True:
check_new_devices_status()
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
[docs]
def start_monitoring_win() -> None:
"""Windows monitor: poll serial ports periodically (no pyudev)."""
_bootstrap_monitoring()
try:
while True:
check_new_devices_status()
time.sleep(1)
except KeyboardInterrupt:
pass
[docs]
def start_monitoring() -> None:
"""UART monitor entry: pyudev on Linux, polling on Windows."""
if sys.platform == 'win32':
start_monitoring_win()
else:
start_monitoring_linux()
[docs]
def main() -> None:
start_monitoring()
if __name__ == '__main__':
main()