Alec Perkins
0f478bf720
Adds the modjam package: a MeshCore-backed test station service for Pi (IDLE + RUNNING states, cuesheet-driven), a control station REPL for the Mac, and a UDP simulator that swaps in for the radio when SIMULATOR=true (drops cross-config packets and a configurable fraction of test-payload datagrams to mimic real radio loss). docker-compose runs three sim stations + control on a bridge net. TSV log format matches the reference traces. Pytest suite covers protocol codec, cuesheet builder, TSV logger, config loader, and UDP radio packet routing/loss; .forgejo/workflows runs it on push to main and on PRs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
45 lines
1 KiB
Python
45 lines
1 KiB
Python
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Awaitable, Callable
|
|
|
|
|
|
@dataclass
|
|
class RxPacket:
|
|
packet_id: int | str
|
|
text: str
|
|
sender: str | None = None
|
|
rssi: int | None = None
|
|
snr: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class SendResult:
|
|
packet_id: int | str
|
|
duration_ms: int
|
|
|
|
|
|
MessageHandler = Callable[[RxPacket], Awaitable[None] | None]
|
|
|
|
|
|
class Radio(ABC):
|
|
@abstractmethod
|
|
async def connect(self) -> None: ...
|
|
|
|
@abstractmethod
|
|
async def set_radio(self, freq_mhz: float, bw_khz: float,
|
|
sf: int, cr: int, tx_power_dbm: int) -> None: ...
|
|
|
|
@abstractmethod
|
|
async def ensure_channel(self, name: str, psk_b64: str) -> None: ...
|
|
|
|
@abstractmethod
|
|
async def send(self, text: str) -> SendResult: ...
|
|
|
|
@abstractmethod
|
|
def on_message(self, cb: MessageHandler) -> None: ...
|
|
|
|
@abstractmethod
|
|
async def get_noise_floor(self) -> int: ...
|
|
|
|
@abstractmethod
|
|
async def disconnect(self) -> None: ...
|