Fix firmware-compatible unscoped flood scope handling

This commit is contained in:
Björkan
2026-06-28 01:23:19 +02:00
parent ce4946351f
commit 55020a5e22
11 changed files with 134 additions and 21 deletions
+8 -5
View File
@@ -1,19 +1,22 @@
"""Helpers for normalizing MeshCore flood-scope / region names."""
_UNSCOPED_SENTINELS = {"", "0", "*"}
def normalize_region_scope(scope: str | None) -> str:
"""Normalize a user-facing region scope into MeshCore's internal form.
Region names are now user-facing plain strings like ``Esperance``.
Internally, MeshCore still expects hashtag-style names like ``#Esperance``.
Region names are user-facing plain strings like ``Esperance``. Internally,
MeshCore still expects hashtag-style names like ``#Esperance``.
Backward compatibility:
- blank/None stays disabled (`""`)
Backward compatibility / firmware parity:
- blank/None stays unscoped (``""``)
- ``"0"`` and ``"*"`` also mean explicit unscoped/plain flood
- existing leading ``#`` is preserved
"""
stripped = (scope or "").strip()
if not stripped:
if stripped in _UNSCOPED_SENTINELS:
return ""
if stripped.startswith("#"):
return stripped
+2 -1
View File
@@ -282,13 +282,14 @@ async def update_settings(update: AppSettingsUpdate) -> AppSettings:
# Apply flood scope to radio immediately if changed
if flood_scope_changed:
from app.services.flood_scope import set_radio_flood_scope
from app.services.radio_runtime import radio_runtime as radio_manager
if radio_manager.is_connected:
try:
scope = result.flood_scope
async with radio_manager.radio_operation("set_flood_scope") as mc:
await mc.commands.set_flood_scope(scope if scope else "")
await set_radio_flood_scope(mc, scope)
logger.info("Applied flood_scope=%r to radio", scope or "(disabled)")
except Exception as e:
logger.warning("Failed to apply flood_scope to radio: %s", e)
+32
View File
@@ -0,0 +1,32 @@
"""Firmware-compatible flood-scope command helpers."""
from typing import Any
from meshcore import EventType
from meshcore.packets import CommandType
from app.region_scope import normalize_region_scope
SET_FLOOD_SCOPE_MODE_UNSCOPED = 1
FORCE_UNSCOPED_FRAME = bytes([CommandType.SET_FLOOD_SCOPE.value, SET_FLOOD_SCOPE_MODE_UNSCOPED])
async def set_radio_flood_scope(mc, scope: str | 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 and must use firmware's
mode-1 command. Sending a mode-0 all-zero key would fall back to the radio's
configured default scope on firmware-compatible companions.
"""
normalized_scope = normalize_region_scope(scope)
if normalized_scope:
return await mc.commands.set_flood_scope(normalized_scope)
return await force_radio_unscoped(mc)
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])
+3 -4
View File
@@ -19,6 +19,7 @@ from app.repository import (
MessageRepository,
)
from app.services import dm_ack_tracker
from app.services.flood_scope import set_radio_flood_scope
from app.services.messages import (
BroadcastFn,
broadcast_message,
@@ -184,7 +185,7 @@ async def send_channel_message_with_effective_scope(
desired_scope or "(unscoped)",
channel.name,
)
override_result = await mc.commands.set_flood_scope(desired_scope)
override_result = await set_radio_flood_scope(mc, desired_scope)
if override_result is not None and override_result.type == EventType.ERROR:
logger.warning(
"Failed to apply flood_scope %r for %s: %s",
@@ -313,9 +314,7 @@ async def send_channel_message_with_effective_scope(
restored = False
for attempt in range(3):
try:
restore_result = await mc.commands.set_flood_scope(
baseline_scope if baseline_scope else ""
)
restore_result = await set_radio_flood_scope(mc, baseline_scope)
if restore_result is not None and restore_result.type == EventType.ERROR:
logger.warning(
"Attempt %d/3: failed to restore flood_scope after sending to %s: %s",
+2 -1
View File
@@ -71,11 +71,12 @@ async def run_post_connect_setup(radio_manager) -> None:
# may not support set_flood_scope)
from app.region_scope import normalize_region_scope
from app.repository import AppSettingsRepository
from app.services.flood_scope import set_radio_flood_scope
app_settings = await AppSettingsRepository.get()
scope = normalize_region_scope(app_settings.flood_scope)
try:
await mc.commands.set_flood_scope(scope if scope else "")
await set_radio_flood_scope(mc, scope)
logger.info("Applied flood_scope=%r", scope or "(disabled)")
except Exception as exc:
logger.warning("set_flood_scope failed (firmware may not support it): %s", exc)