adam/Remote-Terminal-for-MeshCore
1
0
Fork 0
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-08-07 09:13:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

Linting and code cleanup for an imitation of order

Browse Source
This commit is contained in:
Jack Kingsman
2026-01-14 20:08:41 -08:00
parent c6d38ce400
commit 076d466fbd
73 changed files with 2799 additions and 1192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/database.py
+1
View File
@@ -79,6 +79,7 @@ class Database:
# Run any pending migrations
from app.migrations import run_migrations
await run_migrations(self._connection)
async def disconnect(self) -> None:
app/decoder.py
+5 -7
View File
@@ -3,8 +3,8 @@ MeshCore packet decoder for historical packet decryption.
Based on https://github.com/michaelhart/meshcore-decoder
"""
import hmac
import hashlib
import hmac
import logging
from dataclasses import dataclass
from enum import IntEnum
@@ -153,7 +153,7 @@ def parse_packet(raw_packet: bytes) -> PacketInfo | None:
# Extract path data
if len(raw_packet) < offset + path_length:
return None
path = raw_packet[offset:offset + path_length]
path = raw_packet[offset : offset + path_length]
offset += path_length
# Rest is payload
@@ -171,9 +171,7 @@ def parse_packet(raw_packet: bytes) -> PacketInfo | None:
return None
def decrypt_group_text(
payload: bytes, channel_key: bytes
) -> DecryptedGroupText | None:
def decrypt_group_text(payload: bytes, channel_key: bytes) -> DecryptedGroupText | None:
"""
Decrypt a GroupText payload using the channel key.
@@ -344,8 +342,8 @@ def parse_advertisement(payload: bytes) -> ParsedAdvertisement | None:
lon=None,
device_role=device_role,
)
lat_raw = int.from_bytes(payload[offset:offset + 4], byteorder="little", signed=True)
lon_raw = int.from_bytes(payload[offset + 4:offset + 8], byteorder="little", signed=True)
lat_raw = int.from_bytes(payload[offset : offset + 4], byteorder="little", signed=True)
lon_raw = int.from_bytes(payload[offset + 4 : offset + 8], byteorder="little", signed=True)
lat = lat_raw / 1_000_000
lon = lon_raw / 1_000_000
offset += 8
app/event_handlers.py
+24 -16
View File
@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING
from meshcore import EventType
from app.models import Contact
from app.packet_processor import process_raw_packet, track_pending_repeat
from app.packet_processor import process_raw_packet
from app.repository import ContactRepository, MessageRepository
from app.websocket import broadcast_event
@@ -22,14 +22,19 @@ _pending_acks: dict[str, tuple[int, float, int]] = {}
def track_pending_ack(expected_ack: str, message_id: int, timeout_ms: int) -> None:
"""Track a pending ACK for a direct message."""
_pending_acks[expected_ack] = (message_id, time.time(), timeout_ms)
logger.debug("Tracking pending ACK %s for message %d (timeout %dms)", expected_ack, message_id, timeout_ms)
logger.debug(
"Tracking pending ACK %s for message %d (timeout %dms)",
expected_ack,
message_id,
timeout_ms,
)
def _cleanup_expired_acks() -> None:
"""Remove expired pending ACKs."""
now = time.time()
expired = []
for code, (msg_id, created_at, timeout_ms) in _pending_acks.items():
for code, (_msg_id, created_at, timeout_ms) in _pending_acks.items():
if now - created_at > (timeout_ms / 1000) * 2: # 2x timeout as buffer
expired.append(code)
for code in expired:
@@ -82,19 +87,22 @@ async def on_contact_message(event: "Event") -> None:
return
# Broadcast only genuinely new messages
broadcast_event("message", {
"id": msg_id,
"type": "PRIV",
"conversation_key": sender_pubkey,
"text": payload.get("text", ""),
"sender_timestamp": payload.get("sender_timestamp"),
"received_at": received_at,
"path_len": payload.get("path_len"),
"txt_type": payload.get("txt_type", 0),
"signature": payload.get("signature"),
"outgoing": False,
"acked": False,
})
broadcast_event(
"message",
{
"id": msg_id,
"type": "PRIV",
"conversation_key": sender_pubkey,
"text": payload.get("text", ""),
"sender_timestamp": payload.get("sender_timestamp"),
"received_at": received_at,
"path_len": payload.get("path_len"),
"txt_type": payload.get("txt_type", 0),
"signature": payload.get("signature"),
"outgoing": False,
"acked": False,
},
)
# Update contact last_seen and last_contacted
contact = await ContactRepository.get_by_key_prefix(sender_pubkey)
app/main.py
+11 -1
View File
@@ -20,7 +20,17 @@ from app.radio_sync import (
sync_and_offload_all,
sync_radio_time,
)
from app.routers import channels, contacts, health, messages, packets, radio, read_state, settings, ws
from app.routers import (
channels,
contacts,
health,
messages,
packets,
radio,
read_state,
settings,
ws,
)
setup_logging()
logger = logging.getLogger(__name__)
app/migrations.py
+3 -1
View File
@@ -50,7 +50,9 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
# applied += 1
if applied > 0:
logger.info("Applied %d migration(s), schema now at version %d", applied, await get_version(conn))
logger.info(
"Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
)
else:
logger.debug("Schema up to date at version %d", version)
app/models.py
+20 -4
View File
@@ -83,6 +83,7 @@ class Message(BaseModel):
class RawPacket(BaseModel):
"""Raw packet as stored in the database."""
id: int
timestamp: int
data: str = Field(description="Hex-encoded packet data")
@@ -94,6 +95,7 @@ class RawPacket(BaseModel):
class RawPacketDecryptedInfo(BaseModel):
"""Decryption info for a raw packet (when successfully decrypted)."""
channel_name: str | None = None
sender: str | None = None
@@ -104,6 +106,7 @@ class RawPacketBroadcast(BaseModel):
This extends the database model with runtime-computed fields
like payload_type, snr, rssi, and decryption info.
"""
id: int
timestamp: int
data: str = Field(description="Hex-encoded packet data")
@@ -127,11 +130,14 @@ class SendChannelMessageRequest(SendMessageRequest):
class TelemetryRequest(BaseModel):
password: str = Field(default="", description="Repeater password (empty string for no password)")
password: str = Field(
default="", description="Repeater password (empty string for no password)"
)
class NeighborInfo(BaseModel):
"""Information about a neighbor seen by a repeater."""
pubkey_prefix: str = Field(description="Public key prefix (4-12 chars)")
name: str | None = Field(default=None, description="Resolved contact name if known")
snr: float = Field(description="Signal-to-noise ratio in dB")
@@ -140,14 +146,18 @@ class NeighborInfo(BaseModel):
class AclEntry(BaseModel):
"""Access control list entry for a repeater."""
pubkey_prefix: str = Field(description="Public key prefix (12 chars)")
name: str | None = Field(default=None, description="Resolved contact name if known")
permission: int = Field(description="Permission level: 0=Guest, 1=Read-only, 2=Read-write, 3=Admin")
permission: int = Field(
description="Permission level: 0=Guest, 1=Read-only, 2=Read-write, 3=Admin"
)
permission_name: str = Field(description="Human-readable permission name")
class TelemetryResponse(BaseModel):
"""Telemetry data from a repeater, formatted for human readability."""
pubkey_prefix: str = Field(description="12-char public key prefix")
battery_volts: float = Field(description="Battery voltage in volts")
tx_queue_len: int = Field(description="Transmit queue length")
@@ -166,17 +176,23 @@ class TelemetryResponse(BaseModel):
flood_dups: int = Field(description="Duplicate flood packets")
direct_dups: int = Field(description="Duplicate direct packets")
full_events: int = Field(description="Full event queue count")
neighbors: list[NeighborInfo] = Field(default_factory=list, description="List of neighbors seen by repeater")
neighbors: list[NeighborInfo] = Field(
default_factory=list, description="List of neighbors seen by repeater"
)
acl: list[AclEntry] = Field(default_factory=list, description="Access control list")
class CommandRequest(BaseModel):
"""Request to send a CLI command to a repeater."""
command: str = Field(min_length=1, description="CLI command to send")
class CommandResponse(BaseModel):
"""Response from a repeater CLI command."""
command: str = Field(description="The command that was sent")
response: str = Field(description="Response from the repeater")
sender_timestamp: int | None = Field(default=None, description="Timestamp from the repeater's response")
sender_timestamp: int | None = Field(
default=None, description="Timestamp from the repeater's response"
)
app/packet_processor.py
+63 -39
View File
@@ -18,12 +18,17 @@ import time
from app.decoder import (
PayloadType,
parse_packet,
parse_advertisement,
parse_packet,
try_decrypt_packet_with_channel_key,
)
from app.models import CONTACT_TYPE_REPEATER, RawPacketBroadcast, RawPacketDecryptedInfo
from app.repository import ChannelRepository, ContactRepository, MessageRepository, RawPacketRepository
from app.repository import (
ChannelRepository,
ContactRepository,
MessageRepository,
RawPacketRepository,
)
from app.websocket import broadcast_event
logger = logging.getLogger(__name__)
@@ -62,6 +67,7 @@ async def create_message_from_decrypted(
Returns the message ID if created, None if duplicate.
"""
import time as time_module
received = received_at or int(time_module.time())
# Format the message text with sender prefix if present
@@ -88,7 +94,8 @@ async def create_message_from_decrypted(
)
logger.debug(
"Duplicate message detected for channel %s (existing id=%s)",
channel_key_normalized[:8], existing_id
channel_key_normalized[:8],
existing_id,
)
if existing_id:
await RawPacketRepository.mark_decrypted(packet_id, existing_id)
@@ -100,19 +107,22 @@ async def create_message_from_decrypted(
await RawPacketRepository.mark_decrypted(packet_id, msg_id)
# Broadcast new message to connected clients
broadcast_event("message", {
"id": msg_id,
"type": "CHAN",
"conversation_key": channel_key_normalized,
"text": text,
"sender_timestamp": timestamp,
"received_at": received,
"path_len": path_len,
"txt_type": 0,
"signature": None,
"outgoing": False,
"acked": 0,
})
broadcast_event(
"message",
{
"id": msg_id,
"type": "CHAN",
"conversation_key": channel_key_normalized,
"text": text,
"sender_timestamp": timestamp,
"received_at": received,
"path_len": path_len,
"txt_type": 0,
"signature": None,
"outgoing": False,
"acked": 0,
},
)
return msg_id
@@ -196,7 +206,9 @@ async def process_raw_packet(
decrypted_info=RawPacketDecryptedInfo(
channel_name=result["channel_name"],
sender=result["sender"],
) if result["decrypted"] else None,
)
if result["decrypted"]
else None,
)
broadcast_event("raw_packet", broadcast_payload.model_dump())
@@ -231,10 +243,7 @@ async def _process_group_text(
continue
# Successfully decrypted!
logger.debug(
"Decrypted GroupText for channel %s: %s",
channel.name, decrypted.message[:50]
)
logger.debug("Decrypted GroupText for channel %s: %s", channel.name, decrypted.message[:50])
# Check for repeat detection (our own message echoed back)
is_repeat = False
@@ -322,18 +331,22 @@ async def _process_advertisement(
if existing and existing.last_seen:
path_age = timestamp - existing.last_seen
existing_path_len = existing.last_path_len if existing.last_path_len >= 0 else float('inf')
existing_path_len = existing.last_path_len if existing.last_path_len >= 0 else float("inf")
# Keep existing path if it's fresh and shorter (or equal)
if path_age <= PATH_FRESHNESS_SECONDS and existing_path_len <= new_path_len:
use_existing_path = True
logger.debug(
"Keeping existing shorter path for %s (existing=%d, new=%d, age=%ds)",
advert.public_key[:12], existing_path_len, new_path_len, path_age
advert.public_key[:12],
existing_path_len,
new_path_len,
path_age,
)
if use_existing_path:
path_len = existing.last_path_len
assert existing is not None # Guaranteed by the conditions that set use_existing_path
path_len = existing.last_path_len if existing.last_path_len is not None else -1
path_hex = existing.last_path or ""
else:
path_len = new_path_len
@@ -341,12 +354,19 @@ async def _process_advertisement(
logger.debug(
"Parsed advertisement from %s: %s (role=%d, lat=%s, lon=%s, path_len=%d)",
advert.public_key[:12], advert.name, advert.device_role, advert.lat, advert.lon, path_len
advert.public_key[:12],
advert.name,
advert.device_role,
advert.lat,
advert.lon,
path_len,
)
# Use device_role from advertisement for contact type (1=Chat, 2=Repeater, 3=Room, 4=Sensor)
# Use advert.timestamp for last_advert (sender's timestamp), receive timestamp for last_seen
contact_type = advert.device_role if advert.device_role > 0 else (existing.type if existing else 0)
contact_type = (
advert.device_role if advert.device_role > 0 else (existing.type if existing else 0)
)
contact_data = {
"public_key": advert.public_key,
@@ -363,23 +383,27 @@ async def _process_advertisement(
await ContactRepository.upsert(contact_data)
# Broadcast contact update to connected clients
broadcast_event("contact", {
"public_key": advert.public_key,
"name": advert.name,
"type": contact_type,
"flags": existing.flags if existing else 0,
"last_path": path_hex,
"last_path_len": path_len,
"last_advert": advert.timestamp if advert.timestamp > 0 else timestamp,
"lat": advert.lat,
"lon": advert.lon,
"last_seen": timestamp,
"on_radio": existing.on_radio if existing else False,
})
broadcast_event(
"contact",
{
"public_key": advert.public_key,
"name": advert.name,
"type": contact_type,
"flags": existing.flags if existing else 0,
"last_path": path_hex,
"last_path_len": path_len,
"last_advert": advert.timestamp if advert.timestamp > 0 else timestamp,
"lat": advert.lat,
"lon": advert.lon,
"last_seen": timestamp,
"on_radio": existing.on_radio if existing else False,
},
)
# If this is not a repeater, trigger recent contacts sync to radio
# This ensures we can auto-ACK DMs from recent contacts
if contact_type != CONTACT_TYPE_REPEATER:
# Import here to avoid circular import
from app.radio_sync import sync_recent_contacts_to_radio
asyncio.create_task(sync_recent_contacts_to_radio())
app/radio.py
+1
View File
@@ -231,6 +231,7 @@ class RadioManager:
if await self.reconnect():
# Re-register event handlers after successful reconnect
from app.event_handlers import register_event_handlers
if self._meshcore:
register_event_handlers(self._meshcore)
await self._meshcore.start_auto_message_fetching()
app/radio_sync.py
+8 -10
View File
@@ -17,7 +17,7 @@ from contextlib import asynccontextmanager
from meshcore import EventType
from app.config import settings
from app.models import CONTACT_TYPE_REPEATER, Contact
from app.models import Contact
from app.radio import radio_manager
from app.repository import ChannelRepository, ContactRepository
@@ -51,6 +51,7 @@ async def pause_polling():
finally:
_polling_pause_count -= 1
# Background task handle
_sync_task: asyncio.Task | None = None
@@ -97,8 +98,7 @@ async def sync_and_offload_contacts() -> dict:
removed += 1
else:
logger.warning(
"Failed to remove contact %s: %s",
public_key[:12], remove_result.payload
"Failed to remove contact %s: %s", public_key[:12], remove_result.payload
)
except Exception as e:
logger.warning("Error removing contact %s: %s", public_key[:12], e)
@@ -167,10 +167,7 @@ async def sync_and_offload_channels() -> dict:
if clear_result.type == EventType.OK:
cleared += 1
else:
logger.warning(
"Failed to clear channel %d: %s",
idx, clear_result.payload
)
logger.warning("Failed to clear channel %d: %s", idx, clear_result.payload)
except Exception as e:
logger.warning("Error clearing channel %d: %s", idx, e)
@@ -446,8 +443,7 @@ async def sync_recent_contacts_to_radio(force: bool = False) -> dict:
else:
failed += 1
logger.warning(
"Failed to load contact %s: %s",
contact.public_key[:12], result.payload
"Failed to load contact %s: %s", contact.public_key[:12], result.payload
)
except Exception as e:
failed += 1
@@ -456,7 +452,9 @@ async def sync_recent_contacts_to_radio(force: bool = False) -> dict:
if loaded > 0 or failed > 0:
logger.info(
"Contact sync: loaded %d, already on radio %d, failed %d",
loaded, already_on_radio, failed
loaded,
already_on_radio,
failed,
)
return {
app/repository.py
+21 -14
View File
@@ -32,7 +32,9 @@ class ContactRepository:
contact.get("type", 0),
contact.get("flags", 0),
contact.get("last_path") or contact.get("out_path"),
contact.get("last_path_len") if "last_path_len" in contact else contact.get("out_path_len", -1),
contact.get("last_path_len")
if "last_path_len" in contact
else contact.get("out_path_len", -1),
contact.get("last_advert"),
contact.get("lat") or contact.get("adv_lat"),
contact.get("lon") or contact.get("adv_lon"),
@@ -64,9 +66,7 @@ class ContactRepository:
@staticmethod
async def get_by_key(public_key: str) -> Contact | None:
cursor = await db.conn.execute(
"SELECT * FROM contacts WHERE public_key = ?", (public_key,)
)
cursor = await db.conn.execute("SELECT * FROM contacts WHERE public_key = ?", (public_key,))
row = await cursor.fetchone()
return ContactRepository._row_to_contact(row) if row else None
@@ -207,7 +207,7 @@ class ChannelRepository:
"""Get a channel by its key (32-char hex string)."""
cursor = await db.conn.execute(
"SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels WHERE key = ?",
(key.upper(),)
(key.upper(),),
)
row = await cursor.fetchone()
if row:
@@ -241,7 +241,8 @@ class ChannelRepository:
async def get_by_name(name: str) -> Channel | None:
"""Get a channel by name."""
cursor = await db.conn.execute(
"SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels WHERE name = ?", (name,)
"SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels WHERE name = ?",
(name,),
)
row = await cursor.fetchone()
if row:
@@ -303,8 +304,17 @@ class MessageRepository:
received_at, path_len, txt_type, signature, outgoing)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(msg_type, conversation_key, text, sender_timestamp, received_at,
path_len, txt_type, signature, outgoing),
(
msg_type,
conversation_key,
text,
sender_timestamp,
received_at,
path_len,
txt_type,
signature,
outgoing,
),
)
await db.conn.commit()
# rowcount is 0 if INSERT was ignored due to UNIQUE constraint violation
@@ -355,13 +365,9 @@ class MessageRepository:
@staticmethod
async def increment_ack_count(message_id: int) -> int:
"""Increment ack count and return the new value."""
await db.conn.execute(
"UPDATE messages SET acked = acked + 1 WHERE id = ?", (message_id,)
)
await db.conn.execute("UPDATE messages SET acked = acked + 1 WHERE id = ?", (message_id,))
await db.conn.commit()
cursor = await db.conn.execute(
"SELECT acked FROM messages WHERE id = ?", (message_id,)
)
cursor = await db.conn.execute("SELECT acked FROM messages WHERE id = ?", (message_id,))
row = await cursor.fetchone()
return row["acked"] if row else 1
@@ -457,6 +463,7 @@ class RawPacketRepository:
(ts, data),
)
await db.conn.commit()
assert cursor.lastrowid is not None # INSERT always returns a row ID
return cursor.lastrowid
@staticmethod
app/routers/channels.py
+4 -7
View File
@@ -18,7 +18,7 @@ class CreateChannelRequest(BaseModel):
name: str = Field(min_length=1, max_length=32)
key: str | None = Field(
default=None,
description="Channel key as hex string (32 chars = 16 bytes). If omitted or name starts with #, key is derived from name hash."
description="Channel key as hex string (32 chars = 16 bytes). If omitted or name starts with #, key is derived from name hash.",
)
@@ -54,11 +54,10 @@ async def create_channel(request: CreateChannelRequest) -> Channel:
key_bytes = bytes.fromhex(request.key)
if len(key_bytes) != 16:
raise HTTPException(
status_code=400,
detail="Channel key must be exactly 16 bytes (32 hex chars)"
status_code=400, detail="Channel key must be exactly 16 bytes (32 hex chars)"
)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid hex string for key")
raise HTTPException(status_code=400, detail="Invalid hex string for key") from None
else:
# Derive key from name hash (same as meshcore library does)
key_bytes = sha256(request.name.encode("utf-8")).digest()[:16]
@@ -83,9 +82,7 @@ async def create_channel(request: CreateChannelRequest) -> Channel:
@router.post("/sync")
async def sync_channels_from_radio(
max_channels: int = Query(default=40, ge=1, le=40)
) -> dict:
async def sync_channels_from_radio(max_channels: int = Query(default=40, ge=1, le=40)) -> dict:
"""Sync channels from the radio to the database."""
mc = require_connected()
app/routers/contacts.py
+45 -65
View File
@@ -6,15 +6,20 @@ from meshcore import EventType
from app.dependencies import require_connected
from app.models import (
Contact,
TelemetryRequest,
TelemetryResponse,
NeighborInfo,
CONTACT_TYPE_REPEATER,
AclEntry,
CommandRequest,
CommandResponse,
CONTACT_TYPE_REPEATER,
Contact,
NeighborInfo,
TelemetryRequest,
TelemetryResponse,
)
from app.radio import radio_manager
from app.radio_sync import pause_polling
from app.repository import ContactRepository
logger = logging.getLogger(__name__)
# ACL permission level names
ACL_PERMISSION_NAMES = {
@@ -23,11 +28,6 @@ ACL_PERMISSION_NAMES = {
2: "Read-write",
3: "Admin",
}
from app.radio import radio_manager
from app.radio_sync import pause_polling
from app.repository import ContactRepository
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/contacts", tags=["contacts"])
# Delay between repeater radio operations to allow key exchange and path establishment
@@ -54,10 +54,7 @@ async def prepare_repeater_connection(mc, contact: Contact, password: str) -> No
login_result = await mc.commands.send_login(contact.public_key, password)
if login_result.type == EventType.ERROR:
raise HTTPException(
status_code=401,
detail=f"Login failed: {login_result.payload}"
)
raise HTTPException(status_code=401, detail=f"Login failed: {login_result.payload}")
# Wait for key exchange to complete before sending requests
logger.debug("Waiting %.1fs for key exchange to complete", REPEATER_OP_DELAY_SECONDS)
@@ -92,10 +89,7 @@ async def sync_contacts_from_radio() -> dict:
result = await mc.commands.get_contacts()
if result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to get contacts: {result.payload}"
)
raise HTTPException(status_code=500, detail=f"Failed to get contacts: {result.payload}")
contacts = result.payload
count = 0
@@ -131,10 +125,7 @@ async def remove_contact_from_radio(public_key: str) -> dict:
result = await mc.commands.remove_contact(radio_contact)
if result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to remove contact: {result.payload}"
)
raise HTTPException(status_code=500, detail=f"Failed to remove contact: {result.payload}")
await ContactRepository.set_on_radio(contact.public_key, False)
return {"status": "ok"}
@@ -159,10 +150,7 @@ async def add_contact_to_radio(public_key: str) -> dict:
result = await mc.commands.add_contact(contact.to_radio_dict())
if result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to add contact: {result.payload}"
)
raise HTTPException(status_code=500, detail=f"Failed to add contact: {result.payload}")
await ContactRepository.set_on_radio(contact.public_key, True)
return {"status": "ok"}
@@ -222,7 +210,7 @@ async def request_telemetry(public_key: str, request: TelemetryRequest) -> Telem
if contact.type != CONTACT_TYPE_REPEATER:
raise HTTPException(
status_code=400,
detail=f"Contact is not a repeater (type={contact.type}, expected {CONTACT_TYPE_REPEATER})"
detail=f"Contact is not a repeater (type={contact.type}, expected {CONTACT_TYPE_REPEATER})",
)
# Prepare connection (add/remove dance + login)
@@ -233,20 +221,13 @@ async def request_telemetry(public_key: str, request: TelemetryRequest) -> Telem
status = None
for attempt in range(1, 4):
logger.debug("Status request attempt %d/3", attempt)
status = await mc.commands.req_status_sync(
contact.public_key,
timeout=10.0,
min_timeout=5.0
)
status = await mc.commands.req_status_sync(contact.public_key, timeout=10, min_timeout=5)
if status:
break
logger.debug("Status request timeout, retrying...")
if not status:
raise HTTPException(
status_code=504,
detail="No response from repeater after 3 attempts"
)
raise HTTPException(status_code=504, detail="No response from repeater after 3 attempts")
logger.info("Received telemetry from %s: %s", contact.public_key[:12], status)
@@ -256,9 +237,7 @@ async def request_telemetry(public_key: str, request: TelemetryRequest) -> Telem
for attempt in range(1, 4):
logger.debug("Neighbors request attempt %d/3", attempt)
neighbors_data = await mc.commands.fetch_all_neighbours(
contact.public_key,
timeout=10.0,
min_timeout=5.0
contact.public_key, timeout=10, min_timeout=5
)
if neighbors_data:
break
@@ -272,23 +251,21 @@ async def request_telemetry(public_key: str, request: TelemetryRequest) -> Telem
pubkey_prefix = n.get("pubkey", "")
# Try to resolve to a contact name from our database
resolved_contact = await ContactRepository.get_by_key_prefix(pubkey_prefix)
neighbors.append(NeighborInfo(
pubkey_prefix=pubkey_prefix,
name=resolved_contact.name if resolved_contact else None,
snr=n.get("snr", 0.0),
last_heard_seconds=n.get("secs_ago", 0),
))
neighbors.append(
NeighborInfo(
pubkey_prefix=pubkey_prefix,
name=resolved_contact.name if resolved_contact else None,
snr=n.get("snr", 0.0),
last_heard_seconds=n.get("secs_ago", 0),
)
)
# Fetch ACL
logger.info("Fetching ACL from repeater %s", contact.public_key[:12])
acl_data = None
for attempt in range(1, 4):
logger.debug("ACL request attempt %d/3", attempt)
acl_data = await mc.commands.req_acl_sync(
contact.public_key,
timeout=10.0,
min_timeout=5.0
)
acl_data = await mc.commands.req_acl_sync(contact.public_key, timeout=10, min_timeout=5)
if acl_data:
break
logger.debug("ACL request timeout, retrying...")
@@ -302,12 +279,14 @@ async def request_telemetry(public_key: str, request: TelemetryRequest) -> Telem
perm = entry.get("perm", 0)
# Try to resolve to a contact name from our database
resolved_contact = await ContactRepository.get_by_key_prefix(pubkey_prefix)
acl_entries.append(AclEntry(
pubkey_prefix=pubkey_prefix,
name=resolved_contact.name if resolved_contact else None,
permission=perm,
permission_name=ACL_PERMISSION_NAMES.get(perm, f"Unknown({perm})"),
))
acl_entries.append(
AclEntry(
pubkey_prefix=pubkey_prefix,
name=resolved_contact.name if resolved_contact else None,
permission=perm,
permission_name=ACL_PERMISSION_NAMES.get(perm, f"Unknown({perm})"),
)
)
# Convert raw telemetry to response format
# bat is in mV, convert to V (e.g., 3775 -> 3.775)
@@ -364,7 +343,7 @@ async def send_repeater_command(public_key: str, request: CommandRequest) -> Com
if contact.type != CONTACT_TYPE_REPEATER:
raise HTTPException(
status_code=400,
detail=f"Contact is not a repeater (type={contact.type}, expected {CONTACT_TYPE_REPEATER})"
detail=f"Contact is not a repeater (type={contact.type}, expected {CONTACT_TYPE_REPEATER})",
)
# Pause message polling to prevent it from stealing our response
@@ -380,8 +359,7 @@ async def send_repeater_command(public_key: str, request: CommandRequest) -> Com
if send_result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to send command: {send_result.payload}"
status_code=500, detail=f"Failed to send command: {send_result.payload}"
)
# Wait for response (MESSAGES_WAITING event, then get_msg)
@@ -390,18 +368,21 @@ async def send_repeater_command(public_key: str, request: CommandRequest) -> Com
if wait_result is None:
# Timeout - no response received
logger.warning("No response from repeater %s for command: %s", contact.public_key[:12], request.command)
logger.warning(
"No response from repeater %s for command: %s",
contact.public_key[:12],
request.command,
)
return CommandResponse(
command=request.command,
response="(no response - command may have been processed)"
response="(no response - command may have been processed)",
)
response_event = await mc.commands.get_msg()
if response_event.type == EventType.ERROR:
return CommandResponse(
command=request.command,
response=f"(error: {response_event.payload})"
command=request.command, response=f"(error: {response_event.payload})"
)
# Extract the response text and timestamp from the payload
@@ -417,6 +398,5 @@ async def send_repeater_command(public_key: str, request: CommandRequest) -> Com
except Exception as e:
logger.error("Error waiting for response: %s", e)
return CommandResponse(
command=request.command,
response=f"(error waiting for response: {e})"
command=request.command, response=f"(error waiting for response: {e})"
)
app/routers/health.py
-1
View File
@@ -6,7 +6,6 @@ from pydantic import BaseModel
from app.config import settings
from app.radio import radio_manager
router = APIRouter(tags=["health"])
app/routers/messages.py
+23 -21
View File
@@ -5,8 +5,9 @@ from fastapi import APIRouter, HTTPException, Query
from meshcore import EventType
from app.dependencies import require_connected
from app.event_handlers import track_pending_ack, track_pending_repeat
from app.event_handlers import track_pending_ack
from app.models import Message, SendChannelMessageRequest, SendDirectMessageRequest
from app.packet_processor import track_pending_repeat
from app.repository import MessageRepository
logger = logging.getLogger(__name__)
@@ -18,7 +19,9 @@ async def list_messages(
limit: int = Query(default=100, ge=1, le=1000),
offset: int = Query(default=0, ge=0),
type: str | None = Query(default=None, description="Filter by type: PRIV or CHAN"),
conversation_key: str | None = Query(default=None, description="Filter by conversation key (channel key or contact pubkey)"),
conversation_key: str | None = Query(
default=None, description="Filter by conversation key (channel key or contact pubkey)"
),
) -> list[Message]:
"""List messages from the database."""
return await MessageRepository.get_all(
@@ -49,11 +52,11 @@ async def send_direct_message(request: SendDirectMessageRequest) -> Message:
# First check our database for the contact
from app.repository import ContactRepository
db_contact = await ContactRepository.get_by_key_or_prefix(request.destination)
if not db_contact:
raise HTTPException(
status_code=404,
detail=f"Contact not found in database: {request.destination}"
status_code=404, detail=f"Contact not found in database: {request.destination}"
)
# Check if contact is on radio, if not add it
@@ -80,10 +83,7 @@ async def send_direct_message(request: SendDirectMessageRequest) -> Message:
)
if result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to send message: {result.payload}"
)
raise HTTPException(status_code=500, detail=f"Failed to send message: {result.payload}")
# Store outgoing message
now = int(time.time())
@@ -95,13 +95,14 @@ async def send_direct_message(request: SendDirectMessageRequest) -> Message:
received_at=now,
outgoing=True,
)
assert message_id is not None # Outgoing messages are always new (unique timestamp)
# Update last_contacted for the contact
await ContactRepository.update_last_contacted(db_contact.public_key, now)
# Track the expected ACK for this message
expected_ack = result.payload.get("expected_ack")
suggested_timeout = result.payload.get("suggested_timeout", 10000) # default 10s
suggested_timeout: int = result.payload.get("suggested_timeout", 10000) # default 10s
if expected_ack:
ack_code = expected_ack.hex() if isinstance(expected_ack, bytes) else expected_ack
track_pending_ack(ack_code, message_id, suggested_timeout)
@@ -129,13 +130,13 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
mc = require_connected()
# Get channel info from our database
from app.repository import ChannelRepository
from app.decoder import calculate_channel_hash
from app.repository import ChannelRepository
db_channel = await ChannelRepository.get_by_key(request.channel_key)
if not db_channel:
raise HTTPException(
status_code=404,
detail=f"Channel {request.channel_key} not found in database"
status_code=404, detail=f"Channel {request.channel_key} not found in database"
)
# Convert channel key hex to bytes
@@ -143,14 +144,16 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
key_bytes = bytes.fromhex(request.channel_key)
except ValueError:
raise HTTPException(
status_code=400,
detail=f"Invalid channel key format: {request.channel_key}"
)
status_code=400, detail=f"Invalid channel key format: {request.channel_key}"
) from None
expected_hash = calculate_channel_hash(key_bytes)
logger.info(
"Sending to channel %s (%s) via radio slot %d, key hash: %s",
request.channel_key, db_channel.name, TEMP_RADIO_SLOT, expected_hash
request.channel_key,
db_channel.name,
TEMP_RADIO_SLOT,
expected_hash,
)
# Load the channel to a temporary radio slot before sending
@@ -162,7 +165,8 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
if set_result.type == EventType.ERROR:
logger.warning(
"Failed to set channel on radio slot %d before sending: %s",
TEMP_RADIO_SLOT, set_result.payload
TEMP_RADIO_SLOT,
set_result.payload,
)
# Continue anyway - the channel might already be correctly configured
@@ -174,10 +178,7 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
)
if result.type == EventType.ERROR:
raise HTTPException(
status_code=500,
detail=f"Failed to send message: {result.payload}"
)
raise HTTPException(status_code=500, detail=f"Failed to send message: {result.payload}")
# Store outgoing message
now = int(time.time())
@@ -190,6 +191,7 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
received_at=now,
outgoing=True,
)
assert message_id is not None # Outgoing messages are always new (unique timestamp)
# Track for repeat detection (flood messages get confirmed by hearing repeats)
track_pending_repeat(channel_key_upper, request.text, now, message_id)
app/routers/packets.py
+14 -15
View File
@@ -15,8 +15,12 @@ router = APIRouter(prefix="/packets", tags=["packets"])
class DecryptRequest(BaseModel):
key_type: str = Field(description="Type of key: 'channel' or 'contact'")
channel_key: str | None = Field(default=None, description="Channel key as hex (16 bytes = 32 chars)")
channel_name: str | None = Field(default=None, description="Channel name (for hashtag channels, key derived from name)")
channel_key: str | None = Field(
default=None, description="Channel key as hex (16 bytes = 32 chars)"
)
channel_name: str | None = Field(
default=None, description="Channel name (for hashtag channels, key derived from name)"
)
class DecryptResult(BaseModel):
@@ -45,9 +49,7 @@ async def _run_historical_decryption(channel_key_bytes: bytes, channel_key_hex:
processed = 0
decrypted_count = 0
_decrypt_progress = DecryptProgress(
total=total, processed=0, decrypted=0, in_progress=True
)
_decrypt_progress = DecryptProgress(total=total, processed=0, decrypted=0, in_progress=True)
logger.info("Starting historical decryption of %d packets", total)
@@ -84,9 +86,7 @@ async def _run_historical_decryption(channel_key_bytes: bytes, channel_key_hex:
total=total, processed=processed, decrypted=decrypted_count, in_progress=False
)
logger.info(
"Historical decryption complete: %d/%d packets decrypted", decrypted_count, total
)
logger.info("Historical decryption complete: %d/%d packets decrypted", decrypted_count, total)
@router.get("/undecrypted/count")
@@ -179,8 +179,7 @@ async def get_decrypt_progress() -> DecryptProgress | None:
class MaintenanceRequest(BaseModel):
prune_undecrypted_days: int = Field(
ge=1,
description="Delete undecrypted packets older than this many days"
ge=1, description="Delete undecrypted packets older than this many days"
)
@@ -197,7 +196,9 @@ async def run_maintenance(request: MaintenanceRequest) -> MaintenanceResult:
- Deletes undecrypted packets older than the specified number of days
- Runs VACUUM to reclaim disk space
"""
logger.info("Running maintenance: pruning packets older than %d days", request.prune_undecrypted_days)
logger.info(
"Running maintenance: pruning packets older than %d days", request.prune_undecrypted_days
)
# Prune old undecrypted packets
deleted = await RawPacketRepository.prune_old_undecrypted(request.prune_undecrypted_days)
@@ -264,14 +265,12 @@ async def _run_payload_dedup() -> None:
# Delete duplicates (keep the first/oldest packet in each group)
duplicates_removed = 0
for payload_hash, packet_ids in payload_groups.items():
for packet_ids in payload_groups.values():
if len(packet_ids) > 1:
# Keep the first one, delete the rest
ids_to_delete = packet_ids[1:]
for packet_id in ids_to_delete:
await db.conn.execute(
"DELETE FROM raw_packets WHERE id = ?", (packet_id,)
)
await db.conn.execute("DELETE FROM raw_packets WHERE id = ?", (packet_id,))
duplicates_removed += 1
_dedup_progress = DedupProgress(
app/routers/radio.py
+14 -6
View File
@@ -114,13 +114,15 @@ async def set_private_key(update: PrivateKeyUpdate) -> dict:
try:
key_bytes = bytes.fromhex(update.private_key)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid hex string for private key")
raise HTTPException(status_code=400, detail="Invalid hex string for private key") from None
logger.info("Importing private key")
result = await mc.commands.import_private_key(key_bytes)
if result.type == EventType.ERROR:
raise HTTPException(status_code=500, detail=f"Failed to import private key: {result.payload}")
raise HTTPException(
status_code=500, detail=f"Failed to import private key: {result.payload}"
)
return {"status": "ok"}
@@ -134,7 +136,9 @@ async def send_advertisement(flood: bool = True) -> dict:
result = await mc.commands.send_advert(flood=flood)
if result.type == EventType.ERROR:
raise HTTPException(status_code=500, detail=f"Failed to send advertisement: {result.payload}")
raise HTTPException(
status_code=500, detail=f"Failed to send advertisement: {result.payload}"
)
return {"status": "ok", "flood": flood}
@@ -164,7 +168,11 @@ async def reconnect_radio() -> dict:
return {"status": "ok", "message": "Already connected", "connected": True}
if radio_manager.is_reconnecting:
return {"status": "pending", "message": "Reconnection already in progress", "connected": False}
return {
"status": "pending",
"message": "Reconnection already in progress",
"connected": False,
}
logger.info("Manual reconnect requested")
success = await radio_manager.reconnect()
@@ -172,6 +180,7 @@ async def reconnect_radio() -> dict:
if success:
# Re-register event handlers after successful reconnect
from app.event_handlers import register_event_handlers
if radio_manager.meshcore:
register_event_handlers(radio_manager.meshcore)
# Restart auto message fetching
@@ -181,6 +190,5 @@ async def reconnect_radio() -> dict:
return {"status": "ok", "message": "Reconnected successfully", "connected": True}
else:
raise HTTPException(
status_code=503,
detail="Failed to reconnect. Check radio connection and power."
status_code=503, detail="Failed to reconnect. Check radio connection and power."
)
app/routers/read_state.py
+2 -8
View File
@@ -21,14 +21,8 @@ async def mark_all_read() -> dict:
now = int(time.time())
# Update all contacts and channels in one transaction
await db.conn.execute(
"UPDATE contacts SET last_read_at = ?",
(now,)
)
await db.conn.execute(
"UPDATE channels SET last_read_at = ?",
(now,)
)
await db.conn.execute("UPDATE contacts SET last_read_at = ?", (now,))
await db.conn.execute("UPDATE channels SET last_read_at = ?", (now,))
await db.conn.commit()
logger.info("Marked all contacts and channels as read at %d", now)
app/routers/settings.py
+10 -5
View File
@@ -10,7 +10,9 @@ router = APIRouter(prefix="/settings", tags=["settings"])
class AppSettingsResponse(BaseModel):
max_radio_contacts: int = Field(description="Maximum non-repeater contacts to keep on radio for DM ACKs")
max_radio_contacts: int = Field(
description="Maximum non-repeater contacts to keep on radio for DM ACKs"
)
class AppSettingsUpdate(BaseModel):
@@ -18,7 +20,7 @@ class AppSettingsUpdate(BaseModel):
default=None,
ge=1,
le=1000,
description="Maximum non-repeater contacts to keep on radio (1-1000)"
description="Maximum non-repeater contacts to keep on radio (1-1000)",
)
@@ -38,10 +40,13 @@ async def update_settings(update: AppSettingsUpdate) -> AppSettingsResponse:
Set MESHCORE_MAX_RADIO_CONTACTS environment variable for persistent changes.
"""
if update.max_radio_contacts is not None:
logger.info("Updating max_radio_contacts from %d to %d",
settings.max_radio_contacts, update.max_radio_contacts)
logger.info(
"Updating max_radio_contacts from %d to %d",
settings.max_radio_contacts,
update.max_radio_contacts,
)
# Pydantic settings are mutable, we can update them directly
object.__setattr__(settings, 'max_radio_contacts', update.max_radio_contacts)
object.__setattr__(settings, "max_radio_contacts", update.max_radio_contacts)
return AppSettingsResponse(
max_radio_contacts=settings.max_radio_contacts,
app/websocket.py
+11 -6
View File
@@ -96,9 +96,14 @@ def broadcast_health(radio_connected: bool, serial_port: str | None = None) -> N
except OSError:
pass
asyncio.create_task(ws_manager.broadcast("health", {
"status": "ok" if radio_connected else "degraded",
"radio_connected": radio_connected,
"serial_port": serial_port,
"database_size_mb": db_size_mb,
}))
asyncio.create_task(
ws_manager.broadcast(
"health",
{
"status": "ok" if radio_connected else "degraded",
"radio_connected": radio_connected,
"serial_port": serial_port,
"database_size_mb": db_size_mb,
},
)
)