mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-01 10:32:23 +00:00
agessaman
3fc3bbf8fb
`modules/radio_link.py` now holds connecting and reconnecting the transport, the host->radio command serializer and `radio_session`, zombie state and the periodic health probe: 20 methods moved verbatim, so every name is still on `MeshCoreBot`. `_serialize_command_frames` and `_radio_session_held` moved with them and are re-exported from `modules.core`; the serializer's `bot` annotation is now the mixin. `core.py` keeps `import meshcore` because tests patch `modules.core.meshcore.MeshCore`. core.py: 2,164 → 1,599 lines.
613 lines
25 KiB
Python
613 lines
25 KiB
Python
"""Radio link and health, mixed into MeshCoreBot.
|
||
|
||
Connecting and reconnecting the transport, the host->radio command serializer and
|
||
``radio_session``, zombie detection and the periodic health probe."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import contextvars
|
||
import functools
|
||
import time
|
||
from typing import Any
|
||
|
||
import meshcore
|
||
from meshcore import EventType
|
||
|
||
# True while the current task holds the radio through MeshCoreBot.radio_session(),
|
||
# so the frames it sends don't try to take the (non-reentrant) lock again.
|
||
_radio_session_held: contextvars.ContextVar[bool] = contextvars.ContextVar(
|
||
"radio_session_held", default=False
|
||
)
|
||
|
||
|
||
def _serialize_command_frames(bot: RadioLinkMixin, commands: Any) -> bool:
|
||
"""Route every host->radio frame through the bot's radio command lock.
|
||
|
||
The companion firmware processes one host serial frame per main-loop
|
||
iteration and has no mid-frame resync: a burst of concurrent commands can
|
||
overrun the radio's USB-CDC RX buffer, drop a byte, and permanently desync
|
||
its frame parser (commands stop being acted on while RX push frames keep
|
||
flowing). The bot issues commands from many independent asyncio tasks
|
||
(sends, channel/contact ops, scheduler ops, health probes, auto message
|
||
fetch) with no shared serialization.
|
||
|
||
Every meshcore command writes its frame through ``CommandHandler.send()``,
|
||
which waits for the radio's immediate reply (OK, ERROR, MSG_SENT, ...).
|
||
Wrapping ``send`` on the handler instance serializes exactly that exchange
|
||
and paces frames by a minimum interval, so there is at most one in-flight
|
||
companion frame at a time. Library methods call ``self.send``, so composite
|
||
commands (``send_msg_with_retry``, ``req_*_sync``, ``send_login_sync``) and
|
||
``meshcore_cli.next_cmd`` are covered too, while their waits for ACKs and
|
||
remote responses happen outside the lock and don't block other senders.
|
||
|
||
Returns False when ``commands`` is already serialized.
|
||
"""
|
||
send = commands.send
|
||
if getattr(send, "_radio_serialized", False):
|
||
return False
|
||
|
||
@functools.wraps(send)
|
||
async def _serialized_send(*args: Any, **kwargs: Any) -> Any:
|
||
if _radio_session_held.get():
|
||
await bot._pace_radio_command()
|
||
return await send(*args, **kwargs)
|
||
async with bot._get_radio_cmd_lock():
|
||
await bot._pace_radio_command()
|
||
return await send(*args, **kwargs)
|
||
|
||
_serialized_send._radio_serialized = True # type: ignore[attr-defined]
|
||
commands.send = _serialized_send
|
||
return True
|
||
|
||
|
||
|
||
|
||
class RadioLinkMixin:
|
||
"""Mixed into MeshCoreBot."""
|
||
|
||
_allow_offline_trial: Any
|
||
_configure_meshcore_debug_logging: Any
|
||
_radio_cmd_last_ts: Any
|
||
_radio_cmd_lock: Any
|
||
_radio_cmd_min_interval: Any
|
||
_radio_fail_count: Any
|
||
_radio_relinks_in_progress: Any
|
||
_radio_zombie_detected: Any
|
||
_shutdown_event: Any
|
||
_tcp_probe_fail_count: Any
|
||
_transport_reconnect_in_progress: Any
|
||
_transport_reconnect_lock: Any
|
||
channel_manager: Any
|
||
config: Any
|
||
connected: Any
|
||
connection_time: Any
|
||
db_manager: Any
|
||
logger: Any
|
||
meshcore: Any
|
||
set_device_name: Any
|
||
set_radio_clock: Any
|
||
setup_message_handlers: Any
|
||
transmission_tracker: Any
|
||
wait_for_contacts: Any
|
||
|
||
@property
|
||
def is_radio_zombie(self) -> bool:
|
||
"""True when the radio firmware has been confirmed unresponsive.
|
||
|
||
All outbound radio sends should check this flag and abort immediately.
|
||
Only a physical power cycle can recover the radio; the flag is cleared
|
||
automatically when connect() succeeds after a power cycle.
|
||
"""
|
||
return bool(self._radio_zombie_detected)
|
||
|
||
async def _attempt_reconnect(self) -> bool:
|
||
"""Attempt to reconnect to the MeshCore node with exponential backoff.
|
||
|
||
Reads reconnect settings from [Connection]:
|
||
reconnect_max_retries – max attempts before giving up (0 = unlimited, default 0)
|
||
reconnect_delay_seconds – initial wait between attempts (default 10)
|
||
reconnect_max_delay_seconds – cap on wait time (default 60)
|
||
|
||
Returns:
|
||
bool: True if reconnection succeeded, False if max retries exhausted or shutdown.
|
||
"""
|
||
max_retries = self.config.getint('Connection', 'reconnect_max_retries', fallback=0)
|
||
delay = self.config.getfloat('Connection', 'reconnect_delay_seconds', fallback=10.0)
|
||
max_delay = self.config.getfloat('Connection', 'reconnect_max_delay_seconds', fallback=60.0)
|
||
|
||
attempt = 0
|
||
while not self._shutdown_event.is_set():
|
||
if max_retries > 0 and attempt >= max_retries:
|
||
self.logger.error(f"Reconnect failed after {max_retries} attempt(s), giving up")
|
||
return False
|
||
|
||
attempt += 1
|
||
retry_label = f"{attempt}/{max_retries}" if max_retries > 0 else str(attempt)
|
||
self.logger.info(f"Reconnect attempt {retry_label}...")
|
||
|
||
# Clean up the stale connection object
|
||
old_meshcore = self.meshcore
|
||
self.meshcore = None
|
||
self.connected = False
|
||
if old_meshcore is not None:
|
||
try:
|
||
await asyncio.wait_for(old_meshcore.disconnect(), timeout=5.0)
|
||
except Exception:
|
||
pass
|
||
|
||
if await self.connect():
|
||
self.logger.info("Reconnected successfully")
|
||
if hasattr(self, 'transmission_tracker') and self.transmission_tracker:
|
||
self.transmission_tracker._update_bot_prefix()
|
||
return True
|
||
|
||
self.logger.warning(
|
||
f"Reconnect attempt {retry_label} failed, retrying in {delay:.0f}s..."
|
||
)
|
||
# Interruptible sleep so shutdown isn't delayed
|
||
elapsed = 0.0
|
||
while elapsed < delay and not self._shutdown_event.is_set():
|
||
await asyncio.sleep(1.0)
|
||
elapsed += 1.0
|
||
|
||
delay = min(delay * 2, max_delay)
|
||
|
||
return False
|
||
|
||
def _connection_type(self) -> str:
|
||
"""Configured transport: serial, ble, or tcp."""
|
||
return self.config.get('Connection', 'connection_type', fallback='ble').lower()
|
||
|
||
def _radio_probe_fail_threshold(self) -> int:
|
||
return self.config.getint(
|
||
'Connection',
|
||
'radio_probe_fail_threshold',
|
||
fallback=self.config.getint('Bot', 'radio_probe_fail_threshold', fallback=3),
|
||
)
|
||
|
||
def _radio_probe_interval_seconds(self) -> int:
|
||
return max(
|
||
300,
|
||
min(
|
||
900,
|
||
self.config.getint(
|
||
'Connection',
|
||
'radio_probe_interval_seconds',
|
||
fallback=self.config.getint(
|
||
'Bot', 'radio_probe_interval_seconds', fallback=300
|
||
),
|
||
),
|
||
),
|
||
)
|
||
|
||
async def _schedule_transport_reconnect(self, reason: str) -> None:
|
||
"""Queue a transport-level reconnect (non-blocking)."""
|
||
if self._shutdown_event.is_set():
|
||
return
|
||
if reason == 'manual_disconnect':
|
||
return
|
||
if self._transport_reconnect_in_progress:
|
||
return
|
||
if not getattr(self, 'connected', False):
|
||
return
|
||
|
||
self._transport_reconnect_in_progress = True
|
||
self.logger.warning(
|
||
"Transport disconnect detected (%s), scheduling reconnect...",
|
||
reason,
|
||
)
|
||
self._update_radio_connected_metadata(False)
|
||
asyncio.create_task(self._run_transport_reconnect())
|
||
|
||
async def _run_transport_reconnect(self) -> None:
|
||
"""Run reconnect with lock; clear in-progress flag when done."""
|
||
try:
|
||
if self._transport_reconnect_lock is None:
|
||
self._transport_reconnect_lock = asyncio.Lock()
|
||
async with self._transport_reconnect_lock:
|
||
if self._shutdown_event.is_set():
|
||
return
|
||
if not await self._attempt_reconnect():
|
||
self.logger.error("Could not reconnect, shutting down")
|
||
self.connected = False
|
||
finally:
|
||
self._transport_reconnect_in_progress = False
|
||
|
||
def _get_radio_cmd_lock(self) -> asyncio.Lock:
|
||
"""Return the lock that serializes host->radio commands.
|
||
|
||
Created lazily so it binds to the running event loop.
|
||
"""
|
||
if self._radio_cmd_lock is None:
|
||
self._radio_cmd_lock = asyncio.Lock()
|
||
return self._radio_cmd_lock
|
||
|
||
async def _pace_radio_command(self) -> None:
|
||
"""Enforce a minimum gap between consecutive companion frames.
|
||
|
||
Must be called while holding the radio command lock so the timestamp
|
||
bookkeeping stays serialized.
|
||
"""
|
||
interval = self._radio_cmd_min_interval
|
||
if interval <= 0:
|
||
return
|
||
now = time.monotonic()
|
||
wait = interval - (now - self._radio_cmd_last_ts)
|
||
if wait > 0:
|
||
await asyncio.sleep(wait)
|
||
self._radio_cmd_last_ts = time.monotonic()
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def radio_session(self):
|
||
"""Hold the radio for a short sequence of frames that must not interleave.
|
||
|
||
Frames are serialized one at a time, so another task's frame can land
|
||
between two of ours. Use this when that matters, e.g. setting the flood
|
||
scope, sending, and restoring it, so no other send goes out under the
|
||
temporary scope. Keep it short: every other sender waits, so don't wait
|
||
for ACKs or remote responses inside it. Re-entering from the same task
|
||
is a no-op. Tasks created inside the session inherit it, so don't spawn
|
||
work that sends after the session ends.
|
||
"""
|
||
if _radio_session_held.get():
|
||
yield
|
||
return
|
||
async with self._get_radio_cmd_lock():
|
||
token = _radio_session_held.set(True)
|
||
try:
|
||
yield
|
||
finally:
|
||
_radio_session_held.reset(token)
|
||
|
||
def _install_command_serializer(self) -> None:
|
||
"""Serialize and pace every frame ``meshcore.commands`` writes.
|
||
|
||
Idempotent and safe to call after each (re)connect. The handler is
|
||
wrapped in place, so existing call sites (``self.meshcore.commands.*``
|
||
and ``meshcore_cli.next_cmd``) need no per-call changes.
|
||
"""
|
||
if not self.meshcore:
|
||
return
|
||
cmds = getattr(self.meshcore, "commands", None)
|
||
if cmds is None:
|
||
return
|
||
try:
|
||
if _serialize_command_frames(self, cmds):
|
||
self.logger.debug(
|
||
"Installed serialized command gateway (min interval %.0fms)",
|
||
self._radio_cmd_min_interval * 1000,
|
||
)
|
||
except (AttributeError, TypeError) as e:
|
||
self.logger.warning(f"Could not install command serializer: {e}")
|
||
|
||
async def connect(self) -> bool:
|
||
"""Connect to MeshCore node using official package.
|
||
|
||
Establishes a connection to the mesh node via Serial, TCP, or BLE
|
||
based on the configuration.
|
||
|
||
Returns:
|
||
bool: True if connection was successful, False otherwise.
|
||
"""
|
||
new_meshcore = None
|
||
connection_ready = False
|
||
try:
|
||
self.logger.info("Connecting to MeshCore node...")
|
||
|
||
# Get connection type from config
|
||
connection_type = self.config.get('Connection', 'connection_type', fallback='ble').lower()
|
||
# radio_debug: config.ini baseline, overridden by bot_metadata (set via web UI)
|
||
radio_debug = self.config.getboolean('Connection', 'radio_debug', fallback=False)
|
||
try:
|
||
meta_val = self.db_manager.get_metadata('radio.debug')
|
||
if meta_val == 'true':
|
||
radio_debug = True
|
||
elif meta_val == 'false':
|
||
radio_debug = False
|
||
except Exception:
|
||
pass
|
||
self.logger.info(f"Using connection type: {connection_type}")
|
||
if radio_debug:
|
||
self.logger.info("Radio debug logging enabled — meshcore library output will be at DEBUG level")
|
||
|
||
if connection_type == 'serial':
|
||
# Create serial connection
|
||
serial_port = self.config.get('Connection', 'serial_port', fallback='/dev/ttyUSB0')
|
||
self.logger.info(f"Connecting via serial port: {serial_port}")
|
||
new_meshcore = await meshcore.MeshCore.create_serial(serial_port, debug=radio_debug)
|
||
elif connection_type == 'tcp':
|
||
# Create TCP connection
|
||
hostname = self.config.get('Connection', 'hostname', fallback=None)
|
||
tcp_port = self.config.getint('Connection', 'tcp_port', fallback=5000)
|
||
if not hostname:
|
||
self.logger.error("TCP connection requires 'hostname' to be set in config")
|
||
return False
|
||
self.logger.info(f"Connecting via TCP: {hostname}:{tcp_port}")
|
||
new_meshcore = await meshcore.MeshCore.create_tcp(hostname, tcp_port, debug=radio_debug)
|
||
else:
|
||
# Create BLE connection (default)
|
||
ble_device_name = self.config.get('Connection', 'ble_device_name', fallback=None)
|
||
self.logger.info("Connecting via BLE" + (f" to device: {ble_device_name}" if ble_device_name else ""))
|
||
new_meshcore = await meshcore.MeshCore.create_ble(ble_device_name, debug=radio_debug)
|
||
|
||
self.meshcore = new_meshcore
|
||
|
||
# Route meshcore library output through the bot's handlers (including log file)
|
||
self._configure_meshcore_debug_logging(radio_debug)
|
||
|
||
# Serialize all host->radio commands before issuing any (the connect
|
||
# init burst — contacts, channel fetch, clock, name — runs through it).
|
||
self._install_command_serializer()
|
||
|
||
if self.meshcore and self.meshcore.is_connected:
|
||
self.connected = True
|
||
self._update_radio_connected_metadata(True)
|
||
# Track connection time to skip processing old cached messages
|
||
self.connection_time = time.time()
|
||
# Clear zombie state — a successful connect means the radio is alive again
|
||
self._radio_zombie_detected = False
|
||
self._radio_fail_count = 0
|
||
self._tcp_probe_fail_count = 0
|
||
try:
|
||
self.db_manager.set_metadata('bot.radio_zombie', 'false')
|
||
self.db_manager.set_metadata('bot.radio_zombie_since', '')
|
||
except Exception:
|
||
pass
|
||
self.logger.info(f"Connected to: {self.meshcore.self_info} at {self.connection_time}")
|
||
|
||
# Wait for contacts to load
|
||
await self.wait_for_contacts()
|
||
|
||
# A connected transport without channel data cannot route replies.
|
||
if not await self.channel_manager.fetch_channels():
|
||
raise ConnectionError(
|
||
"MeshCore node returned no channels after retries"
|
||
)
|
||
|
||
# Setup message event handlers
|
||
await self.setup_message_handlers()
|
||
|
||
# Set radio clock if needed
|
||
await self.set_radio_clock()
|
||
|
||
# Set device name to match config if needed
|
||
await self.set_device_name()
|
||
|
||
await self._notify_services_transport_reconnected()
|
||
|
||
connection_ready = True
|
||
return True
|
||
else:
|
||
self.logger.error("Failed to connect to MeshCore node")
|
||
return False
|
||
|
||
except (OSError, ConnectionError, TimeoutError, ValueError, AttributeError) as e:
|
||
self.logger.error(f"Connection failed: {e}")
|
||
return False
|
||
finally:
|
||
if not connection_ready:
|
||
self.connected = False
|
||
self._update_radio_connected_metadata(False)
|
||
if new_meshcore is not None:
|
||
try:
|
||
await asyncio.wait_for(new_meshcore.disconnect(), timeout=5.0)
|
||
except Exception as e:
|
||
self.logger.warning(
|
||
"Could not clean up incomplete MeshCore connection: %s",
|
||
e,
|
||
)
|
||
finally:
|
||
if self.meshcore is new_meshcore:
|
||
self.meshcore = None
|
||
|
||
async def _notify_services_transport_reconnected(self) -> None:
|
||
"""Re-bind mesh event subscriptions on running services after transport reconnect."""
|
||
services = getattr(self, 'services', None) or {}
|
||
for name, service in services.items():
|
||
if not service.is_running():
|
||
continue
|
||
try:
|
||
await service.on_transport_reconnected()
|
||
except Exception as e:
|
||
self.logger.error(
|
||
"Service '%s' on_transport_reconnected failed: %s",
|
||
name,
|
||
e,
|
||
exc_info=True,
|
||
)
|
||
|
||
def _update_radio_connected_metadata(self, connected: bool) -> None:
|
||
"""Write radio connection state to bot_metadata for the web viewer."""
|
||
try:
|
||
self.db_manager.set_metadata('radio_connected', '1' if connected else '0')
|
||
except Exception as e:
|
||
self.logger.warning(f"Could not update radio_connected metadata: {e}")
|
||
|
||
async def disconnect_radio(self) -> bool:
|
||
"""Disconnect from the radio, which also stops the bot.
|
||
|
||
Despite the name, this is not a radio-only operation: ``run()``ing loops
|
||
while ``keep_running`` is true. This operation clears ``connected`` without
|
||
setting a reconnect/relink flag, so it ends the main loop and the process
|
||
exits. The web viewer therefore labels the control "Stop Bot" and confirms
|
||
first (issue #240). Keep that in mind before calling this from anywhere that
|
||
only means to drop the radio link.
|
||
|
||
Called by the scheduler via the operation queue.
|
||
"""
|
||
import asyncio
|
||
try:
|
||
if self.meshcore:
|
||
try:
|
||
await asyncio.wait_for(self.meshcore.disconnect(), timeout=10)
|
||
except asyncio.TimeoutError:
|
||
self.logger.warning("Radio disconnect timed out after 10s — forcing disconnected state")
|
||
self.connected = False
|
||
self._update_radio_connected_metadata(False)
|
||
self.logger.info("Radio disconnected via web viewer request")
|
||
return True
|
||
except Exception as e:
|
||
self.logger.error(f"Error disconnecting radio: {e}")
|
||
return False
|
||
|
||
async def reboot_radio(self) -> bool:
|
||
"""Send firmware reboot command, disconnect, wait for reboot, then reconnect."""
|
||
import asyncio
|
||
# Hold the loops open (see keep_running) while connected is False
|
||
self._radio_relinks_in_progress += 1
|
||
try:
|
||
if self.meshcore and self.meshcore.is_connected:
|
||
self.logger.info("Sending firmware reboot command")
|
||
try:
|
||
await asyncio.wait_for(self.meshcore.commands.reboot(), timeout=5)
|
||
except (asyncio.TimeoutError, Exception) as e:
|
||
# Reboot command may drop the connection before a reply arrives
|
||
self.logger.debug(f"Reboot command response: {e} (expected on firmware reboot)")
|
||
# Disconnect cleanly (firmware may have already dropped the link)
|
||
try:
|
||
if self.meshcore:
|
||
await asyncio.wait_for(self.meshcore.disconnect(), timeout=5)
|
||
except (asyncio.TimeoutError, Exception):
|
||
pass
|
||
self.connected = False
|
||
self._update_radio_connected_metadata(False)
|
||
self.logger.info("Waiting for radio to reboot (8s)…")
|
||
await asyncio.sleep(8)
|
||
return await self.connect()
|
||
except Exception as e:
|
||
self.logger.error(f"Error rebooting radio: {e}")
|
||
return False
|
||
finally:
|
||
self._radio_relinks_in_progress -= 1
|
||
|
||
async def reconnect_radio(self) -> bool:
|
||
"""Disconnect then reconnect. Called by scheduler for connect ops."""
|
||
import asyncio
|
||
# Hold the loops open (see keep_running) while connected is False
|
||
self._radio_relinks_in_progress += 1
|
||
try:
|
||
if self.meshcore:
|
||
try:
|
||
await asyncio.wait_for(self.meshcore.disconnect(), timeout=10)
|
||
except asyncio.TimeoutError:
|
||
self.logger.warning("Disconnect timed out during reconnect — proceeding")
|
||
self.connected = False
|
||
self._update_radio_connected_metadata(False)
|
||
return await self.connect()
|
||
except Exception as e:
|
||
self.logger.error(f"Error reconnecting radio: {e}")
|
||
return False
|
||
finally:
|
||
self._radio_relinks_in_progress -= 1
|
||
|
||
def _handle_serial_probe_error(self, threshold: int, interval: int) -> bool:
|
||
"""Serial/BLE: failed get_time may indicate zombie firmware (no transport reconnect)."""
|
||
import datetime as _dt
|
||
|
||
self._radio_fail_count = self._radio_fail_count + 1
|
||
self.logger.warning(
|
||
"Radio health probe failed "
|
||
"(%d/%d): no response to get_time",
|
||
self._radio_fail_count,
|
||
threshold,
|
||
)
|
||
if self._radio_fail_count >= threshold:
|
||
fail_count = self._radio_fail_count
|
||
self._radio_fail_count = 0
|
||
self._radio_zombie_detected = True
|
||
self.logger.critical(
|
||
"ZOMBIE RADIO DETECTED after %d consecutive failed probes "
|
||
"(probe interval %ds). The radio firmware is unresponsive. "
|
||
"A physical POWER CYCLE is required — disconnect/reconnect "
|
||
"will NOT fix this. Probing suspended until next reconnect.",
|
||
fail_count,
|
||
interval,
|
||
)
|
||
try:
|
||
self.db_manager.set_metadata('bot.radio_zombie', 'true')
|
||
self.db_manager.set_metadata(
|
||
'bot.radio_zombie_since',
|
||
_dt.datetime.now(_dt.timezone.utc).isoformat(),
|
||
)
|
||
except Exception:
|
||
pass
|
||
scheduler = getattr(self, 'scheduler', None)
|
||
if scheduler is not None:
|
||
loop = asyncio.get_event_loop()
|
||
loop.run_in_executor(
|
||
None,
|
||
scheduler.send_zombie_alert_email,
|
||
fail_count,
|
||
threshold,
|
||
interval,
|
||
)
|
||
return False
|
||
|
||
async def _handle_tcp_probe_failure(
|
||
self, threshold: int, interval: int, detail: str
|
||
) -> bool:
|
||
"""TCP: repeated probe failures trigger transport reconnect, not zombie."""
|
||
self._tcp_probe_fail_count = self._tcp_probe_fail_count + 1
|
||
self.logger.warning(
|
||
"TCP radio health probe failed (%d/%d): %s",
|
||
self._tcp_probe_fail_count,
|
||
threshold,
|
||
detail,
|
||
)
|
||
if self._tcp_probe_fail_count >= threshold:
|
||
self._tcp_probe_fail_count = 0
|
||
self.logger.warning(
|
||
"TCP transport unresponsive after %d probes (interval %ds) — reconnecting",
|
||
threshold,
|
||
interval,
|
||
)
|
||
await self._schedule_transport_reconnect('tcp_probe_failed')
|
||
return False
|
||
|
||
async def _probe_radio_health(self) -> bool:
|
||
"""Send a lightweight get_time() probe to verify the radio is responding.
|
||
|
||
Serial/BLE: repeated ERROR responses declare zombie firmware (no reconnect).
|
||
TCP: repeated ERROR or timeout responses schedule transport reconnect.
|
||
"""
|
||
if self._radio_zombie_detected:
|
||
return False
|
||
|
||
if not self.meshcore or not self.meshcore.is_connected:
|
||
await self._schedule_transport_reconnect('probe_not_connected')
|
||
return False
|
||
|
||
is_tcp = self._connection_type() == 'tcp'
|
||
threshold = self._radio_probe_fail_threshold()
|
||
interval = self._radio_probe_interval_seconds()
|
||
|
||
try:
|
||
result = await asyncio.wait_for(
|
||
self.meshcore.commands.get_time(), timeout=10.0
|
||
)
|
||
if result.type == EventType.ERROR:
|
||
if is_tcp:
|
||
return await self._handle_tcp_probe_failure(
|
||
threshold, interval, 'no response to get_time'
|
||
)
|
||
return self._handle_serial_probe_error(threshold, interval)
|
||
|
||
if self._radio_fail_count > 0:
|
||
self.logger.info("Radio health probe recovered — resetting fail counter")
|
||
self._radio_fail_count = 0
|
||
self._tcp_probe_fail_count = 0
|
||
self._allow_offline_trial("the radio answered a health probe")
|
||
return True
|
||
except asyncio.TimeoutError:
|
||
if is_tcp:
|
||
return await self._handle_tcp_probe_failure(
|
||
threshold, interval, 'probe timed out'
|
||
)
|
||
self.logger.warning("Radio health probe timed out")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.warning(f"Radio health probe error: {e}")
|
||
return False
|