Files
Remote-Terminal-for-MeshCore/app/services/flood_scope.py
T

61 lines
2.4 KiB
Python

"""Firmware-compatible flood-scope command helpers."""
import logging
from typing import Any
from meshcore import EventType
from meshcore.packets import CommandType
from app.region_scope import normalize_region_scope
logger = logging.getLogger(__name__)
SET_FLOOD_SCOPE_MODE_UNSCOPED = 1
FORCE_UNSCOPED_FRAME = bytes([CommandType.SET_FLOOD_SCOPE.value, SET_FLOOD_SCOPE_MODE_UNSCOPED])
# CMD_SET_FLOOD_SCOPE_KEY mode 1 (the firmware ``send_unscoped`` flag) is companion
# firmware ver 12+. On older firmware the mode-1 frame is rejected, so we fall back
# to resetting the scope override (mode 0, zero key), which makes the radio use its
# configured default scope. True "unscoped while a default scope is set" is not
# achievable pre-v12.
FIRMWARE_VER_UNSCOPED_MODE = 12
def firmware_supports_unscoped_mode(fw_ver: int | None) -> bool:
"""Whether the radio's protocol version supports the mode-1 unscoped command."""
return fw_ver is not None and fw_ver >= FIRMWARE_VER_UNSCOPED_MODE
async def set_radio_flood_scope(mc, scope: str | None, *, fw_ver: int | None = None) -> Any:
"""Apply the standing radio flood-scope state.
A non-empty scope is delegated to meshcore_py's mode-0 ``set_flood_scope``. An
empty scope means explicit unscoped/plain flood:
- firmware >= 12: use the dedicated mode-1 command (``force_radio_unscoped``).
- older / unknown firmware: no dedicated unscoped command exists, so reset the
scope override (mode 0, zero key); the radio falls back to its configured
default scope. ``fw_ver=None`` (version unknown) is treated conservatively as
unsupported so we never emit a frame the radio might reject.
"""
normalized_scope = normalize_region_scope(scope)
if normalized_scope:
return await mc.commands.set_flood_scope(normalized_scope)
if firmware_supports_unscoped_mode(fw_ver):
return await force_radio_unscoped(mc)
logger.debug(
"Radio fw_ver=%s < %d: no dedicated unscoped command; resetting scope override "
"(radio falls back to its configured default scope)",
fw_ver,
FIRMWARE_VER_UNSCOPED_MODE,
)
return await mc.commands.set_flood_scope("")
async def force_radio_unscoped(mc) -> Any:
"""Tell the radio to send following flood packets unscoped until mode 0."""
return await mc.commands.send(FORCE_UNSCOPED_FRAME, [EventType.OK, EventType.ERROR])