From d81f3c8618c621c8ad7abedc8c53b29167d94e69 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Fri, 16 Oct 2026 00:38:47 -0400 Subject: [PATCH] =?UTF-8?q?A=20demo=20mesh=20over=20Montr=C3=A9al,=20for?= =?UTF-8?q?=20screenshots=20that=20stay=20the=20same?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit meshterm --demo runs the simulator in a synthetic Montréal mesh: 14 repeaters on the high points (Mont-Royal to Mont-St-Hilaire), rooms, sensors, and about twenty companions, all invented. A log-distance link model decides who hears whom, so floods arrive on several real paths and traces walk real routes. At each start the demo writes its own directory (/demo) again: 300 days of adverts and a dense week of traffic, bilingual channel, direct, and room conversations whose packets decrypt, traces with their trophy-case records, neighbour tables, a joined room, admin passwords, regions, and Watchtower alerts. Every value comes from a fixed seed and an age before now, so each run gives the same data. The device plays the live traffic from the same seeds. The repository writers that stamp now take an optional time, and record_observations stores a batch in one commit. The test world of --mock is unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/cli/README.md | 2 + meshterm/cli.py | 32 +- meshterm/context.py | 5 +- meshterm/core/connection.py | 7 + meshterm/demo/__init__.py | 2 + meshterm/demo/device.py | 362 ++++++++++++++++++ meshterm/demo/history.py | 371 +++++++++++++++++++ meshterm/demo/home.py | 159 ++++++++ meshterm/demo/script.py | 387 ++++++++++++++++++++ meshterm/demo/traffic.py | 547 ++++++++++++++++++++++++++++ meshterm/demo/world.py | 567 +++++++++++++++++++++++++++++ meshterm/persistence/repository.py | 136 +++++-- meshterm/ui/menu.py | 14 +- tests/test_demo.py | 84 +++++ 14 files changed, 2631 insertions(+), 44 deletions(-) create mode 100644 meshterm/demo/__init__.py create mode 100644 meshterm/demo/device.py create mode 100644 meshterm/demo/history.py create mode 100644 meshterm/demo/home.py create mode 100644 meshterm/demo/script.py create mode 100644 meshterm/demo/traffic.py create mode 100644 meshterm/demo/world.py create mode 100644 tests/test_demo.py diff --git a/docs/cli/README.md b/docs/cli/README.md index 2a799cc..e25ca1f 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -57,6 +57,7 @@ If you give no command, MeshTerm starts the interactive menu. | `--tcp HOST[:PORT]` | The network address of a TCP companion. It selects the TCP transport. The default port is 5000. | | `--spi` | The LoRa radio on the SPI bus of this machine. It selects the SPI transport. MeshTerm runs the node for the length of the command. If one radio is attached, MeshTerm uses that radio, with a profile or without one. If no radio is attached, MeshTerm uses the one SPI profile, or else the wiring of the uConsole AIO. If two radios are attached, MeshTerm refuses (exit `3`) and lists them. Select one with `-p`. Refer to [Adding a radio on the SPI bus](../guide/configuration.md#adding-a-radio-on-the-spi-bus). Linux only. | | `--mock` | Use the built-in simulator instead of real hardware. Nothing transmits, and the run records to `meshterm-mock.db` instead of your real history. | +| `--demo` | Use the simulator in a demo mesh over Montréal: about forty invented nodes, a week of traffic, conversations, traces, and records. The demo has its own directory (`demo/` in the MeshTerm directory), and MeshTerm writes it again at each start. Thus each run starts from the same state, and your history never gets a node of the demo. It cannot take `--db`. | | `--db PATH` | Use this SQLite database instead of `~/.meshterm/meshterm.db` (or `meshterm-mock.db`, with `--mock`). **It moves the database and nothing else.** Refer to [Where MeshTerm keeps its state](#where-meshterm-keeps-its-state). | | `--json` | Print the answer as JSON instead of aligned text. | | `--absolute` | For this run, print times as ISO-8601 instants instead of relative ages. | @@ -99,6 +100,7 @@ radios from two directories in the same shell. | --- | --- | | `meshterm.db` | The history: each packet that MeshTerm heard, each trace that it walked, and each message. **This is the one file that `--db` moves.** | | `meshterm-mock.db` | The same history for the simulator. A `--mock` run records here when the run does not name a database. The file is absent until you do a `--mock` run. | +| `demo/` | The directory of the demo (`--demo`). MeshTerm writes everything in it again at each start of the demo, except the tile cache of the map and the log. | | `config.toml` | The setup of the machine: device profiles, and where the database is. You write this file. MeshTerm only reads it. | | `preferences.toml` | Your overrides of the behaviour of MeshTerm. It lists only what you changed. | | `courier.json` | The outbox: queued messages, both waiting and finished. | diff --git a/meshterm/cli.py b/meshterm/cli.py index bfcb78b..d16e9f8 100644 --- a/meshterm/cli.py +++ b/meshterm/cli.py @@ -23,11 +23,12 @@ from . import __version__ from .context import AppContext from .core import exitcodes, win32dll from .core.admin_store import AdminStore -from .core.config import MOCK_DB_FILENAME, Settings +from .core.config import DB_FILENAME, MOCK_DB_FILENAME, Settings from .core.connection import DeviceCommandError, is_connection_lost from .core.device_config import DeviceConfigError from .core.device_store import DeviceStore from .core.instancelock import InstanceBusy, hold_instance_lock +from .core.models import utcnow from .core.preferences import ( PREFERENCES_FILENAME, PreferenceError, @@ -189,6 +190,9 @@ def main_callback( help="Run a node on the LoRa radio on this machine's SPI bus (selects the SPI transport)", ), mock: bool = typer.Option(False, "--mock", help="Use the built-in simulator"), + demo: bool = typer.Option( + False, "--demo", help="Run the simulator in a demo mesh over Montréal, with its own data" + ), db_path: Path | None = typer.Option( None, "--db", help="SQLite database path (--mock records to meshterm-mock.db)" ), @@ -218,6 +222,8 @@ def main_callback( tcp: An explicit network address ``host[:port]``, which selects the TCP transport. spi: Select the radio on the SPI bus of this machine. MeshTerm itself runs the node. mock: Whether to use the simulator instead of real hardware. + demo: Run the simulator in the demo world (:mod:`meshterm.demo`), with a data + directory of its own that MeshTerm writes again at each start. db_path: Override the location of the database. If it is not set, a ``--mock`` run stores to the simulator's own database and not to the real history. json_output: Print the answer as JSON instead of aligned text. @@ -258,9 +264,30 @@ def main_callback( # contradict each other about the device to talk to are a bad command line. They are # not a question of precedence. raise typer.BadParameter(f"--mock and {named_radios[0]} name different devices; pass one") + if demo and named_radios: + raise typer.BadParameter(f"--demo and {named_radios[0]} name different devices; pass one") + if demo and db_path is not None: + # The demo writes its database again at each start. If it took --db, a typing error + # would put a demo history over the history of the user. + raise typer.BadParameter("--demo writes its own database; drop --db") settings = Settings.load() - if db_path is not None: + if demo: + # The demo is a whole MeshTerm of its own: its own directory, which it writes again + # at each start. The directory of the user never sees a node of the demo. The + # ``emulate`` command does not prepare it, because the MeshTerm in its window gets + # the same flag and prepares it there. + from .demo.home import demo_home, prepare + + mock = True + settings.config_dir = demo_home(settings.config_dir) + settings.db_path = settings.config_dir / DB_FILENAME + if ctx.invoked_subcommand != "emulate": + try: + prepare(settings.config_dir, utcnow()) + except InstanceBusy as exc: + raise typer.BadParameter(str(exc), param_hint="--demo") from None + elif db_path is not None: settings.db_path = db_path elif mock: # The simulator is a fake radio, but it is not a fake MeshTerm. MeshTerm stored all @@ -327,6 +354,7 @@ def main_callback( admin_store=AdminStore(settings.config_dir / "admin.json"), profile=settings.resolve_profile(profile), mock=mock, + demo=demo, port_override=port, ble_override=ble, tcp_override=tcp, diff --git a/meshterm/context.py b/meshterm/context.py index c775206..5ef05f1 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -102,6 +102,8 @@ class AppContext: room and the access that it granted (if the caller does not inject it, the default is ``/rooms.json``). mock: Whether the context uses the simulator device. + demo: Whether the simulator is the demo world of ``--demo`` + (:mod:`meshterm.demo`), which implies ``mock``. port_override: An explicit serial port (from ``--port`` or the interactive picker), which overrides the profile. ble_override: An explicit Bluetooth address (from ``--ble`` or the interactive @@ -145,6 +147,7 @@ class AppContext: room_store: RoomStore | None = None profile: DeviceProfile | None = None mock: bool = False + demo: bool = False port_override: str | None = None ble_override: str | None = None tcp_override: str | None = None @@ -615,7 +618,7 @@ class AppContext: return self._device if self.mock: - self._device = make_device(mock=True, port=None) + self._device = make_device(mock=True, port=None, demo=self.demo) self._active_transport = None self._active_port = None self._active_address = None diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 1dde0c6..278e13f 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -5998,6 +5998,7 @@ def make_device( tcp_port: int | None = None, spi: SpiWiring | None = None, state: Path | None = None, + demo: bool = False, ) -> Device: """Make the correct :class:`Device` for the current run. @@ -6023,6 +6024,8 @@ def make_device( state: The state directory of the SPI radio (refer to :func:`meshterm.core.spiradio.state_dir`). It is necessary when ``transport="spi"``. + demo: With ``mock``, return the simulator of the demo world + (:class:`~meshterm.demo.device.DemoDevice`) instead of the test world. Returns: A :class:`Device` instance that connects when you enter it. @@ -6031,6 +6034,10 @@ def make_device( ValueError: If a real device is requested without an endpoint that its transport can use. """ + if mock and demo: + from ..demo.device import DemoDevice # deferred: the demo builds on this module + + return DemoDevice() if mock: return MockDevice(optimal_tx=mock_optimal_tx) if transport == "ble": diff --git a/meshterm/demo/__init__.py b/meshterm/demo/__init__.py new file mode 100644 index 0000000..c915dcc --- /dev/null +++ b/meshterm/demo/__init__.py @@ -0,0 +1,2 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The demo: a synthetic Montréal mesh that the ``--demo`` flag runs MeshTerm against.""" diff --git a/meshterm/demo/device.py b/meshterm/demo/device.py new file mode 100644 index 0000000..19d1a84 --- /dev/null +++ b/meshterm/demo/device.py @@ -0,0 +1,362 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The device of the demo: the companion of our node in the Montréal world. + +:class:`DemoDevice` is the simulator (:class:`~meshterm.core.connection.MockDevice`) with +the demo world in place of its small test world. It holds the contacts, the channels, and +the settings of a companion that has lived on the mesh for months. It answers a trace +from the link model, a repeater from the region tables and the neighbours of the world, +and it plays the live traffic of the world while the demo runs: the adverts, the packets +of the background, and the lines of the script that arrive after the start +(:data:`~meshterm.demo.script.LIVE_DIRECT`, :data:`~meshterm.demo.script.LIVE_CHANNEL`). + +Everything that the device sends comes from a fixed seed and a time after the start. Thus +two runs play the same traffic at the same moments. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import random +from datetime import timedelta + +from ..core import transmit_gate +from ..core.connection import EventCallback, MockDevice, Unsubscribe +from ..core.events import MeshEvent +from ..core.models import Contact, Hop, Message, NeighbourInfo, RoomAccess, TraceResult, utcnow +from ..core.region_sim import SimulatedRegionMap +from . import script, traffic, world +from .history import TX_POWER + +#: The length of one part of the live traffic (seconds). The device makes the traffic one +#: part at a time, each from its own seed, so a long session never waits for all of it. +_PART_S = 3600.0 + + +class DemoDevice(MockDevice): + """The simulator, in the demo world (refer to the module docstring).""" + + def __init__(self) -> None: + """Build the companion of our node, with the state that the demo starts from.""" + super().__init__( + seed=hash_seed("device"), + optimal_tx=TX_POWER, + admin_password=world.ADMIN_PASSWORD, + ) + now = utcnow() + self._tx_power = TX_POWER + self._contacts = [_contact(node) for node in world.NODES if node.added] + room = world.node(world.JOINED_ROOM) + # The room knows our node as a member. Its posts until now are in the history + # already, so a login brings only the posts that come after the start. + self._room_board = [] + self._room_access = {room.public_key: RoomAccess.MEMBER} + self._room_synced = {room.public_key: now} + self._neighbour_tables = {} + self._region_maps = { + node.node_id: SimulatedRegionMap(world.REGION_TABLES[node.name]) + for node in world.REPEATERS + } + self._info.update( + name=world.SELF.name, + public_key=world.SELF.public_key, + adv_lat=world.SELF.lat, + adv_lon=world.SELF.lon, + adv_loc_policy=1, + max_tx_power=TX_POWER, + # The preset of North America, which the mesh of Montréal uses. + radio_freq=910.525, + radio_bw=62.5, + radio_sf=7, + radio_cr=5, + ) + self._channels = { + channel.slot: { + "channel_idx": channel.slot, + "channel_name": channel.name, + "channel_secret": channel.secret, + } + for channel in world.CHANNELS + } + self._clock_offset = 0 + + #: The route that a flood login teaches the device: one hop, through Mont-Royal. + _LEARNED_ROUTE = (world.node("Mont-Royal").hash,) + + async def get_device_info(self) -> dict: # noqa: D102 - inherited docstring + return { + "model": "MeshCore Simulator", + "ver": "v1.15.0", + "fw_build": "demo", + "ble_pin": self._device_pin, + "repeat": self._client_repeat, + } + + async def get_battery(self) -> dict: # noqa: D102 - inherited docstring + return {"level": 3980, "used_kb": 412, "total_kb": 1024} + + async def get_stats(self) -> dict: # noqa: D102 - inherited docstring + return { + "battery_mv": 3980, + "uptime_secs": 273_640, # 3 d 4 h + "errors": 0, + "queue_len": 0, + "noise_floor": -112, + "last_rssi": -96, + "last_snr": 7.25, + "tx_air_secs": 318, + "rx_air_secs": 9_412, + "recv": 14_803, + "sent": 1_212, + "flood_tx": 388, + "direct_tx": 824, + "flood_rx": 11_590, + "direct_rx": 3_213, + "recv_errors": 41, + } + + # -- the remote nodes ----------------------------------------------------------- + + def _remote_config(self, node: Contact) -> dict[str, str]: + """The settings of a node of the world, as its CLI shows them.""" + cfg = super()._remote_config(node) + known = _node_of(node) + if known is not None and cfg.get("lat") == "0": + cfg.update( + lat=f"{known.lat:.4f}", + lon=f"{known.lon:.4f}", + freq="910.525", + sf="7", + **{"owner.info": "Mesh Montréal · #montreal", "advert.interval": "240"}, + ) + return cfg + + async def send_remote_command( # noqa: D102 - inherited docstring + self, node: Contact, command: str, *, timeout: float = 8.0 + ) -> str | None: + verb = command.strip().split()[0].lower() if command.strip() else "" + if verb == "neighbors" and self._mock_key(node) in self._admin_sessions: + transmit_gate.mark() + await asyncio.sleep(0) + table = self._neighbours_of(node) + return "\n".join(f"{n.node} {n.snr:+.1f}dB {_ago(n)}s" for n in table) or "none" + return await super().send_remote_command(node, command, timeout=timeout) + + async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]: # noqa: D102 + await asyncio.sleep(0) + if self._mock_key(node) not in self._admin_sessions: + return await super().fetch_neighbours(node) # the same refusal as the simulator + return self._neighbours_of(node) + + def _neighbours_of(self, node: Contact) -> list[NeighbourInfo]: + """The neighbour table of a repeater of the world: each node that it hears now.""" + repeater = _node_of(node) + if repeater is None or not repeater.is_repeater: + return [] + now = utcnow() + table = [] + for other in (world.SELF, *world.NODES): + snr = world.link_snr(other, repeater) + if snr is None or not (other is world.SELF or traffic.active(other, 0.0)): + continue + heard_at = now - timedelta(seconds=self._rng.uniform(30.0, 3600.0)) + table.append( + NeighbourInfo(other.public_key[:8], world.reading(snr, self._rng), heard_at) + ) + return table + + # -- traces ----------------------------------------------------------------------- + + async def run_trace( # noqa: D102 - inherited docstring + self, + target: str, + *, + path: str | None = None, + timeout: float | None = None, + ) -> TraceResult: + transmit_gate.mark() + await asyncio.sleep(0.4) # the time on the air, so that the progress is visible + relays = _relays(target, path) + width = len(path.split(",")[0].strip()) // 2 if path else 1 + if relays is None: + return TraceResult(target=target, success=False, tx_power=self._tx_power) + walked = world.walk(relays, self._rng) + if not walked.success: + return TraceResult( + target=target, success=False, tx_power=self._tx_power, path_hash_bytes=width + ) + hops = [ + Hop(index, relay.public_key[: 2 * width], snr) + for index, (relay, snr) in enumerate(zip(relays, walked.snrs, strict=False)) + ] + hops.append(Hop(len(relays), None, walked.snrs[-1])) + return TraceResult( + target=target, + success=True, + hops=hops, + round_trip_ms=walked.rtt_ms, + tx_power=self._tx_power, + path_hash_bytes=width, + ) + + # -- the live traffic --------------------------------------------------------------- + + async def subscribe_events( # noqa: D102 - inherited docstring + self, on_event: EventCallback + ) -> Unsubscribe: + # Each subscriber gets the traffic from the moment that it subscribes, in parts of + # one hour. A part comes from its own seed, so the same moment of each run has the + # same packet. + stop = asyncio.Event() + loop = asyncio.get_running_loop() + start = loop.time() + + async def play() -> None: + part = 0 + while not stop.is_set(): + for item in _live_part(part): + wait = item.offset_s - (loop.time() - start) + if wait > 0: + try: + await asyncio.wait_for(stop.wait(), wait) + except asyncio.TimeoutError: + pass # the moment of the next item came + if stop.is_set(): + return + _deliver(item, on_event) + part += 1 + + task = asyncio.create_task(play()) + self._bg_tasks.add(task) + task.add_done_callback(self._bg_tasks.discard) + self._listeners.append(on_event) + + def unsubscribe() -> None: + stop.set() + if on_event in self._listeners: + self._listeners.remove(on_event) + + return unsubscribe + + +def hash_seed(name: str) -> int: + """A fixed integer seed for one random source of the demo device.""" + return int.from_bytes(hashlib.sha256(f"meshterm-demo/{name}".encode()).digest()[:4], "big") + + +def _contact(node: world.Node) -> Contact: + """The contact that the demo device holds for ``node``. + + The contact has no time of last reception. The app takes it from the history instead, + where it is the time of the last advert that our node heard from the node (refer to + :class:`~meshterm.services.device_state.DeviceState`). A real device knows no more. + """ + return Contact( + name=node.name, + public_key=node.public_key, + key_prefix=node.node_id, + node_type=node.node_type, + lat=node.lat if node.shares_location else None, + lon=node.lon if node.shares_location else None, + route_hops=world.device_route(node), + ) + + +def _node_of(contact: Contact) -> world.Node | None: + """The node of the world that a contact names, by its key (or by its name).""" + key = (contact.public_key or contact.key_prefix or "").lower() + for node in (world.SELF, *world.NODES): + if key and node.public_key.startswith(key[:12]): + return node + return world.BY_NAME.get(contact.name) + + +def _relays(target: str, path: str | None) -> list[world.Node] | None: + """The relays of a trace: the forced path, or the route out to ``target`` and back. + + Returns: + The relays, or ``None`` when a hash of the path names no repeater of the world + (then no relay answers, and the trace times out). + """ + if path: + relays = [] + for token in (part.strip().lower() for part in path.split(",") if part.strip()): + relay = next((r for r in world.REPEATERS if r.public_key.startswith(token)), None) + if relay is None: + return None + relays.append(relay) + return relays + node = world.BY_NAME.get(target) + if node is None or not node.is_repeater: + return None + route = world.best_route(world.SELF, node) + if route is None: + return None + return [*route, node, *reversed(route)] + + +def _ago(entry: NeighbourInfo) -> int: + """The seconds since a repeater last heard a neighbour, as its CLI writes them.""" + if entry.heard_at is None: + return 0 + return max(0, int((utcnow() - entry.heard_at).total_seconds())) + + +def _live_part(part: int) -> list[traffic.Heard]: + """The live traffic of one part: the background, then the lines of the script in it.""" + start, end = part * _PART_S, (part + 1) * _PART_S + heard = traffic.traffic(start, end, f"live/{part}") + rng = random.Random(f"meshterm-demo/live-lines/{part}") + for offset, line in script.LIVE_DIRECT: + if start <= offset < end: + heard.extend(_live_direct(offset, line, rng)) + for offset, name, line in script.LIVE_CHANNEL: + if start <= offset < end: + heard.extend(_live_channel(offset, name, line, rng)) + heard.sort(key=lambda item: item.offset_s) + return heard + + +def _live_direct(offset: float, line: script.Line, rng: random.Random) -> list[traffic.Heard]: + """A direct message that arrives during the demo, and the packets that carried it.""" + sender = world.node(line.sender) + body = { + "payload_typename": "TEXT_MSG", + "dest_hash": world.SELF.hash, + "src_hash": sender.hash, + "cipher_mac": f"{rng.getrandbits(16):04x}", + } + copies = traffic.heard_flood(sender, offset, body, rng) + first = copies[0].offset_s if copies else offset + 1.0 + snr = copies[0].observation.snr if copies and copies[0].observation else None + message = Message(text=line.text, sender=sender.node_id, snr=snr) + return [*copies, traffic.Heard(first + 0.05, message=message)] + + +def _live_channel( + offset: float, name: str, line: script.Line, rng: random.Random +) -> list[traffic.Heard]: + """A channel message that arrives during the demo, and the packets that carried it.""" + channel = next(c for c in world.CHANNELS if c.name == name) + sender = world.node(line.sender) + wire = f"{sender.name}: {line.text}" + body = traffic.grp_txt_body(channel.secret, wire, utcnow() + timedelta(seconds=offset)) + if channel.scope: + body = traffic.scoped(body, channel.scope) + copies = traffic.heard_flood(sender, offset, body, rng, region=channel.scope) + first = copies[0].offset_s if copies else offset + 1.0 + snr = copies[0].observation.snr if copies and copies[0].observation else None + message = Message(text=wire, channel=channel.slot, is_channel=True, snr=snr) + return [*copies, traffic.Heard(first + 0.05, message=message)] + + +def _deliver(item: traffic.Heard, on_event: EventCallback) -> None: + """Send one live item to the subscriber, stamped with the time of now.""" + now = utcnow() + if item.observation is not None: + item.observation.observed_at = now + on_event(MeshEvent.observation_event(item.observation)) + if item.message is not None: + item.message.received_at = now + item.message.sender_timestamp = now + on_event(MeshEvent.message_event(item.message)) diff --git a/meshterm/demo/history.py b/meshterm/demo/history.py new file mode 100644 index 0000000..2357ffd --- /dev/null +++ b/meshterm/demo/history.py @@ -0,0 +1,371 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The history of the demo: what our node heard and did before the demo started. + +At each start, ``--demo`` writes a new database with this module. The database holds what +the companion of our node would have stored after months on the mesh of Montréal: + +* the adverts of each node since our node first heard it, and the traffic of the last + eight days, packet by packet, each copy on its own path (:mod:`meshterm.demo.traffic`), +* the conversations of :mod:`meshterm.demo.script`: the stored messages, and the packets + that carried them, encrypted as the firmware encrypts them, so that the paths of a + message can be found again, +* the traces that our node ran (:data:`~meshterm.demo.script.TRACES`), each try in the + link model, and the records that the walks set in the trophy case, +* the neighbour tables of the repeaters that our node administers. + +Each time comes from ``now`` and a fixed age, and each random value comes from a fixed +seed. Thus two runs give the same database, with only the times moved to the new now. +""" + +from __future__ import annotations + +import random +from collections import defaultdict +from datetime import datetime, timedelta + +from .. import __version__ +from ..core.channels import channel_identity +from ..core.models import ( + PATH_TRACE_TARGET, + ChatMessage, + Hop, + NeighbourInfo, + Observation, + TraceResult, +) +from ..persistence.repository import Repository +from ..services.records import CATEGORY_BY_ID, walk_from_trace, walk_scores +from . import script, traffic, world + +#: How many days of history the demo writes. The traffic of the last eight days is +#: complete. Each older day holds one advert from each node (refer to +#: :func:`~meshterm.demo.traffic.adverts`). +HISTORY_DAYS = 300 + +#: The transmit power of the demo device (dBm), stored with each trace. +TX_POWER = 22 + +#: The map viewport that the demo opens on: the island of Montréal, with the two shores. +MAP_VIEW = (45.5200, -73.6200, 11) + + +def seed(repo: Repository, now: datetime) -> None: + """Write the full history of the demo into ``repo``. + + Args: + repo: An empty database. + now: The moment that the history ends at: the start of the demo. + """ + rng = random.Random("meshterm-demo/history") + heard = traffic.traffic(-HISTORY_DAYS * 86400.0, 0.0, "history") + messages, carried = _conversations(now, rng) + heard.extend(carried) + heard.sort(key=lambda item: item.offset_s) + traffic.stamp(heard, now) + _store_observations(repo, now, [item.observation for item in heard if item.observation]) + # The transcripts are in the order of the row ids. Thus the oldest message goes first. + for message in sorted(messages, key=lambda m: m.created_at): + repo.record_chat_message(message) + _store_traces(repo, now, rng) + _store_neighbours(repo, now, rng) + repo.set_map_view(*MAP_VIEW) + + +def _store_observations(repo: Repository, now: datetime, observations: list[Observation]) -> None: + """Store the observations, in one ``monitor`` run for each day, as the app makes them. + + The app opens one run each time it starts to record. The demo writes one for each of + the dense days, and one for all the older days. + """ + by_day: dict[int, list[Observation]] = defaultdict(list) + for obs in observations: + age_days = (now - obs.observed_at).total_seconds() / 86400.0 + by_day[min(int(age_days), 8)].append(obs) + for _, rows in sorted(by_day.items(), reverse=True): + rows.sort(key=lambda obs: obs.observed_at) + run_id = repo.start_run("monitor", {"mode": "background"}, at=rows[0].observed_at) + count = repo.record_observations(run_id, rows) + repo.finish_run(run_id, "ok", {"observations": count}, at=rows[-1].observed_at) + + +def _first_copy(origin: world.Node, offset: float, body: dict, rng: random.Random, **kwargs): + """The copies of a flood that our node must have received, because the script says so. + + A flood can lose all its copies on weak links. A line of the script arrived by + definition. Thus the function runs the flood again until a copy arrives. + """ + for _ in range(12): + copies = traffic.heard_flood(origin, offset, body, rng, **kwargs) + if copies: + return copies + raise RuntimeError(f"no flood from {origin.name} reaches our node") + + +def _conversations( + now: datetime, rng: random.Random +) -> tuple[list[ChatMessage], list[traffic.Heard]]: + """The stored messages of the script, and the packets that carried them. + + Args: + now: The end of the history. + rng: The source of the delays, the losses, and the noise. + + Returns: + The messages, and the packets that our node heard of them. + """ + messages: list[ChatMessage] = [] + heard: list[traffic.Heard] = _greetings(rng) + for channel in world.CHANNELS: + identity = channel_identity(channel.name, channel.secret) + for line in script.CHANNEL_LINES.get(channel.name, ()): + offset = -line.age_min * 60.0 + sender = world.node(line.sender) + wire = f"{sender.name}: {line.text}" + when = now + timedelta(seconds=offset) + body = traffic.grp_txt_body(channel.secret, wire, when) + if channel.scope: + body = traffic.scoped(body, channel.scope) + if line.sender == script.ME: + heard.extend(traffic.heard_flood(sender, offset, body, rng, region=channel.scope)) + messages.append( + ChatMessage( + text=line.text, + outbound=True, + is_channel=True, + channel_id=identity, + channel_idx=channel.slot, + peer_name=channel.name, + scope=channel.scope or "*", + created_at=when, + ) + ) + continue + copies = _first_copy(sender, offset, body, rng, region=channel.scope) + heard.extend(copies) + messages.append( + ChatMessage( + text=wire, + is_channel=True, + channel_id=identity, + channel_idx=channel.slot, + snr=copies[0].observation.snr if copies[0].observation else None, + created_at=when, + ) + ) + for peer_name, lines in script.DIRECT_LINES.items(): + peer = world.node(peer_name) + for line in lines: + offset = -line.age_min * 60.0 + when = now + timedelta(seconds=offset) + mac = f"{rng.getrandbits(16):04x}" + if line.sender == script.ME: + body = { + "payload_typename": "TEXT_MSG", + "dest_hash": peer.hash, + "src_hash": world.SELF.hash, + "cipher_mac": mac, + } + heard.extend(_send(world.SELF, peer, offset, body, rng)) + if line.acked: + ack = {"payload_typename": "ACK", "ack_crc": f"{rng.getrandbits(32):08x}"} + heard.extend(_send(peer, world.SELF, offset + 2.5, ack, rng)) + messages.append( + ChatMessage( + text=line.text, + outbound=True, + peer=peer.node_id, + peer_name=peer.name, + acked=line.acked, + created_at=when, + ) + ) + continue + body = { + "payload_typename": "TEXT_MSG", + "dest_hash": world.SELF.hash, + "src_hash": peer.hash, + "cipher_mac": mac, + } + copies = _first_copy(peer, offset, body, rng) + heard.extend(copies) + snr = copies[0].observation.snr if copies[0].observation else None + messages.append( + ChatMessage(text=line.text, peer=peer.node_id, snr=snr, created_at=when) + ) + for room_name, lines in script.ROOM_LINES.items(): + room = world.node(room_name) + link = world.link_snr(room, world.SELF) or 0.0 + for line in lines: + when = now - timedelta(minutes=line.age_min) + if line.sender == script.ME: + messages.append( + ChatMessage( + text=line.text, + outbound=True, + peer=room.node_id, + peer_name=room.name, + acked=line.acked, + created_at=when, + ) + ) + continue + author = world.BY_NAME.get(line.sender) + messages.append( + ChatMessage( + text=line.text, + peer=room.node_id, + snr=world.reading(link, rng), + author=author.public_key[:8] if author is not None else line.sender, + created_at=when, + ) + ) + return messages, heard + + +def _greetings(rng: random.Random) -> list[traffic.Heard]: + """The adverts that the people of the script send when they come online to talk. + + A person who opens the app to write often sends an advert first. Thus a node that + spoke a few minutes ago was also heard a few minutes ago, and the lists show it as + recent. The function adds one advert before the first line of each sitting in the last + two days (a sitting is a run of lines less than an hour apart). + """ + lines = [ + line + for group in (script.CHANNEL_LINES, script.DIRECT_LINES) + for conversation in group.values() + for line in conversation + if line.sender != script.ME and line.age_min < 2 * 1440 + ] + last_advert: dict[str, float] = {} + heard: list[traffic.Heard] = [] + for line in sorted(lines, key=lambda line: -line.age_min): + offset = -line.age_min * 60.0 + if offset - last_advert.get(line.sender, -1e12) < 3600.0: + continue + sender = world.node(line.sender) + moment = offset - rng.uniform(60.0, 900.0) + heard.extend( + traffic.heard_flood(sender, moment, traffic.advert_body(sender), rng, advert=True) + ) + last_advert[line.sender] = offset + return heard + + +def _send( + src: world.Node, dst: world.Node, offset: float, body: dict, rng: random.Random +) -> list[traffic.Heard]: + """The packets that our node hears from a direct message (or its ack) from ``src`` to ``dst``. + + The message follows the route that the device holds when it holds one, else it floods. + """ + holder, target = (dst, src) if dst is world.SELF else (src, dst) + route = world.device_route(target) if holder is world.SELF else None + if route is None: + return traffic.heard_flood(src, offset, body, rng) + relays = tuple(world.BY_NAME[_name_of(h)] for h in route) + if src is not world.SELF: + relays = tuple(reversed(relays)) + return traffic.heard_direct(relays, src, offset, body, rng) + + +def _name_of(hash_hex: str) -> str: + """The name of the repeater with this path hash.""" + return next(node.name for node in world.REPEATERS if node.hash == hash_hex) + + +def _store_traces(repo: Repository, now: datetime, rng: random.Random) -> None: + """Run each trace of the script in the link model, store each try, and score the walks. + + A trace that does not come back is tried again a minute later, as a person does, up to + five times. Each try is a run of its own, as in the trace screen. A walk that comes + back goes to the trophy case, through the same scoring as a live trace. + """ + canonical = {node.hash: node.node_id for node in world.NODES} + positions = {node.node_id: (node.lat, node.lon) for node in world.NODES if node.shares_location} + self_pos = (world.SELF.lat, world.SELF.lon) + for age_min, target, names in script.TRACES: + relays = [world.node(name) for name in names] + spec = ",".join(relay.hash for relay in relays) + label = target or PATH_TRACE_TARGET + for attempt in range(5): + when = now - timedelta(minutes=age_min) + timedelta(seconds=75 * attempt) + walked = world.walk(relays, rng) + # The last reading of a walk that came back is the link to our node: the hop + # without a hash that the firmware adds at the end. + hops = [ + Hop(i, relay.hash, snr) + for i, (relay, snr) in enumerate(zip(relays, walked.snrs, strict=False)) + ] + if walked.success: + hops.append(Hop(len(relays), None, walked.snrs[-1])) + result = TraceResult( + target=label, + success=walked.success, + hops=hops if walked.success else [], + round_trip_ms=walked.rtt_ms, + tx_power=TX_POWER, + path_hash_bytes=1, + timestamp=when, + ) + params = {"target": label, "path": spec} + run_id = repo.start_run("trace", params, at=when - timedelta(seconds=2)) + repo.record_trace(run_id, result) + placed: list[str] = [] + derived = walk_from_trace( + result, canonical=canonical.get, positions=positions, self_pos=self_pos + ) + if derived is not None: + walk_spec, route, stats = derived + for category_id, score in walk_scores(stats).items(): + category = CATEGORY_BY_ID[category_id] + row = repo.record_discovery( + category_id, + 1, + walk_spec, + route, + score=score, + stats=stats.as_dict(), + app_version=__version__, + ascending=category.ascending, + at=when, + ) + if row is not None: + placed.append(f"{category.title} — {category.format_score(score)}") + summary = { + "target": label, + "success": result.success, + "min_snr": result.min_snr, + "rtt_ms": result.round_trip_ms, + "records": placed or None, + } + repo.finish_run(run_id, "ok", summary, at=when) + if walked.success: + break + + +def _store_neighbours(repo: Repository, now: datetime, rng: random.Random) -> None: + """Store two fetches of the neighbour table of each repeater that our node administers. + + A repeater lists each node that it hears directly, with the SNR that it measured, the + same as the simulated device answers a live fetch. + """ + for name in world.ADMINISTERED: + repeater = world.node(name) + for age_h in (30.0, 2.0): + fetched = now - timedelta(hours=age_h) + offset = -age_h * 3600.0 + entries = [] + for other in (world.SELF, *world.NODES): + snr = world.link_snr(other, repeater) + if snr is None or not (other is world.SELF or traffic.active(other, offset)): + continue + heard_at = fetched - timedelta(seconds=rng.uniform(60.0, 4 * 3600.0)) + entries.append( + NeighbourInfo(other.public_key[:8], world.reading(snr, rng), heard_at) + ) + params = {"mode": "neighbours", "repeater": repeater.name} + run_id = repo.start_run("trace", params, at=fetched - timedelta(seconds=3)) + repo.record_neighbours(run_id, repeater.node_id, entries, at=fetched) + summary = {"repeater": repeater.name, "neighbours": len(entries)} + repo.finish_run(run_id, "ok", summary, at=fetched) diff --git a/meshterm/demo/home.py b/meshterm/demo/home.py new file mode 100644 index 0000000..3ad295e --- /dev/null +++ b/meshterm/demo/home.py @@ -0,0 +1,159 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The home of the demo: its own data directory, written again at each start. + +``--demo`` never uses the data directory of the user. It uses ``/demo``, and +before the app opens, :func:`prepare` writes everything in it again from the demo world: +the database (:mod:`meshterm.demo.history`) and the stores of the config directory (the +joined room, the admin passwords, the regions, the Watchtower). Thus each run starts from +the same state, whatever the previous run changed, and the history of the user never +gets a node of the demo. + +Two things in the directory stay from one run to the next: the tile cache of the map, +because the tiles are the same and a download is slow, and the log. +""" + +from __future__ import annotations + +import sqlite3 +from datetime import datetime, timedelta +from pathlib import Path + +from ..core.admin_store import AdminStore +from ..core.channels import channel_identity +from ..core.config import DB_FILENAME +from ..core.instancelock import hold_instance_lock +from ..core.models import Contact, LoginResult, RoomAccess, RoomLogin +from ..core.region_sim import SimulatedRegionMap +from ..core.region_store import RegionStore +from ..core.regions import WILDCARD, parse_region_list +from ..core.room_store import RoomStore +from ..core.watch_store import WatchStore +from ..persistence.repository import Repository +from . import history, script, world + +#: The name of the demo directory, under the config directory of the user. +DEMO_DIRNAME = "demo" + +#: The files that the demo writes again at each start. A store that the demo does not +#: fill is deleted, so that nothing that a previous run changed stays. +_STORES = ( + "preferences.toml", + "contacts.json", + "channels.json", + "rooms.json", + "admin.json", + "regions.json", + "watchtower.json", + "courier.json", + "mutes.json", + "remote.json", + "settings.json", + "adverts.json", + "devices.json", +) + + +def demo_home(config_dir: Path) -> Path: + """The data directory of the demo, under the config directory of the user.""" + return config_dir / DEMO_DIRNAME + + +def prepare(home: Path, now: datetime) -> None: + """Write the demo directory again: a new history that ends at ``now``, and new stores. + + The function holds the instance lock of the directory while it writes. Thus it never + writes under a demo that runs in another window. + + Args: + home: The demo directory (refer to :func:`demo_home`). + now: The moment that the history ends at. + + Raises: + InstanceBusy: Another MeshTerm runs the demo now. + """ + with hold_instance_lock(home): + for name in _STORES: + (home / name).unlink(missing_ok=True) + db_path = home / DB_FILENAME + _empty(db_path) + repo = Repository(db_path) + try: + history.seed(repo, now) + finally: + repo.close() + _write_stores(home, now) + + +def _empty(db_path: Path) -> None: + """Delete each row of an existing database, and keep the file. + + The function does not delete the file, because another process can have it open: the + ``emulate`` command opens the database of the demo, then starts the MeshTerm that runs + in its window, which prepares the demo. Windows does not let a process delete a file + that is open. + """ + if not db_path.exists(): + return + conn = sqlite3.connect(db_path) + try: + tables = [ + row[0] + for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'") + if row[0] != "sqlite_sequence" and not row[0].startswith("sqlite_") + ] + conn.execute("PRAGMA foreign_keys = OFF") + for table in tables: + conn.execute(f'DELETE FROM "{table}"') + if conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'sqlite_sequence'").fetchone(): + conn.execute("DELETE FROM sqlite_sequence") + conn.commit() + finally: + conn.close() + + +def contact(node: world.Node) -> Contact: + """The contact that the demo device holds for ``node``, as a store keys it.""" + return Contact( + name=node.name, + public_key=node.public_key, + key_prefix=node.node_id, + node_type=node.node_type, + ) + + +def _write_stores(home: Path, now: datetime) -> None: + """Write the stores of the config directory that the demo fills. + + Each store writes its own file through its own methods, so that the files have the + form that the app reads. + + Args: + home: The demo directory. + now: The start of the demo, from which the times of the alerts come. + """ + room = world.node(world.JOINED_ROOM) + RoomStore(home / "rooms.json").record( + contact(room), + world.ROOM_PASSWORD, + RoomLogin(LoginResult.ACCEPTED, RoomAccess.MEMBER, flood=False), + ) + admins = AdminStore(home / "admin.json") + for name in world.ADMINISTERED: + admins.remember(contact(world.node(name)), world.ADMIN_PASSWORD) + regions = RegionStore(home / "regions.json") + for repeater in world.REPEATERS: + listed = SimulatedRegionMap(world.REGION_TABLES.get(repeater.name)).allowed_names() + regions.learn_carried(repeater.node_id, [WILDCARD, *parse_region_list(listed)]) + for channel in world.CHANNELS: + if channel.scope: + regions.set_channel_scope(channel_identity(channel.name, channel.secret), channel.scope) + watch = WatchStore(home / "watchtower.json") + for name in script.WATCHED: + node = world.node(name) + watch.watch(node.node_id, node.name, node_type=node.node_type) + for age_min, kind, name, message in script.ALERTS: + watch.add_alert(kind, name, message, when=now - timedelta(minutes=age_min)) + for node in world.NODES: + if node.quiet_hours is not None: + watch.remember_known(node.node_id) + watch.flush() diff --git a/meshterm/demo/script.py b/meshterm/demo/script.py new file mode 100644 index 0000000..98da41a --- /dev/null +++ b/meshterm/demo/script.py @@ -0,0 +1,387 @@ +# SPDX-License-Identifier: Apache-2.0 +"""What the people of the demo world say: the channels, the direct messages, and the room. + +The conversations are in French and in English, as on the real mesh of Montréal. Each line +has a time on a virtual clock (refer to :func:`at`). The virtual clock sets "now" at +:data:`NOW_LOCAL`, an evening, when the mesh is busy. The history writes each line at its +age before the real time of the run. Thus the ages on the screens ("12m ago") are the +same at each run. When the clock of the run is also set to that evening (the screenshots +do this), the times of day are correct too: the dinner is at dinner time. + +The script only holds text and times. :mod:`meshterm.demo.history` turns each line into +the stored message and into the packets that our node heard when the line went over the +air. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .world import SELF + +#: The local time of day that "now" stands for on the virtual clock (hours). +NOW_LOCAL = 19.5 + + +def at(days_ago: int, clock: str) -> float: + """The age (minutes before now) of a moment on the virtual clock. + + Args: + days_ago: 0 for today, 1 for yesterday, and so on. + clock: The local time of day, as ``"HH:MM"``. + + Returns: + The minutes from that moment to now. A time later than :data:`NOW_LOCAL` on day + 0 is a bug in the script, because it is in the future. + """ + hours, minutes = (int(part) for part in clock.split(":")) + age = days_ago * 1440.0 + NOW_LOCAL * 60.0 - (hours * 60.0 + minutes) + if age < 0: + raise ValueError(f"{clock} today is after now on the virtual clock") + return age + + +@dataclass(frozen=True, slots=True) +class Line: + """One line of a conversation. + + Attributes: + age_min: How many minutes before now the line was sent. + sender: The name of the node that sent it: a node of the world, :data:`ME` for our + node, or, in a room, the room itself (a notice of its admin) or an 8-hex key + prefix (an author that no contact names). + text: The body. + acked: For our own direct message or room post: whether the ack came back. + """ + + age_min: float + sender: str + text: str + acked: bool = True + + +#: The sender name of our own lines. +ME = SELF.name + +#: The channel conversations, by channel name, oldest line first. +CHANNEL_LINES: dict[str, tuple[Line, ...]] = { + "Public": ( + Line( + at(6, "20:12"), + "Félix", + "Nouveau répéteur à Longueuil, sur le toit de mon immeuble. Qui m'entend?", + ), + Line( + at(6, "20:15"), "Poutine Patrol", "Reçu fort et clair depuis le Vieux-Longueuil, +9 dB" + ), + Line( + at(6, "20:21"), + "Ana Brossard", + "Heard it from Brossard too, two hops through Mont-Royal", + ), + Line(at(6, "20:24"), "Félix", "Merci! Je le laisse en test cette semaine"), + Line(at(5, "21:40"), "Kim-NDG", "Anyone else getting duplicates through Oratoire tonight?"), + Line( + at(5, "21:46"), + "Jean-Seb", + "C'est normal, Oratoire et Mont-Royal s'entendent très bien. Rien de cassé.", + ), + Line(at(5, "21:48"), "Kim-NDG", "ok good, I thought my T-Deck was haunted 👻"), + Line( + at(4, "07:55"), + "Sam Verdun", + "Verdun-Quai est hors ligne, batterie à plat. Je monte sur le toit demain matin", + ), + Line(at(3, "08:31"), "Sam Verdun", "Verdun-Quai est de retour, batterie neuve 🔋"), + Line(at(3, "08:40"), ME, "Merci Sam! Je le reçois via Mont-Royal à +3 dB"), + Line( + at(2, "21:05"), + "Dave-Lachine", + "Lachine path is solid tonight: Lachine → Mont-Royal → Stade, worst link -1.5 dB", + ), + Line( + at(2, "21:19"), + "Bixi Bandit", + "Riding up Berri with a node in my bag, pinging every 2 min. Zero drops so far 🚲", + ), + Line(at(1, "12:10"), "Chloé L.", "Les couleurs sont superbes sur le mont Royal 🍁"), + Line( + at(1, "12:25"), + "Jean-Seb", + "Et le répéteur du chalet entend tout le monde, même à travers les feuilles 😄", + ), + Line(at(0, "13:20"), "Noor", "Premier test depuis Saint-Laurent, quelqu'un me reçoit?"), + Line(at(0, "13:24"), "Marie-Ève", "@[Noor] reçu via Mont-Royal, +6 dB. Bienvenue!"), + Line( + at(0, "17:55"), + "Mathilde", + "Bonsoir la gang! Première fois sur MeshCore, je suis à Saint-Henri 👋", + ), + Line(at(0, "17:58"), "Marie-Ève", "Bienvenue @[Mathilde]! Tu passes par quel répéteur?"), + Line(at(0, "18:02"), "Mathilde", "Mont-Royal je pense, ou Oratoire. Comment je vérifie?"), + Line( + at(0, "18:05"), + ME, + "Ton advert m'arrive en un hop via Mont-Royal, SNR +8. Trace path te montre tout", + ), + Line( + at(0, "18:30"), "Jean-Seb", "Rappel: rencontre mesh samedi 14h au marché Jean-Talon ☕" + ), + Line( + at(0, "18:49"), + "Noor", + "@[Ruelle Verte] t'as encore la carte de couverture? On veut la voir samedi", + ), + Line( + at(0, "18:52"), + ME, + "Oui! https://www.openstreetmap.org/#map=11/45.5200/-73.6400 je l'apporte imprimée", + ), + Line(at(0, "19:18"), "Poutine Patrol", "J'apporte des poutines pour tout le monde 🍟"), + Line( + at(0, "19:27"), "Félix", "Longueuil ↔ Stade à +6 dB depuis la nouvelle antenne, enfin!" + ), + ), + "#montreal": ( + Line( + at(6, "18:40"), + "Raphaël", + "Pont Viau fermé ce soir pour des travaux, passez par Papineau", + ), + Line( + at(4, "16:10"), + "Olivier.T", + "Anyone know if the Jean-Talon market is open late on Thursday?", + ), + Line(at(4, "16:21"), "Marie-Ève", "Jusqu'à 18h je crois, 20h le vendredi"), + Line( + at(3, "20:05"), + "Jean-Seb", + "Panne de courant à Ahuntsic. Les nœuds sur batterie tiennent le coup 💪", + ), + Line( + at(3, "20:13"), "Félix", "Même chose à Hochelaga, mais Stade-Olympique n'a rien senti" + ), + Line(at(2, "11:40"), "Kim-NDG", "Smoked meat run at noon, who's in?"), + Line(at(2, "11:44"), "Poutine Patrol", "Toujours."), + Line(at(2, "11:47"), "Sam Verdun", "Moi! On se rejoint au métro?"), + Line( + at(1, "09:15"), + "Chloé L.", + "Neige annoncée la semaine prochaine. Rentrez vos antennes de balcon ⛄", + ), + Line( + at(0, "11:10"), + ME, + "Quelqu'un a vu la grue sur Saint-Denis? Elle bloque mon lien vers Stade 😑", + ), + Line(at(0, "11:18"), "Félix", "Haha, la grue a un meilleur SNR que nous autres"), + Line( + at(0, "17:10"), + "Raphaël", + "La ligne orange roule au ralenti entre Henri-Bourassa et Montmorency", + ), + Line(at(0, "19:05"), "Olivier.T", "Bixi stations are all full downtown, good luck tonight"), + ), + "#velo": ( + Line( + at(5, "10:20"), + "Bixi Bandit", + "Tour du canal de Lachine dimanche 9h, départ au marché Atwater", + ), + Line( + at(5, "10:31"), "Dave-Lachine", "Count me in. I'll bring the T-Deck for a coverage test" + ), + Line(at(5, "10:52"), "Sam Verdun", "Je serai là avec le vélo cargo 🚲"), + Line( + at(3, "13:30"), + "Bixi Bandit", + "Résultat du tour: 11 répéteurs entendus entre Atwater et Lachine", + ), + Line( + at(3, "13:42"), "Dave-Lachine", "And the Lachine repeater heard all of us. Great test!" + ), + Line( + at(1, "16:20"), + "Sam Verdun", + "La piste du REV Saint-Denis est encore en travaux près de Jean-Talon", + ), + Line( + at(0, "18:20"), "Bixi Bandit", "Quelqu'un pour une montée du mont Royal demain matin?" + ), + ), + "Famille": ( + Line(at(4, "19:02"), "Léa", "Souper chez grand-maman dimanche, 17h"), + Line(at(4, "19:10"), ME, "Parfait, j'apporte le dessert"), + Line(at(4, "19:12"), "Léa", "N'oublie pas la tourtière de tante Jo!"), + Line(at(0, "17:41"), "Léa", "T'es où? Le souper est prêt 😄"), + Line(at(0, "17:44"), ME, "J'arrive, je suis au coin de la rue"), + ), +} + +#: The direct conversations, by the name of the other node, oldest line first. +DIRECT_LINES: dict[str, tuple[Line, ...]] = { + "Jean-Seb": ( + Line( + at(7, "09:12"), + "Jean-Seb", + "Peux-tu redémarrer Stade-Olympique? Il ne répond plus aux logins", + ), + Line(at(7, "09:31"), ME, "C'est fait, il est revenu"), + ), + "Bixi Bandit": ( + Line(at(2, "17:02"), "Bixi Bandit", "Le lien vers Mont-St-Hilaire tient la route?"), + Line(at(2, "17:06"), ME, "Oui, -5 dB depuis Mont-Royal. Fragile, mais ça passe"), + ), + "Dave-Lachine": ( + Line( + at(2, "20:20"), + ME, + "Can you trace me from Lachine tonight? Testing the path through Oratoire", + ), + Line(at(2, "20:51"), "Dave-Lachine", "Done: 3 hops, worst link -2.5 dB at Oratoire"), + Line(at(2, "20:55"), ME, "Merci Dave!", acked=False), + ), + "Marie-Ève": ( + Line(at(1, "18:20"), "Marie-Ève", "Tu viens à la rencontre samedi?"), + Line(at(1, "18:31"), ME, "Oui! J'apporte deux nœuds de rechange pour les nouveaux"), + Line(at(1, "18:40"), "Marie-Ève", "Super. Je réserve une table au café du marché"), + Line(at(1, "18:44"), ME, "Parfait, à samedi 👍"), + Line( + at(0, "18:40"), "Marie-Ève", "Mathilde est nouvelle, tu peux l'aider avec son Heltec?" + ), + Line(at(0, "18:43"), ME, "Bien sûr, je lui écris"), + ), + "Mathilde": ( + Line( + at(0, "18:45"), + ME, + "Salut! Marie-Ève m'a dit que tu commences. Besoin d'aide pour le Heltec?", + ), + Line(at(0, "19:00"), "Mathilde", "Oui merci! Je ne sais pas quelle fréquence choisir"), + Line(at(0, "19:02"), ME, "910.525 MHz, BW 62.5, SF7. C'est le réglage de Montréal"), + Line(at(0, "19:10"), "Mathilde", "Ça marche! Je vois trois répéteurs maintenant 🎉"), + ), +} + +#: The board of each room, oldest post first. The room name as the sender is a notice of +#: the room admin. ``5e1d9c07`` is a member that no contact of our node names. +ROOM_LINES: dict[str, tuple[Line, ...]] = { + "Le Babillard": ( + Line( + at(2, "19:30"), + "Jean-Seb", + "Rencontre mesh samedi 14h au café du marché Jean-Talon. Apportez vos nœuds!", + ), + Line(at(2, "19:50"), "Marie-Ève", "J'y serai, avec des antennes à donner"), + Line( + at(1, "22:50"), + "Le Babillard", + "Maintenance de Mont-Royal mardi de 22h à 23h. Prévoyez des détours.", + ), + Line( + at(1, "12:50"), + "Noor", + "Perdu: un Heltec V3 dans un boîtier jaune, parc La Fontaine. Récompense: une poutine", + ), + Line(at(1, "13:00"), "Poutine Patrol", "J'offre la poutine si tu le retrouves 😄"), + Line(at(0, "07:50"), "5e1d9c07", "Quelqu'un a un connecteur SMA à prêter?"), + Line(at(0, "08:40"), "Félix", "Moi, passe à Hochelaga quand tu veux"), + Line(at(0, "08:50"), ME, "J'apporte le testeur d'antennes samedi"), + Line( + at(0, "16:30"), "Kim-NDG", "Slides from the last meetup are on the board, ask Jean-Seb" + ), + Line(at(0, "19:15"), "Jean-Seb", "Il reste six places pour samedi"), + ), +} + +#: The lines that arrive while the demo runs, as ``(seconds after the start, line)``. The +#: first one is a direct message, so the header shows one unread message soon after the +#: start. The age of each line is unused. +LIVE_DIRECT: tuple[tuple[float, Line], ...] = ( + (6.0, Line(0.0, "Bixi Bandit", "T'as vu le record? 34 km jusqu'à Mont-St-Hilaire 🔥")), +) +LIVE_CHANNEL: tuple[tuple[float, str, Line], ...] = ( + ( + 45.0, + "Public", + Line(0.0, "Chloé L.", "Quelqu'un monte au mont Royal demain pour un test de portée?"), + ), + (160.0, "#montreal", Line(0.0, "Kim-NDG", "Le 80 est encore en retard sur du Parc 🙄")), +) + +#: The traces that our node ran in the past days, as ``(age in minutes, target, path)``. +#: The target is a repeater name for a target trace, or ``None`` for a composed path. A +#: path names the relays, without our node at the two ends. The history runs each one in +#: the link model, tries again a few times when the trace does not come back (as a person +#: does), and stores each try. +TRACES: tuple[tuple[float, str | None, tuple[str, ...]], ...] = ( + (at(6, "21:30"), "Longueuil", ("Mont-Royal", "Longueuil", "Mont-Royal")), + (at(5, "20:10"), "Pointe-Claire", ("Mont-Royal", "Pointe-Claire", "Mont-Royal")), + ( + at(5, "20:20"), + None, + ("Mont-Royal", "Pointe-Claire", "Île-Bizard", "Pointe-Claire", "Mont-Royal"), + ), + (at(4, "21:00"), "Laval-Chomedey", ("Ahuntsic", "Laval-Chomedey", "Ahuntsic")), + (at(3, "22:15"), None, ("Mont-Royal", "Mont-St-Hilaire", "Mont-Royal")), + (at(2, "20:40"), None, ("Oratoire", "Lachine", "Oratoire")), + ( + at(2, "21:50"), + None, + ("Mont-Royal", "Mont-St-Bruno", "Mont-St-Hilaire", "Mont-St-Bruno", "Mont-Royal"), + ), + (at(1, "20:30"), None, ("Mont-Royal", "Laval-Chomedey", "Ahuntsic", "Stade-Olympique")), + ( + at(1, "21:10"), + None, + ( + "Centre-Ville", + "Verdun-Quai", + "Brossard", + "Mont-St-Bruno", + "Mont-St-Hilaire", + "Mont-St-Bruno", + "Longueuil", + "Stade-Olympique", + "Ahuntsic", + "Laval-Chomedey", + "Mont-Royal", + "Oratoire", + "Lachine", + "Pointe-Claire", + "Mont-Royal", + ), + ), + ( + at(1, "21:40"), + None, + ( + "Oratoire", + "Lachine", + "Pointe-Claire", + "Mont-Royal", + "Brossard", + "Mont-St-Bruno", + "Mont-St-Hilaire", + "Stade-Olympique", + ), + ), + (at(0, "16:05"), "Brossard", ("Mont-Royal", "Brossard", "Mont-Royal")), + (at(0, "18:12"), None, ("Mont-Royal", "Longueuil", "Stade-Olympique")), +) + +#: The Watchtower: the nodes that our node watches, and the alerts that it raised and the +#: user did not acknowledge yet, as ``(age in minutes, kind, node name, message)``. The +#: messages are the words of :mod:`meshterm.services.watchtower`. +WATCHED: tuple[str, ...] = ("Mont-Royal", "Mont-St-Hilaire", "Verdun-Quai", "Longueuil") +ALERTS: tuple[tuple[float, str, str, str], ...] = ( + (at(4, "07:40"), "silence", "Verdun-Quai", "nothing heard for 6 h"), + ( + at(3, "08:31"), + "recovered", + "Verdun-Quai", + "back on the air — heard again after a silence alarm", + ), + (at(0, "17:55"), "new-node", "Mathilde", "first appearance — never heard on this mesh before"), +) diff --git a/meshterm/demo/traffic.py b/meshterm/demo/traffic.py new file mode 100644 index 0000000..f41d9ae --- /dev/null +++ b/meshterm/demo/traffic.py @@ -0,0 +1,547 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The packets of the demo world: what our node hears from the mesh, packet by packet. + +Two parts of the demo use this module. The history (:mod:`meshterm.demo.history`) asks it +for the packets of the past days, and stores them. The simulated device +(:mod:`meshterm.demo.device`) asks it for the packets of the next hour, and sends each one +as a live event at its time. Thus the live feed continues the stored history without a +seam. + +Each transmission on the mesh goes through the link model of :mod:`meshterm.demo.world`. +A flood reaches our node on each path that the race of the relays gives +(:func:`~meshterm.demo.world.flood`), so one transmission is often heard several times, +on several paths, as on a real mesh. A direct packet follows the route of its sender, and +our node hears it only where a relay of that route is in its range. Each packet that our +node hears becomes one ``packet`` observation, with the fields that the meshcore library +gives for an overheard packet (refer to +:func:`~meshterm.core.connection.packet_observation_from_event`). An advert also becomes +one ``advert`` observation, as the companion reports it. + +The rate of the traffic follows the hour of the virtual clock (:data:`RHYTHM`): the mesh is +quiet at night and busy in the evening. +""" + +from __future__ import annotations + +import hashlib +import random +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import datetime, timedelta + +from ..core.channels import channel_hash +from ..core.models import ( + NODE_TYPE_REPEATER, + NODE_TYPE_ROOM, + NODE_TYPE_SENSOR, + Message, + Observation, +) +from ..core.regions import region_key, scope_body, transport_code +from .script import NOW_LOCAL +from .world import ( + FOREIGN_CHANNELS, + NODES, + REGION_TABLES, + REPEATERS, + SELF, + Copy, + Node, + best_route, + flood, + link_snr, + loss_chance, + reading, +) + +#: The relative rate of the traffic for each local hour of the day, from midnight. +RHYTHM: tuple[float, ...] = ( + 0.35, 0.2, 0.15, 0.12, 0.12, 0.18, 0.35, 0.6, 0.75, 0.7, 0.65, 0.7, + 0.8, 0.7, 0.65, 0.7, 0.8, 0.95, 1.0, 1.0, 1.0, 0.9, 0.7, 0.5, +) # fmt: skip + +#: The number of transmissions that start on the mesh in each minute at the busiest hour. +#: The adverts and the lines of the script come on top of this rate. +PEAK_PER_MIN = 1.1 + +#: The payload classes of the background traffic, with their relative weights. A flood +#: is heard several times. Thus the counts of the stored packets have other proportions: +#: the channel texts and the requests lead, as on the real mesh of Montréal. +_KINDS: tuple[tuple[str, float], ...] = ( + ("GRP_TXT", 18.0), + ("REQ", 16.0), + ("RESPONSE", 10.0), + ("TEXT_MSG", 9.0), + ("ACK", 5.0), + ("ANON_REQ", 4.0), + ("PATH", 3.0), + ("TRACE", 2.0), + ("GRP_DATA", 0.3), + ("CONTROL", 0.3), + ("MULTIPART", 0.05), +) + +#: The 4-bit payload type of each class in the packet header (MeshCore's ``Packet.h``). The +#: transport code of a scoped flood comes from it. +_PAYLOAD_TYPE = {"GRP_TXT": 5, "ADVERT": 4} + +#: How often each type of node sends a flood advert (hours), and how often a repeater +#: sends a local advert that only its neighbours hear (hours). +_ADVERT_EVERY_H = {"repeater": 12.0, "room": 12.0, "sensor": 6.0, "companion": 9.0} +_LOCAL_ADVERT_EVERY_H = 4.0 + + +def rhythm(local_hour: float) -> float: + """The relative rate of the traffic at a local hour of the virtual clock.""" + return RHYTHM[int(local_hour) % 24] + + +def local_hour(offset_s: float) -> float: + """The local hour on the virtual clock, ``offset_s`` seconds after now (negative: before).""" + return (NOW_LOCAL + offset_s / 3600.0) % 24.0 + + +def _rssi(snr: float, rng: random.Random) -> float: + """A plausible RSSI (dBm) for a reception with this SNR, over a noise floor near -110 dBm.""" + return float(round(-108.0 + 1.1 * snr + rng.gauss(0.0, 1.5))) + + +def active(node: Node, offset_s: float) -> bool: + """Whether ``node`` transmits at this moment (``offset_s`` seconds after now). + + A node transmits after our node first heard it, and before it went quiet. A node that + is active now stays active after now, so the live traffic continues the history. A + contact that our node never heard never transmits. + """ + if node.quiet_hours is None: + return False + first = -node.first_heard_days * 86400.0 + if node.quiet_hours <= 0: + return first <= offset_s + return first <= offset_s <= -node.quiet_hours * 3600.0 + + +def _kind_of(node: Node) -> str: + """The advert schedule that a node follows (refer to :data:`_ADVERT_EVERY_H`).""" + return { + NODE_TYPE_REPEATER: "repeater", + NODE_TYPE_ROOM: "room", + NODE_TYPE_SENSOR: "sensor", + }.get(node.node_type, "companion") + + +def _phase(node: Node, salt: str) -> float: + """A fixed fraction in ``[0, 1)`` for a node: where its periodic schedule starts.""" + return hashlib.sha256(f"{salt}/{node.name}".encode()).digest()[0] / 256.0 + + +# --------------------------------------------------------------------------- packets + + +@dataclass(frozen=True, slots=True) +class Heard: + """One event that our node hears: an observation or a received message. + + Attributes: + offset_s: When it is heard, in seconds after now (negative: in the past). + observation: The observation, or ``None``. + message: The received message, or ``None``. + """ + + offset_s: float + observation: Observation | None = None + message: Message | None = None + + +def _mac(rng: random.Random) -> str: + """A random 2-byte MAC (hex): the tag of an encrypted body that our node cannot read.""" + return f"{rng.getrandbits(16):04x}" + + +def grp_txt_body(secret: bytes, text: str, sent_at: datetime, attempt: int = 0) -> dict: + """The fields of a GRP_TXT packet for ``text`` on the channel with ``secret``. + + The body is what the firmware sends: the timestamp of the sender and the attempt byte, + then the text, padded to whole AES blocks and encrypted in ECB mode with the key of the + channel. The MAC is the first two bytes of the HMAC-SHA256 of the ciphertext. Thus + MeshTerm decrypts a stored copy exactly as it decrypts a real one (refer to + :func:`~meshterm.core.channels.decrypt_channel_text`). + + Args: + secret: The 16-byte key of the channel. + text: The text of the message, with the ``Name: `` prefix of the sender. + sent_at: The time on the clock of the sender. + attempt: The resend counter of the sender. + + Returns: + The ``payload_typename``, ``chan_hash``, ``cipher_mac``, and ``crypted`` fields. + """ + from Crypto.Cipher import AES + from Crypto.Hash import HMAC, SHA256 + + plain = int(sent_at.timestamp()).to_bytes(4, "little") + bytes([attempt & 0x03]) + plain += text.encode("utf-8") + plain += b"\x00" * (-len(plain) % 16) + crypted = AES.new(secret, AES.MODE_ECB).encrypt(plain) + mac = HMAC.new(secret, digestmod=SHA256) + mac.update(crypted) + return { + "payload_typename": "GRP_TXT", + "chan_hash": channel_hash(secret), + "cipher_mac": mac.digest()[:2].hex(), + "crypted": crypted.hex(), + } + + +def scoped(body: dict, region: str) -> dict: + """``body`` as a scoped flood: the route type, the transport code, and the HMAC input. + + Args: + body: The fields of a GRP_TXT packet (refer to :func:`grp_txt_body`). + region: The bare region name. + + Returns: + A copy of ``body`` with ``route_typename``, ``transport_code``, and ``scope_body``. + """ + payload = ( + bytes.fromhex(body["chan_hash"]) + + bytes.fromhex(body["cipher_mac"]) + + bytes.fromhex(body["crypted"]) + ) + hmac_input = scope_body(_PAYLOAD_TYPE["GRP_TXT"], payload) + code = transport_code(region_key(region), hmac_input) + return { + **body, + "route_typename": "TC_FLOOD", + "transport_code": code.to_bytes(2, "little").hex() + "0000", + "scope_body": hmac_input.hex(), + } + + +def _relays_region(relay: Node, region: str | None) -> bool: + """Whether ``relay`` relays a flood in ``region`` (``None``: an unscoped flood).""" + if region is None: + return True + return any(name == region and allowed for name, _, allowed in REGION_TABLES.get(relay.name, ())) + + +def heard_flood( + origin: Node, + start_s: float, + body: dict, + rng: random.Random, + *, + region: str | None = None, + advert: bool = False, +) -> list[Heard]: + """The packets that our node hears from one flood. + + Args: + origin: The node that transmits the flood. + start_s: When it transmits (seconds after now). + body: The payload fields of the packet. + rng: The source of the delays, the losses, and the noise. + region: The scope of the flood, or ``None`` for an unscoped flood. Only the + repeaters that carry the region relay it. + advert: Whether the packet is an advert. Then the companion also reports one + ``advert`` observation, with the SNR of the first copy. + + Returns: + One :class:`Heard` for each copy, and one more for the advert report. + """ + copies = [ + copy + for copy in flood(origin, rng) + if all(_relays_region(relay, region) for relay in copy.relays) + ] + route = "TC_FLOOD" if region else "FLOOD" + out = [ + _packet(origin if advert else None, start_s, copy, {"route_typename": route, **body}, rng) + for copy in copies + ] + if advert and copies: + first = copies[0] + out.append(Heard(start_s + first.delay_s, observation=_advert_report(origin, first, rng))) + return out + + +def _packet( + origin: Node | None, start_s: float, copy: Copy, body: dict, rng: random.Random +) -> Heard: + """One ``packet`` observation for one copy of a transmission.""" + path = ",".join(relay.hash for relay in copy.relays) + raw = {**body, "path_len": len(copy.relays), "path_hash_size": 1} + share = origin is not None and origin.shares_location + return Heard( + start_s + copy.delay_s, + observation=Observation( + node=origin.node_id if origin is not None else None, + public_key=origin.public_key if origin is not None else None, + name=origin.name if origin is not None else None, + kind="packet", + node_type=origin.node_type if origin is not None else None, + snr=copy.snr, + rssi=_rssi(copy.snr, rng), + lat=origin.lat if share else None, + lon=origin.lon if share else None, + path=path, + raw=raw, + ), + ) + + +def _advert_report(origin: Node, copy: Copy, rng: random.Random) -> Observation: + """The ``advert`` observation that the companion reports for an advert that it heard.""" + return Observation( + node=origin.node_id, + public_key=origin.public_key, + name=origin.name, + kind="advert", + node_type=origin.node_type, + snr=copy.snr, + rssi=_rssi(copy.snr, rng), + lat=origin.lat if origin.shares_location else None, + lon=origin.lon if origin.shares_location else None, + ) + + +def advert_body(origin: Node) -> dict: + """The decoded fields of an advert packet from ``origin``.""" + body: dict = { + "payload_typename": "ADVERT", + "adv_key": origin.public_key, + "adv_name": origin.name, + "adv_type": origin.node_type, + } + if origin.shares_location: + body.update(adv_lat=origin.lat, adv_lon=origin.lon) + return body + + +def heard_direct( + route: tuple[Node, ...], src: Node, start_s: float, body: dict, rng: random.Random +) -> list[Heard]: + """The packets that our node hears from one direct packet along ``route``. + + The sender writes the route into the packet, and each relay removes itself before it + transmits. Thus our node hears the sender with the full route, and each relay with + the part of the route that is still in front of the packet. + + Args: + route: The relays between the sender and the recipient. + src: The sender. + start_s: When the sender transmits (seconds after now). + body: The payload fields of the packet. + rng: The source of the delays, the losses, and the noise. + + Returns: + One :class:`Heard` for each transmission that our node hears. + """ + out: list[Heard] = [] + elapsed = 0.0 + for index, transmitter in enumerate((src, *route)): + elapsed += rng.uniform(0.2, 0.9) + if transmitter.name == SELF.name: + continue + snr = link_snr(transmitter, SELF) + if snr is None or rng.random() < loss_chance(snr): + continue + remaining = route[index:] + copy = Copy(remaining, reading(snr, rng), elapsed) + out.append(_packet(None, start_s, copy, {"route_typename": "DIRECT", **body}, rng)) + return out + + +# --------------------------------------------------------------------------- traffic + + +def _pick(nodes: list[Node], rng: random.Random) -> Node: + """A node, with the weight of its :attr:`~meshterm.demo.world.Node.talk`.""" + return rng.choices(nodes, weights=[node.talk for node in nodes])[0] + + +def adverts(start_s: float, end_s: float, rng: random.Random) -> Iterator[Heard]: + """The adverts that our node hears between two moments (seconds after now). + + Each node sends a flood advert on a fixed period, with a fixed phase and some jitter. + A repeater also sends local adverts, which only its neighbours hear. More than eight + days before now, the history keeps only one advert of each node each day: enough for + the dates of first and last reception, and not a full week of packets for each day of + the year. + """ + dense_from = -8 * 86400.0 + for origin in (*NODES, SELF): + kind = _kind_of(origin) + period = _ADVERT_EVERY_H[kind] * 3600.0 + moment = start_s - (start_s % period) + _phase(origin, "advert") * period + while moment < end_s: + sparse = moment < dense_from + when = moment + rng.uniform(-0.1, 0.1) * period + step = 86400.0 if sparse else period + if start_s <= when < end_s and (origin is SELF or active(origin, when)): + if origin is SELF: + # Our node hears its own advert only where the repeaters relay it. The + # history keeps these packets only for the recent days. + if not sparse: + yield from heard_flood(origin, when, advert_body(origin), rng) + elif sparse: + copies = flood(origin, rng) + if copies: + yield Heard( + when + copies[0].delay_s, + observation=_advert_report(origin, copies[0], rng), + ) + else: + yield from heard_flood(origin, when, advert_body(origin), rng, advert=True) + moment += step + if kind != "repeater" or link_snr(origin, SELF) is None: + continue + period = _LOCAL_ADVERT_EVERY_H * 3600.0 + moment = max(start_s, dense_from) + moment = moment - (moment % period) + _phase(origin, "local") * period + while moment < end_s: + if start_s <= moment and active(origin, moment): + snr = link_snr(origin, SELF) + if snr is not None and rng.random() >= loss_chance(snr): + copy = Copy((), reading(snr, rng), 0.3) + body = {"route_typename": "FLOOD", **advert_body(origin)} + yield _packet(origin, moment, copy, body, rng) + yield Heard(moment + 0.3, observation=_advert_report(origin, copy, rng)) + moment += period + + +def background(start_s: float, end_s: float, rng: random.Random) -> Iterator[Heard]: + """The background traffic of the mesh between two moments (seconds after now). + + The transmissions arrive as a Poisson process, with the rate of the hour + (:func:`rhythm`). Each one has a class (:data:`_KINDS`), a sender, and a recipient + where its class has one. Then it becomes the packets that our node hears. + """ + kinds = [kind for kind, _ in _KINDS] + weights = [weight for _, weight in _KINDS] + moment = start_s + while True: + rate_s = PEAK_PER_MIN * rhythm(local_hour(moment)) / 60.0 + moment += rng.expovariate(max(rate_s, 1e-6)) + if moment >= end_s: + return + awake = [node for node in NODES if active(node, moment)] + people = [node for node in awake if node.node_type == SELF.node_type] + servers = [node for node in awake if node.node_type != SELF.node_type] + if not people or not servers: + continue + kind = rng.choices(kinds, weights=weights)[0] + yield from _transmission(kind, moment, people, servers, rng) + + +def _transmission( + kind: str, moment: float, people: list[Node], servers: list[Node], rng: random.Random +) -> Iterator[Heard]: + """The packets that our node hears from one background transmission of class ``kind``.""" + sender = _pick(people, rng) + if kind in ("GRP_TXT", "GRP_DATA"): + secret = rng.choice(FOREIGN_CHANNELS) + body = { + "payload_typename": kind, + "chan_hash": channel_hash(secret), + "cipher_mac": _mac(rng), + "crypted": rng.randbytes(16 * rng.randint(2, 6)).hex(), + } + yield from heard_flood(sender, moment, body, rng) + return + if kind == "CONTROL": + # A discovery request: one hop, heard only near its sender. + snr = link_snr(sender, SELF) + if snr is not None: + copy = Copy((), reading(snr, rng), 0.2) + yield _packet( + None, moment, copy, {"route_typename": "DIRECT", "payload_typename": kind}, rng + ) + return + if kind == "MULTIPART": + yield from heard_flood(sender, moment, {"payload_typename": kind}, rng) + return + if kind == "TRACE": + yield from _trace(moment, rng) + return + if kind in ("REQ", "ANON_REQ"): + target = _pick(servers, rng) + else: + target = _pick([node for node in people if node is not sender] or people, rng) + if kind == "RESPONSE": + sender, target = _pick(servers, rng), sender + body: dict = {"payload_typename": kind, "dest_hash": target.hash, "cipher_mac": _mac(rng)} + if kind == "ANON_REQ": + body["src_key"] = sender.public_key + elif kind != "ACK": + body["src_hash"] = sender.hash + if kind == "ACK": + body = {"payload_typename": kind, "ack_crc": f"{rng.getrandbits(32):08x}"} + route = best_route(sender, target) + if route is not None and kind in ("ACK", "RESPONSE", "TEXT_MSG", "REQ") and rng.random() < 0.55: + yield from heard_direct(route, sender, moment, body, rng) + else: + yield from heard_flood(sender, moment, body, rng) + + +def _trace(moment: float, rng: random.Random) -> Iterator[Heard]: + """A trace that another node runs, as our node hears it pass. + + The trace walks a few repeaters, out and back. Our node hears each relay of the walk + that is in its range, with the SNR readings that the packet carries at that point. + """ + start = rng.choice(REPEATERS) + out = [start] + for _ in range(rng.randint(1, 3)): + options = [r for r in REPEATERS if r not in out and link_snr(out[-1], r) is not None] + if not options: + break + out.append(rng.choice(options)) + walk = [*out, *reversed(out[:-1])] + snrs: list[float] = [] + elapsed = 0.0 + tag = f"{rng.getrandbits(32):08x}" + for a, b in zip(walk, walk[1:], strict=False): + snr = link_snr(a, b) + if snr is None or rng.random() < loss_chance(snr): + return + snrs.append(reading(snr, rng)) + elapsed += rng.uniform(0.2, 0.6) + heard = link_snr(b, SELF) + if heard is not None and rng.random() >= loss_chance(heard): + body = { + "route_typename": "DIRECT", + "payload_typename": "TRACE", + "trace_tag": tag, + "trace_snrs": list(snrs), + } + yield _packet(None, moment, Copy((), reading(heard, rng), elapsed), body, rng) + + +def traffic(start_s: float, end_s: float, seed: str) -> list[Heard]: + """All the traffic that our node hears between two moments, in time order. + + Args: + start_s: The start, in seconds after now (negative: in the past). + end_s: The end, in seconds after now. + seed: The seed of the random source. The same seed gives the same traffic. + + Returns: + The adverts and the background packets, sorted by time. + """ + rng = random.Random(f"meshterm-demo/{seed}") + heard = [*adverts(start_s, end_s, rng)] + dense_from = max(start_s, -8 * 86400.0) + if dense_from < end_s: + heard.extend(background(dense_from, end_s, rng)) + heard = [item for item in heard if start_s <= item.offset_s < end_s] + heard.sort(key=lambda item: item.offset_s) + return heard + + +def stamp(heard: list[Heard], now: datetime) -> list[Heard]: + """Give each observation in ``heard`` its time, from ``now`` and its offset.""" + for item in heard: + if item.observation is not None: + item.observation.observed_at = now + timedelta(seconds=item.offset_s) + return heard diff --git a/meshterm/demo/world.py b/meshterm/demo/world.py new file mode 100644 index 0000000..ccb06fb --- /dev/null +++ b/meshterm/demo/world.py @@ -0,0 +1,567 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The demo world: a synthetic MeshCore mesh over Montréal. + +The ``--demo`` flag runs MeshTerm against this world. Two parts of the demo read it. The +simulated device (:mod:`meshterm.demo.device`) answers as the companion of our node. The +history (:mod:`meshterm.demo.history`) fills the database with what that companion heard +in the days before. Thus each screen shows one consistent mesh: a node on the map is a +node that the contacts, the chat, and the routes also know, and a route that a trace +walks is a route that the floods also use. + +The world is fixed data and pure functions. Nothing in this module reads the clock, and +each random value comes from a generator with a fixed seed. Thus the world is the same at +each run, and the screenshots of the website show the same mesh each time that somebody +makes them again. + +All the names are invented, and no node is a real node. The places are real: the +repeaters stand on the high points of the Montréal area, and the links between them come +from a simple model of the radio path (refer to :func:`link_snr`). +""" + +from __future__ import annotations + +import functools +import hashlib +import heapq +import math +import random +from collections.abc import Sequence +from dataclasses import dataclass, field +from itertools import pairwise + +from ..core.channels import DEFAULT_PUBLIC_SECRET, derive_secret +from ..core.geo import haversine_km +from ..core.models import ( + NODE_TYPE_CHAT, + NODE_TYPE_REPEATER, + NODE_TYPE_ROOM, + NODE_TYPE_SENSOR, +) + +# --------------------------------------------------------------------------- the nodes + + +@dataclass(frozen=True, slots=True, eq=False) +class Node: + """One node of the demo world. + + Each node exists once, in :data:`NODES` or as :data:`SELF`. Thus two nodes are equal + only when they are the same object, and the hash of a node is cheap: the link model + keys its caches on nodes millions of times. + + Attributes: + name: The name that the node advertises. + hue: The first byte of its public key. This byte is the hash of the node in a + path, and it selects the hue of its name on each screen. The world gives + each node its own byte, so that no two nodes of the world have the same hash, + and the hues of the repeaters stay far apart. + node_type: The advert type (``NODE_TYPE_*``). + lat: The latitude of the node. The link model always uses it. + lon: The longitude of the node. + site_db: The gain of the site, in dB, against a node at street level: a mountain + top is worth more than a roof, and a roof more than a pocket. + shares_location: Whether the adverts of the node carry its position. A node that + does not share it still has one, because the link model needs it. + added: Whether the node is in the contact list of our device. A node that is not + added is a discovered node: MeshTerm only heard it. + first_heard_days: How many days before now our node first heard this node. + quiet_hours: How many hours before now the node transmitted for the last time. + Zero is a node that is active now. ``None`` is a contact that our node never + heard (a contact added from a shared card). + talk: How often the node starts a transmission, relative to an ordinary node. + public_key: The full public key (64 hex digits): the hue byte, then a fixed digest + of the name. + """ + + name: str + hue: int + node_type: int + lat: float + lon: float + site_db: float + shares_location: bool = True + added: bool = True + first_heard_days: float = 120.0 + quiet_hours: float | None = 0.0 + talk: float = 1.0 + public_key: str = field(init=False, repr=False) + + def __post_init__(self) -> None: + """Derive the public key from the hue byte and the name.""" + tail = hashlib.sha256(f"meshterm-demo/{self.name}".encode()).digest()[1:] + object.__setattr__(self, "public_key", (bytes([self.hue]) + tail).hex()) + + @property + def node_id(self) -> str: + """The canonical id of the node in the history: the first 12 hex digits of its key.""" + return self.public_key[:12] + + @property + def hash(self) -> str: + """The one-byte path hash of the node (2 hex digits).""" + return self.public_key[:2] + + @property + def is_repeater(self) -> bool: + """Whether the node relays floods.""" + return self.node_type == NODE_TYPE_REPEATER + + +_R, _C, _ROOM, _S = NODE_TYPE_REPEATER, NODE_TYPE_CHAT, NODE_TYPE_ROOM, NODE_TYPE_SENSOR + +#: Our node: the companion that the demo device is. It stands in La Petite-Patrie, near the +#: Jean-Talon market, with an antenna on the roof of a triplex. +SELF = Node("Ruelle Verte", 0x91, _C, 45.5350, -73.6030, 2.0, first_heard_days=400.0) + +# fmt: off +#: The other nodes of the world. The repeaters stand on the high points. The companions are +#: the people of the mesh. The order is the order of the contact list on the device. +NODES: tuple[Node, ...] = ( + # The repeaters, on the mountains, the towers, and the roofs. + Node("Mont-Royal", 0x2A, _R, 45.5060, -73.5880, 13.0, first_heard_days=380.0), + Node("Oratoire", 0x4F, _R, 45.4924, -73.6178, 8.0, first_heard_days=300.0), + Node("Centre-Ville", 0xBC, _R, 45.5015, -73.5690, 8.0, first_heard_days=350.0), + Node("Stade-Olympique", 0x98, _R, 45.5580, -73.5517, 10.0, first_heard_days=360.0), + Node("Ahuntsic", 0x05, _R, 45.5560, -73.6600, 6.0, first_heard_days=210.0), + Node("Laval-Chomedey", 0xCF, _R, 45.5640, -73.7450, 7.0, first_heard_days=190.0), + Node("Verdun-Quai", 0xF3, _R, 45.4555, -73.5640, 5.0, first_heard_days=240.0), + Node("Lachine", 0x85, _R, 45.4340, -73.6800, 7.0, first_heard_days=160.0), + Node("Pointe-Claire", 0xAA, _R, 45.4485, -73.8165, 8.0, first_heard_days=130.0), + Node("Île-Bizard", 0x3C, _R, 45.4950, -73.8900, 6.0, first_heard_days=95.0), + Node("Longueuil", 0x73, _R, 45.5300, -73.5150, 6.0, first_heard_days=6.0), + Node("Brossard", 0x18, _R, 45.4570, -73.4660, 6.0, first_heard_days=70.0), + Node("Mont-St-Bruno", 0xE1, _R, 45.5530, -73.3260, 13.0, first_heard_days=280.0), + Node("Mont-St-Hilaire", 0x61, _R, 45.5490, -73.1570, 15.0, first_heard_days=45.0), + # The room servers. + Node("Le Babillard", 0x3B, _ROOM, 45.5225, -73.5840, 4.0, first_heard_days=330.0), + Node("Café Mesh", 0x64, _ROOM, 45.5250, -73.5990, 3.0, first_heard_days=60.0, + quiet_hours=9 * 24.0), + # The sensors. + Node("Météo-Verdun", 0x58, _S, 45.4585, -73.5660, 3.0, first_heard_days=200.0, + talk=0.6), + Node("Lac-aux-Castors", 0x9E, _S, 45.4985, -73.5970, 4.0, first_heard_days=150.0, + talk=0.6), + Node("Niveau-Fleuve", 0xCB, _S, 45.5110, -73.5470, 2.0, first_heard_days=110.0, + talk=0.6, quiet_hours=30.0), + # The companions: the people of the mesh. + Node("Marie-Ève", 0xE6, _C, 45.5255, -73.5790, 1.0, first_heard_days=320.0, + talk=1.6), + Node("Jean-Seb", 0xAD, _C, 45.5590, -73.6650, 2.0, first_heard_days=300.0, + talk=1.4), + Node("Félix", 0x1A, _C, 45.5450, -73.5440, 2.0, first_heard_days=280.0, + talk=1.2), + Node("Dave-Lachine", 0x84, _C, 45.4385, -73.6710, 2.0, first_heard_days=150.0), + Node("Kim-NDG", 0xBF, _C, 45.4735, -73.6145, 1.0, first_heard_days=260.0), + Node("Bixi Bandit", 0x94, _C, 45.5110, -73.5695, 0.0, shares_location=False, + first_heard_days=90.0, talk=1.3), + Node("Noor", 0x50, _C, 45.5075, -73.6880, 1.0, first_heard_days=75.0), + Node("Chloé L.", 0xC9, _C, 45.5170, -73.6060, 1.0, first_heard_days=220.0), + Node("Sam Verdun", 0x68, _C, 45.4610, -73.5690, 1.0, first_heard_days=240.0), + Node("Poutine Patrol", 0x22, _C, 45.5340, -73.5050, 1.0, first_heard_days=40.0), + Node("Olivier.T", 0x33, _C, 45.5455, -73.5770, 1.0, first_heard_days=180.0, + quiet_hours=50.0), + Node("Raphaël", 0x0B, _C, 45.5810, -73.7480, 1.0, first_heard_days=130.0, + quiet_hours=5.0), + Node("Ana Brossard", 0x3F, _C, 45.4550, -73.4710, 1.0, first_heard_days=65.0, + quiet_hours=14.0), + Node("Léa", 0xDD, _C, 45.5450, -73.6240, 1.0, shares_location=False, + first_heard_days=200.0, talk=0.5), + Node("Rachid", 0x60, _C, 45.5240, -73.6000, 1.0, shares_location=False, + first_heard_days=100.0, quiet_hours=4 * 24.0, talk=0.5), + Node("Tomás", 0x77, _C, 45.4520, -73.8120, 1.0, first_heard_days=160.0, + quiet_hours=62 * 24.0, talk=0.5), + Node("Mathilde", 0xF7, _C, 45.4780, -73.5870, 0.0, first_heard_days=0.05), + Node("Sylvie-Gaspésie", 0x6C, _C, 48.8300, -64.4800, 0.0, shares_location=False, + first_heard_days=0.0, quiet_hours=None), + # Discovered nodes: heard, but nobody added them. + Node("T1000-E", 0x4C, _C, 45.5600, -73.7100, 0.0, added=False, + first_heard_days=3.0, talk=0.4), + Node("Wio Tracker", 0xA0, _C, 45.4650, -73.5480, 0.0, added=False, + first_heard_days=12.0, quiet_hours=20.0, talk=0.4), + Node("Station Mobile", 0xB2, _C, 45.5200, -73.5300, 0.0, shares_location=False, + added=False, first_heard_days=1.5, talk=0.4), +) +# fmt: on + +#: Each node of the world by name, our node included. +BY_NAME: dict[str, Node] = {node.name: node for node in (SELF, *NODES)} + +#: The repeaters, in the order of :data:`NODES`. +REPEATERS: tuple[Node, ...] = tuple(node for node in NODES if node.is_repeater) + + +def node(name: str) -> Node: + """The node of the world with this name. A name that the world does not have is a bug.""" + return BY_NAME[name] + + +# --------------------------------------------------------------------------- the radio + +#: The weakest SNR (dB) at which the radio still decodes a packet. The demo device uses the +#: North American preset (SF7 at 62.5 kHz), whose demodulator stops near -7.5 dB. +FLOOR_DB = -7.5 + +#: The SNR (dB) of a link between two nodes at street level, one kilometre apart. +_ONE_KM_DB = 5.0 + +#: The loss (dB) for each decade of distance. Free space loses 20 dB for each decade. A +#: city loses more, because of the buildings. +_DECADE_DB = 25.0 + +#: Above this SNR (dB), a reading grows slowly: a receiver gives no SNR much above +12 dB, +#: whatever the strength of the signal. +_SOFT_CAP_DB = 10.0 + + +def _skew(a: Node, b: Node) -> float: + """A fixed offset (dB) for the link from ``a`` to ``b``: the terrain that no model knows. + + Two parts make the offset. The larger part is the same in each direction (a hill or a + tower between the two sites). The smaller part depends on the direction, because the + two radios are not the same. Thus a pair of nodes never has exactly the same SNR in the + two directions, as on a real mesh. + """ + pair = "|".join(sorted((a.name, b.name))) + common = hashlib.sha256(f"terrain/{pair}".encode()).digest() + oneway = hashlib.sha256(f"radio/{a.name}>{b.name}".encode()).digest() + return (common[0] / 255 - 0.5) * 3.0 + (oneway[0] / 255 - 0.5) * 1.0 + + +@functools.cache +def link_snr(a: Node, b: Node) -> float | None: + """The mean SNR (dB) at ``b`` of a packet that ``a`` transmits. + + The model is a log-distance path loss: :data:`_ONE_KM_DB` at one kilometre, less + :data:`_DECADE_DB` for each decade of distance, plus the gain of the two sites and the + fixed terrain offset of the link (:func:`_skew`). It is simple, but it gives what a + real mesh shows: the mountains hear each other over 30 km, and two pockets in the city + hear each other over one or two kilometres. + + The world is fixed, so the function keeps each answer in a cache. The history asks for + the same links almost two million times. + + Args: + a: The node that transmits. + b: The node that receives. + + Returns: + The mean SNR in dB, or ``None`` when ``b`` cannot decode ``a`` (below + :data:`FLOOR_DB`, or the same node). + """ + if a.name == b.name: + return None + km = max(haversine_km(a.lat, a.lon, b.lat, b.lon), 0.05) + snr = _ONE_KM_DB - _DECADE_DB * math.log10(km) + a.site_db + b.site_db + _skew(a, b) + if snr > _SOFT_CAP_DB: + snr = _SOFT_CAP_DB + (snr - _SOFT_CAP_DB) * 0.25 + return round(snr, 2) if snr >= FLOOR_DB else None + + +def loss_chance(snr: float) -> float: + """The probability that one packet on a link with this mean SNR is lost. + + A link well above the floor loses nothing. A link near the floor loses most packets. + """ + return max(0.0, min(0.9, (FLOOR_DB + 4.0 - snr) / 8.0)) + + +def reading(snr: float, rng: random.Random) -> float: + """One SNR reading of a link: the mean, plus noise, in the quarter-dB steps of the firmware.""" + return round((snr + rng.gauss(0.0, 1.0)) * 4) / 4 + + +def hears(a: Node, b: Node) -> bool: + """Whether ``b`` can decode ``a``.""" + return link_snr(a, b) is not None + + +# --------------------------------------------------------------------------- routes + + +@functools.cache +def best_route(origin: Node, target: Node) -> tuple[Node, ...] | None: + """The repeaters on the best route from ``origin`` to ``target``, in order. + + The cost of each link is one hop, plus a penalty for a weak link. Thus the route has + few hops, but it does not go over a link that loses half of its packets when a strong + detour of the same length exists. A device learns a route of this type from the floods + that it receives. + + Args: + origin: The node at the start. + target: The node at the end. + + Returns: + The relays between the two ends (an empty tuple when they hear each other + directly), or ``None`` when no route connects them. + """ + direct = link_snr(origin, target) + if direct is not None and direct > FLOOR_DB + 3.0: + return () + costs: dict[str, float] = {} + queue: list[tuple[float, int, str, tuple[Node, ...]]] = [] + order = 0 + for relay in REPEATERS: + snr = link_snr(origin, relay) + if snr is not None and relay.name != origin.name: + order += 1 + heapq.heappush(queue, (_link_cost(snr), order, relay.name, (relay,))) + best: tuple[float, tuple[Node, ...]] | None = None + if direct is not None: + best = (_link_cost(direct), ()) + while queue: + cost, _, name, route = heapq.heappop(queue) + if name in costs: + continue + costs[name] = cost + last = route[-1] + tail = link_snr(last, target) + if tail is not None: + total = cost + _link_cost(tail) + if best is None or total < best[0]: + best = (total, route) + for relay in REPEATERS: + if relay.name in costs or relay.name == origin.name: + continue + snr = link_snr(last, relay) + if snr is not None: + order += 1 + heapq.heappush(queue, (cost + _link_cost(snr), order, relay.name, (*route, relay))) + return best[1] if best is not None else None + + +def _link_cost(snr: float) -> float: + """The cost of one link for :func:`best_route`: one hop, more when the link is weak.""" + return 1.0 + 4.0 * loss_chance(snr) + + +@dataclass(frozen=True, slots=True) +class Copy: + """One copy of a flood that our node hears. + + Attributes: + relays: The repeaters that the copy went through, in order. Empty when our node + heard the origin directly. + snr: The SNR of the reception at our node (dB), from the last relay. + delay_s: The time from the transmission at the origin to the reception (seconds). + """ + + relays: tuple[Node, ...] + snr: float + delay_s: float + + +def flood(origin: Node, rng: random.Random, *, listener: Node = SELF) -> list[Copy]: + """The copies of one flood from ``origin`` that ``listener`` hears, earliest first. + + Each repeater relays the first copy that it receives, and only that copy. Thus the + function runs the flood as a race: each link adds the airtime of the packet and the + random delay of the relay, and each repeater keeps the route of its first copy. Our node + hears the origin directly when it is in range, and each repeater in range that the + flood reached. A weak link can lose a copy (:func:`loss_chance`). Thus two floods from + the same node do not always arrive on the same paths. + + Args: + origin: The node that transmits the flood. + rng: The source of the delays, the losses, and the noise. + listener: The node that hears the copies (our node by default). + + Returns: + The copies that the listener receives. + """ + arrivals: dict[Node, tuple[float, tuple[Node, ...]]] = {} + queue: list[tuple[float, int, Node, tuple[Node, ...]]] = [] + skip = (origin, listener) + order = 0 + for relay, snr in _hearers(origin): + if relay in skip or rng.random() < loss_chance(snr): + continue + order += 1 + heapq.heappush(queue, (_airtime(rng), order, relay, (relay,))) + while queue: + when, _, relay, route = heapq.heappop(queue) + if relay in arrivals: + continue + arrivals[relay] = (when, route) + sent = when + rng.uniform(0.2, 1.4) # the random transmit delay of the repeater + for other, snr in _hearers(relay): + if other in arrivals or other in skip or rng.random() < loss_chance(snr): + continue + order += 1 + heapq.heappush(queue, (sent + _airtime(rng), order, other, (*route, other))) + copies: list[Copy] = [] + direct = link_snr(origin, listener) + if direct is not None and rng.random() >= loss_chance(direct): + copies.append(Copy((), reading(direct, rng), _airtime(rng))) + for when, route in arrivals.values(): + relay = route[-1] + snr = link_snr(relay, listener) + if snr is None or rng.random() < loss_chance(snr): + continue + sent = when + rng.uniform(0.2, 1.4) + copies.append(Copy(route, reading(snr, rng), sent + _airtime(rng))) + copies.sort(key=lambda copy: copy.delay_s) + return copies + + +@functools.cache +def _hearers(node: Node) -> tuple[tuple[Node, float], ...]: + """The repeaters that hear ``node``, each with the mean SNR of the link.""" + return tuple((relay, snr) for relay in REPEATERS if (snr := link_snr(node, relay)) is not None) + + +def _airtime(rng: random.Random) -> float: + """The time on the air of one short packet at SF7, with the time to decode it (seconds).""" + return rng.uniform(0.15, 0.4) + + +@dataclass(frozen=True, slots=True) +class TraceWalk: + """The result of one simulated trace along a forced path. + + Attributes: + snrs: The SNR reading at the receiving end of each link, in order: the first relay, + then each next relay, then our node for the last link. It is shorter than the + walk when the trace died on the way. + success: Whether the trace came back to our node. + rtt_ms: The round trip of the trace (milliseconds), when it came back. + """ + + snrs: list[float] = field(default_factory=list) + success: bool = False + rtt_ms: float | None = None + + +def walk(relays: Sequence[Node], rng: random.Random) -> TraceWalk: + """Simulate one trace from our node through ``relays`` and back to our node. + + A trace is a direct packet: each relay transmits it to the next node of the path, and + measures the SNR of what it received. The last link comes back to our node. A link + that does not exist, or a link that loses the packet, stops the trace. + + Args: + relays: The forced path, as nodes. + rng: The source of the noise and the losses. + + Returns: + The readings, and whether the trace came back. + """ + chain = [SELF, *relays, SELF] + snrs: list[float] = [] + elapsed = 0.0 + for a, b in pairwise(chain): + snr = link_snr(a, b) + if snr is None or rng.random() < loss_chance(snr): + return TraceWalk(snrs, False, None) + snrs.append(reading(snr, rng)) + elapsed += _airtime(rng) + rng.uniform(0.05, 0.25) + return TraceWalk(snrs, True, round(elapsed * 1000.0, 1)) + + +# --------------------------------------------------------------------------- channels + + +@dataclass(frozen=True, slots=True) +class Channel: + """One channel in a slot of the demo device. + + Attributes: + slot: The slot index on the device. + name: The name of the channel as the device stores it. + secret: The 16-byte secret. + scope: The region under which MeshTerm sends on this channel, or ``None`` for the + default scope of the device. + """ + + slot: int + name: str + secret: bytes + scope: str | None = None + + +#: The channels of the demo device. ``Famille`` is a private channel, with a fixed secret +#: that is not derived from its name. +CHANNELS: tuple[Channel, ...] = ( + Channel(0, "Public", DEFAULT_PUBLIC_SECRET), + Channel(1, "#montreal", derive_secret("#montreal"), scope="mtl"), + Channel(2, "#velo", derive_secret("#velo")), + Channel(3, "Famille", hashlib.sha256(b"meshterm-demo/Famille").digest()[:16]), +) + +#: Hashtag channels that other people of the mesh use and our device does not have. Our +#: node hears their packets, but it cannot read them. +FOREIGN_CHANNELS: tuple[bytes, ...] = tuple( + derive_secret(name) for name in ("#quebec", "#radioamateur", "#test", "#plateau") +) + +# --------------------------------------------------------------------------- regions + +#: The region table of each repeater, as ``(name, parent, allows floods)`` rows, in the +#: form that :class:`~meshterm.core.region_sim.SimulatedRegionMap` takes. The island +#: carries ``mtl``, the north shore ``laval``, and the south shore ``rive-sud``. Each +#: repeater also carries the region of the whole province. +REGION_TABLES: dict[str, list[tuple[str, str, bool]]] = {} + +_ISLAND = [("qc", "*", True), ("mtl", "qc", True), ("mtl-ouest", "mtl", True)] +_NORTH = [("qc", "*", True), ("mtl", "qc", True), ("laval", "qc", True)] +_SOUTH = [("qc", "*", True), ("mtl", "qc", False), ("rive-sud", "qc", True)] +for _name in ("Oratoire", "Centre-Ville", "Stade-Olympique", "Verdun-Quai", "Ahuntsic"): + REGION_TABLES[_name] = _ISLAND +for _name in ("Lachine", "Pointe-Claire", "Île-Bizard"): + REGION_TABLES[_name] = [*_ISLAND, ("ouest-de-l-ile", "mtl-ouest", True)] +REGION_TABLES["Laval-Chomedey"] = _NORTH +for _name in ("Longueuil", "Brossard", "Mont-St-Bruno"): + REGION_TABLES[_name] = _SOUTH +REGION_TABLES["Mont-St-Hilaire"] = [*_SOUTH, ("monteregie", "rive-sud", True)] +# The hub of the island carries each region. Its table is too long for one reply, so the +# region editor shows how it gets the rest. +REGION_TABLES["Mont-Royal"] = [ + ("qc", "*", True), + ("mtl", "qc", True), + ("mtl-ouest", "mtl", True), + ("mtl-est", "mtl", True), + ("ouest-de-l-ile", "mtl-ouest", True), + ("laval", "qc", True), + ("rive-sud", "qc", True), + ("monteregie", "rive-sud", False), + ("rive-nord", "qc", False), + ("plateau", "mtl", True), +] + +#: The contacts to which the device learned a route. A real device learns few routes: it +#: learns one only from a flood that it received from the node, and forgets it when a send +#: fails. Thus most contacts of the demo stay on flood, as on hardware. +ROUTED: tuple[str, ...] = ( + "Mont-Royal", + "Stade-Olympique", + "Le Babillard", + "Marie-Ève", + "Jean-Seb", + "Dave-Lachine", + "Bixi Bandit", + "Mathilde", +) + + +def device_route(target: Node) -> tuple[str, ...] | None: + """The route that the demo device holds for ``target``, as path hashes. + + Returns: + The hashes of the relays (empty for a direct neighbour), or ``None`` when the + device holds no route (the contact is on flood). + """ + if target.name not in ROUTED: + return None + route = best_route(SELF, target) + return None if route is None else tuple(relay.hash for relay in route) + + +#: The repeaters for which our node remembers the admin password. The demo device accepts +#: this password on each repeater, as the simulator does. +ADMIN_PASSWORD = "admin" +ADMINISTERED: tuple[str, ...] = ("Mont-Royal", "Stade-Olympique") + +#: The room that our node joined, and its room password (the stock password of a room +#: server, which the simulator accepts). +JOINED_ROOM = "Le Babillard" +ROOM_PASSWORD = "hello" diff --git a/meshterm/persistence/repository.py b/meshterm/persistence/repository.py index 05c20da..12602e9 100644 --- a/meshterm/persistence/repository.py +++ b/meshterm/persistence/repository.py @@ -75,6 +75,17 @@ ACTIVITY_DRAWN_BUCKETS = 24 #: recent past gets several readings for a median. HOP_EVIDENCE_LIMIT = 20000 +#: The statement that stores one observation. :meth:`Repository.record_observation` and +#: :meth:`Repository.record_observations` share it, with the values of +#: :meth:`Repository._observation_row`. +_INSERT_OBSERVATION = ( + "INSERT INTO observations " + "(run_id, node, public_key, name, kind, node_type, snr, rssi, lat, lon, path, " + "observed_at, chan_hash, cipher_mac, crypted, payload_typename, dest, src, tag, " + "trace_snrs, route, transport_code, scope_body) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +) + def _packet_raw(row: sqlite3.Row) -> dict | None: """Rebuild the minimum raw payload with which MeshTerm reads a stored ``packet`` row. @@ -350,7 +361,14 @@ class Repository: # -- runs ------------------------------------------------------------------- - def start_run(self, tool: str, args: dict[str, Any], profile: str | None = None) -> int: + def start_run( + self, + tool: str, + args: dict[str, Any], + profile: str | None = None, + *, + at: datetime | None = None, + ) -> int: """Store the start of a tool run. Args: @@ -358,6 +376,8 @@ class Repository: args: The arguments with which the tool was called (they must be JSON-serializable). profile: The name of the active device profile, if any. + at: When the run started. The default is now. Only the demo history gives + another time, because it writes the runs of past days. Returns: The primary key of the new run. @@ -365,25 +385,33 @@ class Repository: cur = self._conn.execute( "INSERT INTO runs (tool, profile, args_json, status, started_at) " "VALUES (?, ?, ?, 'running', ?)", - (tool, profile, json.dumps(args, default=str), utcnow().isoformat()), + (tool, profile, json.dumps(args, default=str), (at or utcnow()).isoformat()), ) self._conn.commit() return int(cur.lastrowid) - def finish_run(self, run_id: int, status: str, summary: dict | None = None) -> None: + def finish_run( + self, + run_id: int, + status: str, + summary: dict | None = None, + *, + at: datetime | None = None, + ) -> None: """Mark a run as finished. Args: run_id: The run to update. status: The final status (``ok`` or ``error``). summary: An optional JSON-serializable summary of the result. + at: When the run finished. The default is now (refer to :meth:`start_run`). """ self._conn.execute( "UPDATE runs SET status = ?, summary_json = ?, finished_at = ? WHERE id = ?", ( status, json.dumps(summary, default=str) if summary is not None else None, - utcnow().isoformat(), + (at or utcnow()).isoformat(), run_id, ), ) @@ -683,7 +711,12 @@ class Repository: return paths def record_neighbours( - self, run_id: int, repeater: str, neighbours: list[NeighbourInfo] + self, + run_id: int, + repeater: str, + neighbours: list[NeighbourInfo], + *, + at: datetime | None = None, ) -> None: """Store one snapshot of the neighbour table that MeshTerm got from a remote repeater. @@ -699,8 +732,10 @@ class Repository: repeater: The canonical id of the repeater that sent the table. neighbours: The entries that MeshTerm got (this list can be empty: then no rows are stored). + at: When MeshTerm got the table. The default is now (refer to + :meth:`start_run`). """ - fetched_at = utcnow().isoformat() + fetched_at = (at or utcnow()).isoformat() self._conn.executemany( "INSERT INTO neighbour_reports " "(run_id, repeater, neighbour, snr, heard_at, fetched_at) " @@ -840,6 +875,7 @@ class Repository: app_version: str, keep: int = 5, ascending: bool = False, + at: datetime | None = None, ) -> int | None: """Offer one walk to a category leaderboard. Store it only if it gets a place. @@ -862,11 +898,13 @@ class Repository: keep: The board size (the rows kept for each category and width). ascending: ``True`` for categories in which lower scores win (Thin thread looks for the weakest link that survives). + at: When the walk was made. The default is now (refer to :meth:`start_run`). Returns: The id of the stored row when the walk got a place (a new row, or a better score on a known route), or ``None`` when the walk did not get on the board. """ + when = (at or utcnow()).isoformat() def beats(challenger: float, standing: float) -> bool: return challenger < standing if ascending else challenger > standing @@ -888,7 +926,7 @@ class Repository: score, json.dumps(stats), app_version, - utcnow().isoformat(), + when, existing["id"], ), ) @@ -906,7 +944,7 @@ class Repository: score, json.dumps(stats), app_version, - utcnow().isoformat(), + when, ), ) new_id = int(cursor.lastrowid) @@ -1043,6 +1081,32 @@ class Repository: run_id: The run that owns the observation. obs: The observation to store. """ + self._conn.execute(_INSERT_OBSERVATION, self._observation_row(run_id, obs)) + self._conn.commit() + + def record_observations(self, run_id: int, observations: Iterable[Observation]) -> int: + """Store many overheard packets of one run in one transaction. + + The rows are the same as the rows of :meth:`record_observation`. The difference is + the commit: one for all the rows, not one for each row. The demo history writes + tens of thousands of packets at each start, and a commit for each row costs more + than a second for each thousand rows on an SD card. + + Args: + run_id: The run that owns the observations. + observations: The observations to store. + + Returns: + The number of stored rows. + """ + rows = [self._observation_row(run_id, obs) for obs in observations] + self._conn.executemany(_INSERT_OBSERVATION, rows) + self._conn.commit() + return len(rows) + + @staticmethod + def _observation_row(run_id: int, obs: Observation) -> tuple: + """The values of one ``observations`` row, in the order of :data:`_INSERT_OBSERVATION`.""" chan_hash = cipher_mac = crypted = None raw = obs.raw if isinstance(obs.raw, dict) else {} # The payload class identifies a relayed 'packet' that names no origin node. Keep @@ -1082,39 +1146,31 @@ class Repository: transport_code = raw.get("transport_code") if route in SCOPED_ROUTES else None body = raw_scope_body(raw) if route == "TC_FLOOD" and transport_code else None scope_body = body.hex() if body is not None else None - self._conn.execute( - "INSERT INTO observations " - "(run_id, node, public_key, name, kind, node_type, snr, rssi, lat, lon, path, " - "observed_at, chan_hash, cipher_mac, crypted, payload_typename, dest, src, tag, " - "trace_snrs, route, transport_code, scope_body) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - run_id, - obs.node, - obs.public_key, - obs.name, - obs.kind, - obs.node_type, - obs.snr, - obs.rssi, - obs.lat, - obs.lon, - obs.path, - obs.observed_at.isoformat(), - chan_hash, - cipher_mac, - crypted, - typename, - dest, - src, - tag, - trace_snrs, - route, - transport_code, - scope_body, - ), + return ( + run_id, + obs.node, + obs.public_key, + obs.name, + obs.kind, + obs.node_type, + obs.snr, + obs.rssi, + obs.lat, + obs.lon, + obs.path, + obs.observed_at.isoformat(), + chan_hash, + cipher_mac, + crypted, + typename, + dest, + src, + tag, + trace_snrs, + route, + transport_code, + scope_body, ) - self._conn.commit() def observation_count(self) -> int: """Return the total number of overheard packets stored in all the runs. diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index cf105dc..d258c7d 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -348,7 +348,12 @@ def _header_segments(ctx: AppContext, cache: dict) -> list[Text]: if "device" in atoms: device = Text() - if ctx.mock: + if getattr(ctx, "demo", False): + # The demo shows the name of its node, as a real session does, and says that + # it is the demo where a real session says how the device is connected. + device.append(_demo_name()) + device.append(" (demo)", style="muted") + elif ctx.mock: device.append("simulator", style="warn") else: name, where = _device_label(ctx, cache) @@ -375,6 +380,13 @@ def _header_segments(ctx: AppContext, cache: dict) -> list[Text]: return segments +def _demo_name() -> str: + """The name of our node in the demo world.""" + from ..demo.world import SELF + + return SELF.name + + def _device_label(ctx: AppContext, cache: dict) -> tuple[str, str]: """The device segment of the header: ``(node name, where)``, cached for each connection. diff --git a/tests/test_demo.py b/tests/test_demo.py new file mode 100644 index 0000000..125984d --- /dev/null +++ b/tests/test_demo.py @@ -0,0 +1,84 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The demo world: the same history at each run, and a device that agrees with it.""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta +from pathlib import Path + +from meshterm.core.channels import channel_identity +from meshterm.core.models import utcnow +from meshterm.demo import history, world +from meshterm.demo.device import DemoDevice +from meshterm.demo.home import prepare +from meshterm.persistence.repository import Repository + + +def _seed(path: Path, now) -> Repository: # noqa: ANN001 - a datetime + repo = Repository(path) + history.seed(repo, now) + return repo + + +def test_each_node_has_its_own_hash() -> None: + """No two nodes of the world share a path hash, so each hop resolves to one node.""" + hashes = [node.hash for node in (world.SELF, *world.NODES)] + assert len(hashes) == len(set(hashes)) + + +def test_the_history_is_the_same_at_each_run(tmp_path: Path) -> None: + """Two runs give the same rows, with the times moved by the gap between the two starts.""" + now = utcnow() + first = _seed(tmp_path / "a.db", now) + second = _seed(tmp_path / "b.db", now + timedelta(hours=5)) + assert first.table_counts() == second.table_counts() + assert first.kind_counts() == second.kind_counts() + public = world.CHANNELS[0] + identity = channel_identity(public.name, public.secret) + + def lines(repo: Repository, start) -> list[tuple[str, int]]: # noqa: ANN001 - a datetime + rows = repo.recent_chat_messages(is_channel=True, channel_id=identity, limit=500) + return [(m.text, round((start - m.created_at).total_seconds())) for m in rows] + + assert lines(first, now) == lines(second, now + timedelta(hours=5)) + + +def test_a_channel_line_has_its_heard_copies(tmp_path: Path) -> None: + """The packets of a channel line decrypt to the line, so its paths can be found.""" + from meshterm.services.message_paths import channel_arrivals + + repo = _seed(tmp_path / "h.db", utcnow()) + public = world.CHANNELS[0] + identity = channel_identity(public.name, public.secret) + last = repo.recent_chat_messages(is_channel=True, channel_id=identity, limit=1)[-1] + arrivals = channel_arrivals(repo, last, channel_name=public.name, secret=public.secret) + assert arrivals + + +def test_the_demo_device_traces_a_path() -> None: + """A forced path through linked repeaters comes back with a reading for each link.""" + + async def run(): # noqa: ANN202 + device = DemoDevice() + await device.connect() + spec = ",".join(world.node(n).hash for n in ("Mont-Royal", "Oratoire", "Mont-Royal")) + for _ in range(10): + result = await device.run_trace("(path)", path=spec) + if result.success: + return result + return result + + result = asyncio.run(run()) + assert result.success + assert [h.node for h in result.hops] == ["2a", "4f", "2a", None] + + +def test_prepare_writes_a_home(tmp_path: Path) -> None: + """The demo home gets a database and the stores of the joined room and the alerts.""" + home = tmp_path / "demo" + prepare(home, utcnow()) + prepare(home, utcnow()) # a second start writes it again over the first + assert (home / "meshterm.db").exists() + assert (home / "rooms.json").exists() + assert (home / "watchtower.json").exists()