diff --git a/meshtools/core/connection.py b/meshtools/core/connection.py index ced8a66..c41f919 100644 --- a/meshtools/core/connection.py +++ b/meshtools/core/connection.py @@ -22,7 +22,16 @@ from typing import Callable, Optional from .channels import CHANNEL_SLOT_PROBE_CAP from .events import MeshEvent -from .models import Ack, Contact, Hop, Message, Observation, TraceResult +from .models import ( + NODE_TYPE_CHAT, + NODE_TYPE_REPEATER, + Ack, + Contact, + Hop, + Message, + Observation, + TraceResult, +) #: Module logger; enable DEBUG on ``meshtools.core.connection`` to trace the message pump. _log = logging.getLogger(__name__) @@ -1265,10 +1274,14 @@ class MockDevice(Device): ``advert`` otherwise, carrying a location for the simulated repeaters. """ lat_lon = self._MOCK_LOCATIONS.get(contact.name) + # The two located nodes are the simulated repeaters (fixed infrastructure); the rest + # advertise as ordinary chat nodes, so the map has both classes to prioritise. + node_type = NODE_TYPE_REPEATER if contact.name in self._MOCK_LOCATIONS else NODE_TYPE_CHAT return Observation( node=contact.key_prefix or contact.public_key[:12], name=contact.name, kind="telemetry" if seq % 4 == 3 else "advert", + node_type=node_type, snr=round(self._rng.gauss(6.0, 3.0), 1), rssi=round(self._rng.gauss(-95.0, 8.0), 1), lat=lat_lon[0] if lat_lon else None, @@ -1428,6 +1441,7 @@ def observation_from_event(event, kind: str) -> Optional[Observation]: # noqa: node=node, name=payload.get("adv_name") or payload.get("name"), kind=kind, + node_type=_as_int(payload.get("adv_type", payload.get("type"))), snr=_as_float(payload.get("snr")), rssi=_as_float(payload.get("rssi")), lat=_as_float(lat) if lat else None, @@ -1487,6 +1501,16 @@ def _as_float(value: object) -> Optional[float]: return None +def _as_int(value: object) -> Optional[int]: + """Best-effort int conversion, returning ``None`` on missing/garbage values.""" + if value is None: + return None + try: + return int(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None + + def _advert_time(last_advert: object) -> Optional[datetime]: """Convert a contact's ``last_advert`` Unix timestamp to a UTC datetime. diff --git a/meshtools/core/geo.py b/meshtools/core/geo.py new file mode 100644 index 0000000..5fd3e07 --- /dev/null +++ b/meshtools/core/geo.py @@ -0,0 +1,281 @@ +"""Pure geographic helpers for the terminal map: Web Mercator, tiles, and viewports. + +This module is deliberately I/O-free and rendering-free so it can be unit-tested without a +radio, a terminal, or the network. It turns latitude/longitude into the Web Mercator "world +pixel" space that slippy-map vector tiles live in, models the on-screen :class:`Viewport` +(which braille dot maps to which coordinate), and measures great-circle distances. + +The map renders one braille **dot** per Web Mercator pixel at the viewport's zoom, so a +viewport of ``dot_w`` × ``dot_h`` dots shows exactly that many mercator pixels — panning and +zooming are then just moving and scaling this window over the world. Braille sub-cells are +about square in a monospace font (2 dots wide, 4 tall, in a cell roughly twice as tall as +wide), so mercator pixels land on screen without obvious distortion. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + +#: Mean radius of the Earth in kilometres, used for haversine distances. +EARTH_RADIUS_KM = 6371.0088 + +#: Web Mercator tile edge in pixels at its own zoom — the slippy-map convention. The world +#: is ``TILE_PX * 2**zoom`` pixels square at a given zoom. +TILE_PX = 256 + + +@dataclass(frozen=True, slots=True) +class BBox: + """An axis-aligned latitude/longitude bounding box (decimal degrees).""" + + min_lat: float + min_lon: float + max_lat: float + max_lon: float + + @classmethod + def around(cls, points: list[tuple[float, float]]) -> "BBox": + """Return the tightest box containing every ``(lat, lon)`` point. + + Args: + points: One or more latitude/longitude pairs (must be non-empty). + + Returns: + The enclosing bounding box. + + Raises: + ValueError: If ``points`` is empty. + """ + if not points: + raise ValueError("cannot build a bounding box from no points") + lats = [lat for lat, _ in points] + lons = [lon for _, lon in points] + return cls(min(lats), min(lons), max(lats), max(lons)) + + @property + def center(self) -> tuple[float, float]: + """The box's centre as a ``(lat, lon)`` pair.""" + return ((self.min_lat + self.max_lat) / 2, (self.min_lon + self.max_lon) / 2) + + +def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: + """Return the great-circle distance between two points in kilometres. + + Args: + lat1: First point's latitude (degrees). + lon1: First point's longitude (degrees). + lat2: Second point's latitude (degrees). + lon2: Second point's longitude (degrees). + + Returns: + The distance along the Earth's surface in kilometres. + """ + p1, p2 = math.radians(lat1), math.radians(lat2) + dphi = math.radians(lat2 - lat1) + dlambda = math.radians(lon2 - lon1) + a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2 + return 2 * EARTH_RADIUS_KM * math.asin(math.sqrt(a)) + + +def clamp_lat(lat: float) -> float: + """Clamp latitude to the Web Mercator limit (~±85.051°) where the projection is finite.""" + return max(-85.05112878, min(85.05112878, lat)) + + +def lonlat_to_world(lat: float, lon: float, zoom: float) -> tuple[float, float]: + """Project ``(lat, lon)`` to Web Mercator world pixels at ``zoom``. + + Args: + lat: Latitude in degrees (clamped to the mercator limit). + lon: Longitude in degrees. + zoom: Zoom level (may be fractional). + + Returns: + ``(x, y)`` in world pixels, where the world spans ``TILE_PX * 2**zoom`` on each axis + and ``y`` grows southward (north is up / smaller ``y``). + """ + scale = TILE_PX * (2.0**zoom) + x = (lon + 180.0) / 360.0 * scale + s = math.sin(math.radians(clamp_lat(lat))) + y = (0.5 - math.log((1 + s) / (1 - s)) / (4 * math.pi)) * scale + return x, y + + +def world_to_lonlat(x: float, y: float, zoom: float) -> tuple[float, float]: + """Invert :func:`lonlat_to_world`: world pixels at ``zoom`` back to ``(lat, lon)``. + + Args: + x: World-pixel x. + y: World-pixel y. + zoom: The zoom the pixels were computed at. + + Returns: + The ``(lat, lon)`` in degrees. + """ + scale = TILE_PX * (2.0**zoom) + lon = x / scale * 360.0 - 180.0 + n = math.pi - 2 * math.pi * y / scale + lat = math.degrees(math.atan(math.sinh(n))) + return lat, lon + + +@dataclass(frozen=True, slots=True) +class Viewport: + """The on-screen window over the world: what each braille dot maps to. + + One dot equals one Web Mercator pixel at :attr:`zoom`, so the visible area is exactly + ``dot_w`` × ``dot_h`` mercator pixels centred on ``(center_lat, center_lon)``. Panning and + zooming return new viewports; nothing here mutates. + + Attributes: + center_lat: Latitude at the centre of the view. + center_lon: Longitude at the centre of the view. + zoom: Display zoom (integer here; may exceed the tile source's max, in which case + lower-zoom tiles are magnified — see :meth:`tiles` / :meth:`feature_to_dot`). + dot_w: Viewport width in braille dots. + dot_h: Viewport height in braille dots. + """ + + center_lat: float + center_lon: float + zoom: int + dot_w: int + dot_h: int + + @property + def origin_world(self) -> tuple[float, float]: + """Top-left corner of the view in world pixels at :attr:`zoom`.""" + cx, cy = lonlat_to_world(self.center_lat, self.center_lon, self.zoom) + return cx - self.dot_w / 2, cy - self.dot_h / 2 + + def lonlat_to_dot(self, lat: float, lon: float) -> tuple[float, float]: + """Project a coordinate to a (possibly off-screen) dot position in the view.""" + ox, oy = self.origin_world + wx, wy = lonlat_to_world(lat, lon, self.zoom) + return wx - ox, wy - oy + + def tile_zoom(self, max_tile_zoom: int) -> int: + """The tile zoom to fetch: the display zoom, capped at the source's max.""" + return max(0, min(self.zoom, max_tile_zoom)) + + def tiles(self, max_tile_zoom: int) -> list[tuple[int, int, int]]: + """Return the ``(z, x, y)`` tiles covering the view at the fetchable tile zoom. + + When the display zoom exceeds ``max_tile_zoom`` the lower-zoom tiles that cover the + same ground are returned (the renderer magnifies them), so zooming in past the + source's limit still works. + + Args: + max_tile_zoom: The highest zoom the tile source actually serves. + + Returns: + Tile coordinates, clamped to the valid range, de-duplicated in row-major order. + """ + tz = self.tile_zoom(max_tile_zoom) + scale = 2.0 ** (self.zoom - tz) # display px per tile-zoom px + ox, oy = self.origin_world + n = 2**tz + # Visible rect in tile-zoom world pixels, then in tile indices. + tx0 = int((ox / scale) // TILE_PX) + tx1 = int(((ox + self.dot_w) / scale) // TILE_PX) + ty0 = int((oy / scale) // TILE_PX) + ty1 = int(((oy + self.dot_h) / scale) // TILE_PX) + out: list[tuple[int, int, int]] = [] + for ty in range(ty0, ty1 + 1): + if not 0 <= ty < n: + continue + for tx in range(tx0, tx1 + 1): + out.append((tz, tx % n, ty)) # wrap x around the antimeridian + return out + + def feature_to_dot( + self, tile_x: int, tile_y: int, tile_zoom: int, extent: int, lx: float, ly: float + ) -> tuple[float, float]: + """Project a tile-local point to a dot position in this viewport. + + Args: + tile_x: The tile's x index at ``tile_zoom``. + tile_y: The tile's y index at ``tile_zoom``. + tile_zoom: The zoom the tile was fetched at. + extent: The tile's internal coordinate extent (e.g. 4096). + lx: Local x within the tile, ``0..extent``. + ly: Local y within the tile, ``0..extent``. + + Returns: + The ``(x, y)`` dot position (may be outside the canvas; the caller clips). + """ + scale = 2.0 ** (self.zoom - tile_zoom) + wx = (tile_x + lx / extent) * TILE_PX * scale + wy = (tile_y + ly / extent) * TILE_PX * scale + ox, oy = self.origin_world + return wx - ox, wy - oy + + def panned(self, frac_x: float, frac_y: float) -> "Viewport": + """Return a viewport shifted by a fraction of its own width/height. + + Args: + frac_x: Eastward shift as a fraction of the view width (negative = west). + frac_y: Southward shift as a fraction of the view height (negative = north). + + Returns: + A new :class:`Viewport` at the shifted centre, same zoom and size. + """ + cx, cy = lonlat_to_world(self.center_lat, self.center_lon, self.zoom) + cx += frac_x * self.dot_w + cy += frac_y * self.dot_h + lat, lon = world_to_lonlat(cx, cy, self.zoom) + return Viewport(clamp_lat(lat), lon, self.zoom, self.dot_w, self.dot_h) + + def zoomed(self, delta: int, *, min_zoom: int = 2, max_zoom: int = 19) -> "Viewport": + """Return a viewport zoomed by ``delta`` levels about the same centre (clamped).""" + z = max(min_zoom, min(max_zoom, self.zoom + delta)) + return Viewport(self.center_lat, self.center_lon, z, self.dot_w, self.dot_h) + + def resized(self, dot_w: int, dot_h: int) -> "Viewport": + """Return the same view centred as before but at a new canvas size.""" + return Viewport(self.center_lat, self.center_lon, self.zoom, dot_w, dot_h) + + @classmethod + def fit( + cls, + points: list[tuple[float, float]], + dot_w: int, + dot_h: int, + *, + pad: float = 0.18, + min_zoom: int = 2, + max_zoom: int = 16, + default_zoom: int = 14, + ) -> "Viewport": + """Build a viewport framing ``points`` — centred on them at the tightest fitting zoom. + + Args: + points: Latitude/longitude pairs to frame (may be empty). + dot_w: Canvas width in dots. + dot_h: Canvas height in dots. + pad: Fraction of the canvas kept as margin around the points. + min_zoom: Lowest zoom to consider. + max_zoom: Highest zoom to consider. + default_zoom: Zoom used when the points don't constrain it (0 or 1 point). + + Returns: + A :class:`Viewport` centred on the points at a zoom where they fit with margin. + Empty input centres on the world at ``min_zoom``. + """ + if not points: + return cls(0.0, 0.0, min_zoom, dot_w, dot_h) + box = BBox.around(points) + center_lat, center_lon = box.center + if len(points) == 1 or (box.min_lat == box.max_lat and box.min_lon == box.max_lon): + return cls(center_lat, center_lon, default_zoom, dot_w, dot_h) + avail_w = dot_w * (1 - pad) + avail_h = dot_h * (1 - pad) + chosen = min_zoom + for z in range(max_zoom, min_zoom - 1, -1): + x0, y0 = lonlat_to_world(box.max_lat, box.min_lon, z) # NW corner + x1, y1 = lonlat_to_world(box.min_lat, box.max_lon, z) # SE corner + if abs(x1 - x0) <= avail_w and abs(y1 - y0) <= avail_h: + chosen = z + break + return cls(center_lat, center_lon, chosen, dot_w, dot_h) diff --git a/meshtools/core/models.py b/meshtools/core/models.py index d4bac1d..b4a5919 100644 --- a/meshtools/core/models.py +++ b/meshtools/core/models.py @@ -15,6 +15,22 @@ from typing import Optional #: Label used for our own (local) device when framing a trace path's endpoints. LOCAL_DEVICE_LABEL = "us" +# MeshCore advert types — the low nibble of an advert's flag byte — classify the kind of +# node an advert announces. Repeaters are fixed infrastructure and carry the mesh, so +# location-aware features (e.g. the map) prioritise them over ordinary leaf nodes. +NODE_TYPE_CHAT = 1 +NODE_TYPE_REPEATER = 2 +NODE_TYPE_ROOM = 3 +NODE_TYPE_SENSOR = 4 + +#: Human-readable label for each advert type, for legends and summaries. +NODE_TYPE_LABELS = { + NODE_TYPE_CHAT: "node", + NODE_TYPE_REPEATER: "repeater", + NODE_TYPE_ROOM: "room", + NODE_TYPE_SENSOR: "sensor", +} + def utcnow() -> datetime: """Return a timezone-aware UTC timestamp. @@ -54,6 +70,8 @@ class Observation: node: Key prefix / hash identifying the transmitting node, if known. name: Friendly name advertised by the node, if carried. kind: Packet class, e.g. ``advert`` or ``telemetry``. + node_type: Advert type of the transmitting node (see the ``NODE_TYPE_*`` constants), + e.g. :data:`NODE_TYPE_REPEATER`; ``None`` when the packet did not carry it. snr: Signal-to-noise ratio (dB) the companion measured, if reported. rssi: Received signal strength (dBm), if reported. lat: Advertised latitude (decimal degrees), when the node shares location. @@ -65,6 +83,7 @@ class Observation: node: Optional[str] name: Optional[str] = None kind: str = "advert" + node_type: Optional[int] = None snr: Optional[float] = None rssi: Optional[float] = None lat: Optional[float] = None @@ -277,6 +296,8 @@ class HeardNode: last_seen: Timestamp of the most recent observation. lat: Most recent advertised latitude, if the node shared one. lon: Most recent advertised longitude, if the node shared one. + node_type: Most recent advert type seen for the node (see the ``NODE_TYPE_*`` + constants), if any observation carried it. """ node: Optional[str] @@ -288,12 +309,18 @@ class HeardNode: last_seen: datetime lat: Optional[float] = None lon: Optional[float] = None + node_type: Optional[int] = None @property def has_location(self) -> bool: """Whether this node reported a usable latitude/longitude.""" return self.lat is not None and self.lon is not None + @property + def is_repeater(self) -> bool: + """Whether this node advertised itself as a repeater.""" + return self.node_type == NODE_TYPE_REPEATER + @classmethod def from_observations( cls, node: Optional[str], observations: list["Observation"] @@ -315,6 +342,7 @@ class HeardNode: (o for o in reversed(ordered) if o.lat is not None and o.lon is not None), None ) name = next((o.name for o in reversed(ordered) if o.name), None) + node_type = next((o.node_type for o in reversed(ordered) if o.node_type is not None), None) return cls( node=node, name=name, @@ -325,6 +353,7 @@ class HeardNode: last_seen=latest.observed_at, lat=located.lat if located else None, lon=located.lon if located else None, + node_type=node_type, ) diff --git a/meshtools/core/mvt.py b/meshtools/core/mvt.py new file mode 100644 index 0000000..f481f05 --- /dev/null +++ b/meshtools/core/mvt.py @@ -0,0 +1,278 @@ +"""A tiny, dependency-free Mapbox Vector Tile (MVT) decoder. + +Vector tiles are Protocol-Buffer messages (spec: github.com/mapbox/vector-tile-spec). Rather +than pull in a protobuf runtime and a geometry library, this decodes the small subset the map +needs by hand: the protobuf wire format (varints, length-delimited fields) and the MVT +geometry command encoding (MoveTo/LineTo/ClosePath with zig-zag deltas). It is pure and +offline — the network fetch lives in :mod:`meshtools.services.basemap`. + +The result is a list of :class:`Layer`, each with decoded :class:`Feature` geometry in +*tile-local* integer coordinates (``0..extent``) and a resolved ``tags`` dict, ready to be +projected to the screen by :class:`meshtools.core.geo.Viewport`. +""" + +from __future__ import annotations + +import gzip +from dataclasses import dataclass, field +from typing import Any, Optional + +# MVT geometry types (Feature.type). +GEOM_POINT = 1 +GEOM_LINE = 2 +GEOM_POLYGON = 3 + +# Geometry command ids (low 3 bits of a command integer). +_CMD_MOVE_TO = 1 +_CMD_LINE_TO = 2 +_CMD_CLOSE = 7 + + +class _Reader: + """A minimal protobuf wire-format reader over a byte buffer.""" + + __slots__ = ("b", "i") + + def __init__(self, b: bytes) -> None: + self.b = b + self.i = 0 + + def eof(self) -> bool: + return self.i >= len(self.b) + + def varint(self) -> int: + """Read a base-128 varint.""" + result = shift = 0 + while True: + byte = self.b[self.i] + self.i += 1 + result |= (byte & 0x7F) << shift + if not byte & 0x80: + return result + shift += 7 + + def tag(self) -> tuple[int, int]: + """Read a field tag, returning ``(field_number, wire_type)``.""" + key = self.varint() + return key >> 3, key & 0x7 + + def blob(self) -> bytes: + """Read a length-delimited byte string.""" + n = self.varint() + v = self.b[self.i : self.i + n] + self.i += n + return v + + def skip(self, wire_type: int) -> None: + """Skip a field of the given wire type.""" + if wire_type == 0: + self.varint() + elif wire_type == 2: + self.blob() + elif wire_type == 5: + self.i += 4 + elif wire_type == 1: + self.i += 8 + else: # pragma: no cover - groups are obsolete and never appear in MVT + raise ValueError(f"unsupported wire type {wire_type}") + + +def _zigzag(n: int) -> int: + """Decode a protobuf zig-zag encoded signed integer.""" + return (n >> 1) ^ -(n & 1) + + +@dataclass(slots=True) +class Feature: + """One decoded vector-tile feature. + + Attributes: + geom_type: :data:`GEOM_POINT`, :data:`GEOM_LINE`, or :data:`GEOM_POLYGON`. + rings: The geometry as a list of parts, each a list of ``(x, y)`` integer points in + tile-local coordinates (``0..extent``). Points have one part holding every point; + lines/polygons have one part per line-string / ring. + tags: Resolved attribute dict (e.g. ``{"class": "primary", "name": "Rue X"}``). + """ + + geom_type: int + rings: list[list[tuple[int, int]]] + tags: dict[str, Any] + + def get(self, key: str, default: Any = None) -> Any: + """Return a tag value, or ``default`` if absent.""" + return self.tags.get(key, default) + + @property + def name(self) -> Optional[str]: + """The feature's display name, preferring the local ``name`` then latin fallbacks.""" + for key in ("name", "name:latin", "name:en", "name_en", "name_int"): + val = self.tags.get(key) + if val: + return str(val) + return None + + +@dataclass(slots=True) +class Layer: + """A named vector-tile layer and its decoded features. + + Attributes: + name: Layer id (e.g. ``transportation``, ``water``, ``place``). + extent: The tile's internal coordinate extent (typically 4096). + features: The decoded features. + """ + + name: str + extent: int + features: list[Feature] = field(default_factory=list) + + +def _decode_geometry(cmds: list[int]) -> list[list[tuple[int, int]]]: + """Decode an MVT geometry command stream into a list of point rings.""" + rings: list[list[tuple[int, int]]] = [] + current: list[tuple[int, int]] = [] + x = y = 0 + i = 0 + n = len(cmds) + while i < n: + command = cmds[i] + i += 1 + cmd_id = command & 0x7 + count = command >> 3 + if cmd_id == _CMD_MOVE_TO: + for _ in range(count): + x += _zigzag(cmds[i]) + y += _zigzag(cmds[i + 1]) + i += 2 + # Each MoveTo starts a new part (multipoint keeps them in one part below). + if current: + rings.append(current) + current = [(x, y)] + elif cmd_id == _CMD_LINE_TO: + for _ in range(count): + x += _zigzag(cmds[i]) + y += _zigzag(cmds[i + 1]) + i += 2 + current.append((x, y)) + elif cmd_id == _CMD_CLOSE: + if current: + current.append(current[0]) # close the ring back to its start + if current: + rings.append(current) + return rings + + +def _decode_value(buf: bytes) -> Any: + """Decode an MVT attribute Value message into a Python scalar.""" + r = _Reader(buf) + while not r.eof(): + field_no, wire = r.tag() + if field_no == 1 and wire == 2: + return r.blob().decode("utf-8", "ignore") # string_value + if field_no == 2 and wire == 5: # float_value + import struct + + v = struct.unpack(" list[int]: + """Decode a packed repeated uint32 field into a list of ints.""" + r = _Reader(buf) + out: list[int] = [] + while not r.eof(): + out.append(r.varint()) + return out + + +def _decode_feature(buf: bytes, keys: list[str], values: list[Any]) -> Optional[Feature]: + """Decode a single Feature message, resolving its tags against the layer's key/value pools.""" + r = _Reader(buf) + geom_type = 0 + tag_ints: list[int] = [] + geom_ints: list[int] = [] + while not r.eof(): + field_no, wire = r.tag() + if field_no == 2 and wire == 2: + tag_ints = _packed_uints(r.blob()) + elif field_no == 3 and wire == 0: + geom_type = r.varint() + elif field_no == 4 and wire == 2: + geom_ints = _packed_uints(r.blob()) + else: + r.skip(wire) + if not geom_ints: + return None + tags: dict[str, Any] = {} + for k, v in zip(tag_ints[0::2], tag_ints[1::2]): + if 0 <= k < len(keys) and 0 <= v < len(values): + tags[keys[k]] = values[v] + return Feature(geom_type=geom_type, rings=_decode_geometry(geom_ints), tags=tags) + + +def _decode_layer(buf: bytes) -> Layer: + """Decode a single Layer message and all of its features.""" + r = _Reader(buf) + name = "" + extent = 4096 + keys: list[str] = [] + values: list[Any] = [] + feature_blobs: list[bytes] = [] + while not r.eof(): + field_no, wire = r.tag() + if field_no == 1 and wire == 2: + name = r.blob().decode("utf-8", "ignore") + elif field_no == 2 and wire == 2: + feature_blobs.append(r.blob()) + elif field_no == 3 and wire == 2: + keys.append(r.blob().decode("utf-8", "ignore")) + elif field_no == 4 and wire == 2: + values.append(_decode_value(r.blob())) + elif field_no == 5 and wire == 0: + extent = r.varint() + else: + r.skip(wire) + layer = Layer(name=name, extent=extent) + for blob in feature_blobs: + feat = _decode_feature(blob, keys, values) + if feat is not None: + layer.features.append(feat) + return layer + + +def decode_tile(data: bytes) -> list[Layer]: + """Decode a vector tile (optionally gzip-compressed) into its layers. + + Args: + data: Raw ``.pbf`` bytes, gzip-compressed or not. + + Returns: + The decoded layers, in the order they appear in the tile. An empty or unparseable + tile yields an empty list rather than raising. + """ + if data[:2] == b"\x1f\x8b": + data = gzip.decompress(data) + r = _Reader(data) + layers: list[Layer] = [] + while not r.eof(): + field_no, wire = r.tag() + if field_no == 3 and wire == 2: # Tile.layers + layers.append(_decode_layer(r.blob())) + else: + r.skip(wire) + return layers diff --git a/meshtools/persistence/db.py b/meshtools/persistence/db.py index d8dba3d..b095731 100644 --- a/meshtools/persistence/db.py +++ b/meshtools/persistence/db.py @@ -10,7 +10,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path -SCHEMA_VERSION = 4 +SCHEMA_VERSION = 5 _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_meta ( @@ -83,6 +83,7 @@ CREATE TABLE IF NOT EXISTS observations ( node TEXT, name TEXT, kind TEXT NOT NULL DEFAULT 'advert', + node_type INTEGER, -- advert type (repeater/chat/room/...), when carried snr REAL, rssi REAL, lat REAL, @@ -149,8 +150,8 @@ def _migrate(conn: sqlite3.Connection) -> None: added columns are created here too (not in ``_SCHEMA``), so ``executescript`` never references a column an older database hasn't grown yet. """ - columns = {row["name"] for row in conn.execute("PRAGMA table_info(messages)")} - if "channel_id" not in columns: + message_cols = {row["name"] for row in conn.execute("PRAGMA table_info(messages)")} + if "channel_id" not in message_cols: # v3 -> v4: channel history moved from being keyed by slot index to a # slot-independent channel identity. Existing rows are backfilled lazily at runtime # (see Repository.backfill_channel_ids) once the device's channels can be read. @@ -158,3 +159,9 @@ def _migrate(conn: sqlite3.Connection) -> None: conn.execute( "CREATE INDEX IF NOT EXISTS idx_messages_channel_id ON messages(channel_id)" ) + + observation_cols = {row["name"] for row in conn.execute("PRAGMA table_info(observations)")} + if "node_type" not in observation_cols: + # v4 -> v5: observations gained the transmitting node's advert type, so the map can + # tell repeaters from leaf nodes. Older rows simply carry NULL (type unknown). + conn.execute("ALTER TABLE observations ADD COLUMN node_type INTEGER") diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 13665ae..b31bab9 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -320,13 +320,14 @@ class Repository: """ self._conn.execute( "INSERT INTO observations " - "(run_id, node, name, kind, snr, rssi, lat, lon, observed_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + "(run_id, node, name, kind, node_type, snr, rssi, lat, lon, observed_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( run_id, obs.node, obs.name, obs.kind, + obs.node_type, obs.snr, obs.rssi, obs.lat, @@ -361,7 +362,7 @@ class Repository: Returns: One :class:`HeardNode` per distinct node, ordered by most-recently heard. """ - sql = "SELECT node, name, snr, rssi, lat, lon, observed_at FROM observations" + sql = "SELECT node, name, node_type, snr, rssi, lat, lon, observed_at FROM observations" params: list[Any] = [] if since is not None: sql += " WHERE observed_at >= ?" @@ -373,6 +374,7 @@ class Repository: obs = Observation( node=row["node"], name=row["name"], + node_type=row["node_type"], snr=row["snr"], rssi=row["rssi"], lat=row["lat"], diff --git a/meshtools/services/basemap.py b/meshtools/services/basemap.py new file mode 100644 index 0000000..e70e297 --- /dev/null +++ b/meshtools/services/basemap.py @@ -0,0 +1,182 @@ +"""Fetch and cache OpenStreetMap vector tiles for the terminal map. + +This is the only networked part of the map. It resolves a tile source's (versioned) tile-URL +template from its TileJSON, fetches ``.pbf`` vector tiles over HTTPS, and caches them on disk +so panning back over ground you've seen is instant and later sessions work offline. Decoding +is delegated to the pure :mod:`meshtools.core.mvt`. + +The default source is **OpenFreeMap** (openfreemap.org) — full-planet OpenStreetMap vector +tiles, free and requiring no API key. Everything here is best-effort: with no network and no +cached tiles the loader simply returns ``None`` and the map falls back to plotting nodes on a +blank grid, never raising into the UI. + +Fetches are blocking (stdlib ``urllib``); callers on an event loop should run +:meth:`BasemapSource.load_tile` via ``asyncio.to_thread`` so the UI stays responsive. +""" + +from __future__ import annotations + +import json +import logging +import urllib.request +from pathlib import Path +from typing import Optional + +from ..core.mvt import Layer, decode_tile + +#: OpenFreeMap planet TileJSON — its ``tiles`` array holds the current versioned template. +DEFAULT_TILEJSON_URL = "https://tiles.openfreemap.org/planet" + +#: Sent on every request; identifies the client per common tile-usage etiquette. +_USER_AGENT = "MeshTools/0.1 (+https://github.com/; mesh node map)" + +#: Fallback max tile zoom if the TileJSON doesn't declare one (OpenFreeMap serves 14). +_DEFAULT_MAX_ZOOM = 14 + +_log = logging.getLogger(__name__) + + +class BasemapSource: + """A cached OpenStreetMap vector-tile source. + + Attributes: + cache_dir: Directory under which tiles and the resolved TileJSON are cached. + """ + + def __init__( + self, + cache_dir: Path, + *, + tilejson_url: str = DEFAULT_TILEJSON_URL, + timeout: float = 12.0, + ) -> None: + """Open a tile source backed by an on-disk cache. + + Args: + cache_dir: Where fetched tiles and TileJSON metadata are stored. + tilejson_url: URL of the source's TileJSON document. + timeout: Per-request network timeout in seconds. + """ + self.cache_dir = cache_dir + self._tilejson_url = tilejson_url + self._timeout = timeout + self._template: Optional[str] = None + self._max_zoom: Optional[int] = None + self._resolved = False # whether we've tried (success or offline) this session + + # -- metadata --------------------------------------------------------------- + + def _tilejson_path(self) -> Path: + return self.cache_dir / "tilejson.json" + + def _resolve(self) -> None: + """Resolve and cache the tile-URL template and max zoom (best-effort, once).""" + if self._resolved: + return + self._resolved = True + # Prefer a freshly fetched TileJSON (the template is versioned and rotates), but fall + # back to a previously cached copy so a session started offline can still use disk + # tiles and even re-fetch if the template is still valid. + data = self._http_get(self._tilejson_url) + if data is not None: + try: + self._tilejson_path().parent.mkdir(parents=True, exist_ok=True) + self._tilejson_path().write_bytes(data) + except OSError: # pragma: no cover - cache write failure is non-fatal + pass + if data is None and self._tilejson_path().exists(): + try: + data = self._tilejson_path().read_bytes() + except OSError: # pragma: no cover + data = None + if data is None: + return + try: + doc = json.loads(data) + tiles = doc.get("tiles") or [] + if tiles: + self._template = str(tiles[0]) + self._max_zoom = int(doc.get("maxzoom", _DEFAULT_MAX_ZOOM)) + except (ValueError, TypeError): # pragma: no cover - malformed TileJSON + pass + + @property + def max_zoom(self) -> int: + """The highest zoom the source serves (resolved lazily; sensible default offline).""" + self._resolve() + return self._max_zoom if self._max_zoom is not None else _DEFAULT_MAX_ZOOM + + @property + def available(self) -> bool: + """Whether a tile-URL template is known (network reachable, or one was cached).""" + self._resolve() + return self._template is not None + + # -- tiles ------------------------------------------------------------------ + + def _tile_path(self, z: int, x: int, y: int) -> Path: + return self.cache_dir / "tiles" / str(z) / str(x) / f"{y}.pbf" + + def load_tile(self, z: int, x: int, y: int) -> Optional[list[Layer]]: + """Return the decoded layers for a tile, from cache or the network. + + Args: + z: Tile zoom. + x: Tile x index. + y: Tile y index. + + Returns: + The decoded layers, or ``None`` if the tile is unavailable (offline and + uncached, or a genuinely empty/missing tile). + """ + path = self._tile_path(z, x, y) + raw = self._read_cached(path) + if raw is None: + raw = self._fetch_tile(z, x, y) + if raw is not None: + self._write_cached(path, raw) + if not raw: + return None + try: + return decode_tile(raw) + except Exception as exc: # noqa: BLE001 - a corrupt tile must not crash the map + _log.debug("failed to decode tile %s/%s/%s: %s", z, x, y, exc) + return None + + def _fetch_tile(self, z: int, x: int, y: int) -> Optional[bytes]: + """Fetch a tile's raw bytes from the network, or ``None`` if unavailable.""" + self._resolve() + if self._template is None: + return None + url = ( + self._template.replace("{z}", str(z)).replace("{x}", str(x)).replace("{y}", str(y)) + ) + data = self._http_get(url) + # A 404/empty response is a legitimately empty tile; cache an empty marker so we + # don't re-request it every repaint. + return data if data is not None else b"" + + @staticmethod + def _read_cached(path: Path) -> Optional[bytes]: + try: + return path.read_bytes() if path.exists() else None + except OSError: # pragma: no cover + return None + + @staticmethod + def _write_cached(path: Path, raw: bytes) -> None: + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(raw) + except OSError: # pragma: no cover - cache write failure is non-fatal + pass + + def _http_get(self, url: str) -> Optional[bytes]: + """GET a URL, returning the body, or ``None`` on any network/HTTP failure.""" + req = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT}) + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + return resp.read() + except Exception as exc: # noqa: BLE001 - offline / 404 / timeout are all "no tile" + _log.debug("tile fetch failed for %s: %s", url, exc) + return None diff --git a/meshtools/tools/map.py b/meshtools/tools/map.py new file mode 100644 index 0000000..18ce27b --- /dev/null +++ b/meshtools/tools/map.py @@ -0,0 +1,227 @@ +"""The ``map`` tool: plot the mesh's located nodes on an OpenStreetMap terminal map. + +Nodes that share their location in adverts are collected from the passive-monitor history +(:meth:`~meshtools.persistence.repository.Repository.heard_nodes`), our own node is added when +its position is known, and everything is drawn over a real street basemap rendered as Unicode +braille (streets, rivers, place names — the same terminal-map idea as ``mapscii``). + +In the menu it opens a full-screen, pannable/zoomable map (see +:mod:`meshtools.ui.map_screen`). On the CLI it prints a one-shot render fitted to the nodes. +Repeaters are prioritised over ordinary nodes: a distinct marker, drawn on top, listed first. +The basemap is best-effort — with no network (and no cached tiles) the nodes are plotted on a +blank grid instead, and the tool still works offline. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Optional + +import typer +from rich.console import Group +from rich.table import Table +from rich.text import Text + +from ..context import AppContext +from ..core.models import NODE_TYPE_REPEATER +from .base import Tool, ToolResult, register + +if TYPE_CHECKING: + from ..ui.map_render import MapMarker + + +@register +class MapTool(Tool): + """Show the mesh's location-sharing nodes on a braille OpenStreetMap map.""" + + name = "map" + help = "Show mesh nodes on a street map (pannable; repeaters highlighted)." + category = "Messaging" + order = 30 + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Gather located nodes and open the interactive map, or render a static one. + + Args: + ctx: Shared application context. + params: Optional ``static`` (force one-shot render), ``width``, ``zoom``, + ``basemap`` (set ``False`` to skip fetching tiles). + + Returns: + A :class:`ToolResult` summarizing how many nodes were plotted. + """ + from ..ui.surface import TuiUi + + markers = await self._gather(ctx) + if not markers: + ctx.ui.note( + "[muted]no nodes have shared a location yet — turn on the passive " + "monitor and wait for located adverts[/muted]" + ) + return ToolResult(summary={"located": 0}) + + interactive = isinstance(ctx.ui, TuiUi) and not params.get("static") + if interactive: + from ..ui.map_screen import open_map + + await open_map(ctx, markers) + else: + await self._render_static(ctx, markers, params) + + repeaters = sum(1 for m in markers if m.is_repeater and not m.is_self) + self_located = any(m.is_self for m in markers) + return ToolResult( + summary={ + "located": len(markers), + "repeaters": repeaters, + "nodes": len(markers) - repeaters - (1 if self_located else 0), + "self_located": self_located, + }, + message=None if interactive else ( + f"[ok]✓[/ok] mapped [brand]{len(markers)}[/brand] located nodes " + f"([accent]{repeaters}[/accent] repeaters)" + ), + ) + + # -- marker gathering ------------------------------------------------------- + + async def _gather(self, ctx: AppContext) -> list["MapMarker"]: + """Collect located heard nodes plus our own node (when its position is known).""" + from ..ui.map_render import MapMarker + + markers: list[MapMarker] = [] + for node in ctx.repo.heard_nodes(): + if not node.has_location: + continue + detail = f"{node.count} pkts" + if node.median_snr is not None: + detail += f" · {node.median_snr:+.1f} dB" + markers.append( + MapMarker( + label=node.name or node.node or "?", + lat=float(node.lat), + lon=float(node.lon), + is_repeater=node.is_repeater, + detail=detail, + ) + ) + self_marker = await self._self_marker(ctx) + if self_marker is not None: + markers.append(self_marker) + return markers + + @staticmethod + async def _self_marker(ctx: AppContext) -> Optional["MapMarker"]: + """Build a marker for our own node from the device, if its location is known.""" + from ..ui.map_render import MapMarker + + try: + info = await (await ctx.device()).get_self_info() + except Exception: # noqa: BLE001 - the map is useful without our own position + return None + lat, lon = _as_float(info.get("adv_lat")), _as_float(info.get("adv_lon")) + if lat is None or lon is None or (abs(lat) < 1e-6 and abs(lon) < 1e-6): + return None # a device with no fix reports 0/0 + return MapMarker( + label=str(info.get("name") or "this node"), + lat=lat, + lon=lon, + is_repeater=info.get("adv_type") == NODE_TYPE_REPEATER, + is_self=True, + ) + + # -- static (CLI) render ---------------------------------------------------- + + async def _render_static( + self, ctx: AppContext, markers: list["MapMarker"], params: dict[str, Any] + ) -> None: + """Fetch tiles synchronously and print a one-shot map fitted to the nodes.""" + import asyncio + + from ..core.geo import Viewport + from ..ui.map_render import render_map + from ..ui.map_screen import basemap_source + + cell_w = int(params.get("width") or min(ctx.console.size.width - 2, 160)) + cell_h = max(12, cell_w * 4 // 13) # keep roughly the terminal's aspect + dot_w, dot_h = cell_w * 2, cell_h * 4 + + source = basemap_source(ctx) + max_zoom = await asyncio.to_thread(lambda: source.max_zoom) + coords = [(m.lat, m.lon) for m in markers] + if params.get("zoom") is not None: + from ..core.geo import BBox + + center = BBox.around(coords).center + vp = Viewport(center[0], center[1], int(params["zoom"]), dot_w, dot_h) + else: + vp = Viewport.fit(coords, dot_w, dot_h, max_zoom=max_zoom) + + tiles = {} + if params.get("basemap", True): + for tile in vp.tiles(max_zoom): + tiles[tile] = await asyncio.to_thread(source.load_tile, *tile) + + lines = render_map(vp, tiles, markers) + body = Text.from_ansi("\n".join(lines)) + ctx.ui.show(Group(body, Text(""), _legend(markers))) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``map`` subcommand. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + @app.command(name=self.name, help=self.help) + def _map( + width: Optional[int] = typer.Option( + None, "--width", "-w", help="Map width in character cells." + ), + zoom: Optional[int] = typer.Option( + None, "--zoom", "-z", help="Fixed zoom level (omit to fit all nodes)." + ), + basemap: bool = typer.Option( + True, "--basemap/--no-basemap", help="Draw the OpenStreetMap street basemap." + ), + ) -> None: + params: dict[str, Any] = {"static": True, "basemap": basemap} + if width is not None: + params["width"] = width + if zoom is not None: + params["zoom"] = zoom + run_tool_command(self, params) + + +def _legend(markers: list["MapMarker"]) -> Table: + """A compact legend: self → repeaters → leaf nodes, with coordinates and detail.""" + from ..ui.map_render import _NODE, _REPEATER, _SELF + + table = Table(box=None, padding=(0, 2, 0, 0), expand=False) + table.add_column("") + table.add_column("node") + table.add_column("type") + table.add_column("coords", justify="right") + table.add_column("seen") + ordered = sorted(markers, key=lambda m: -m._rank()) + for m in ordered: + glyph, color = _SELF if m.is_self else (_REPEATER if m.is_repeater else _NODE) + kind = "you" if m.is_self else ("repeater" if m.is_repeater else "node") + table.add_row( + Text(glyph, style=color), + Text(m.label, style=color if m.is_self else ""), + Text(kind, style="muted"), + Text(f"{m.lat:.4f}, {m.lon:.4f}", style="muted"), + Text(m.detail, style="muted"), + ) + return table + + +def _as_float(value: object) -> Optional[float]: + """Best-effort float conversion, returning ``None`` on missing/garbage values.""" + if value is None: + return None + try: + return float(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None diff --git a/meshtools/ui/map_render.py b/meshtools/ui/map_render.py new file mode 100644 index 0000000..7fe5a07 --- /dev/null +++ b/meshtools/ui/map_render.py @@ -0,0 +1,313 @@ +"""Compose a rendered map frame: basemap geometry, labels, and mesh nodes. + +Given a :class:`~meshtools.core.geo.Viewport` and the decoded vector tiles covering it, this +projects every street, river, water body and place label onto a :class:`MapCanvas`, then +overlays the mesh nodes with their names on top. It is pure and synchronous — tile *fetching* +happens elsewhere (:mod:`meshtools.services.basemap`) — so both the interactive screen and +the one-shot CLI render call the same code. + +Colours target a dark terminal (the app theme): warm roads, grey minor streets, blue water +and rivers, faint green parks. Because a cell shows one colour, draw priorities keep the +important feature visible where things overlap (rivers over water, major roads over minor). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Optional + +from ..core.geo import Viewport +from ..core.mvt import GEOM_LINE, GEOM_POLYGON, Layer +from .mapcanvas import RGB, MapCanvas, parse_hex + +# -- node markers ------------------------------------------------------------- + + +@dataclass(slots=True) +class MapMarker: + """One mesh node to overlay on the map. + + Attributes: + label: Node name shown beside the marker. + lat: Latitude in decimal degrees. + lon: Longitude in decimal degrees. + is_repeater: Whether the node is a repeater (prioritised marker). + is_self: Whether this is our own node (highlighted). + detail: Extra text for the CLI legend (e.g. ``"18 pkts · +6.0 dB"``). + """ + + label: str + lat: float + lon: float + is_repeater: bool = False + is_self: bool = False + detail: str = "" + + def _rank(self) -> int: + """Draw order: self on top of repeaters on top of leaf nodes.""" + return 2 if self.is_self else (1 if self.is_repeater else 0) + + +_SELF = ("★", "#ffffff") +_REPEATER = ("▲", "#ff5d73") +_NODE = ("●", "#22d3ee") + + +# -- basemap styling ---------------------------------------------------------- + +# Road class -> (colour, priority). Higher priority wins a shared cell. +_ROAD_STYLE: dict[str, tuple[str, int]] = { + "motorway": ("#f2a13d", 27), + "trunk": ("#f2a13d", 26), + "primary": ("#e8b24a", 25), + "secondary": ("#d7c257", 24), + "tertiary": ("#a7adb8", 23), + "minor": ("#828894", 21), + "unclassified": ("#828894", 21), + "residential": ("#828894", 21), + "living_street": ("#767c88", 20), + "service": ("#5f646e", 19), + "pedestrian": ("#5a5f69", 15), + "path": ("#4b5560", 12), + "track": ("#4b5560", 12), + "footway": ("#454e58", 11), + "cycleway": ("#455864", 12), +} +_ROAD_DEFAULT = ("#5f646e", 18) +_RAIL = ("#8a8f98", 22) + +# Waterway class -> (colour, priority). +_WATERWAY_STYLE: dict[str, tuple[str, int]] = { + "river": ("#49b0ec", 30), + "canal": ("#49b0ec", 30), + "stream": ("#3f8fbf", 29), + "ditch": ("#3a7ba6", 28), + "drain": ("#3a7ba6", 28), +} +_WATER_FILL = ("#153b56", 6) +_GREEN_FILL = ("#173a29", 4) +_GREEN_CLASSES = {"wood", "forest", "grass", "park", "meadow", "scrub", "wetland", "cemetery"} + +# Label styling per place class: (colour, bold, rank, min_display_zoom). Lower rank places +# claim screen space first; min zoom hides the fine detail until you zoom in. +_PLACE_STYLE: dict[str, tuple[str, bool, int, int]] = { + "city": ("#f8fafc", True, 0, 0), + "town": ("#e6ebf2", True, 1, 9), + "village": ("#cdd6e2", False, 2, 12), + "suburb": ("#9fb0c4", False, 3, 12), + "quarter": ("#94a5ba", False, 4, 13), + "neighbourhood": ("#8496ab", False, 5, 13), + "hamlet": ("#8496ab", False, 6, 14), +} +_WATER_LABEL = ("#7dd3fc", False, 2) +_STREET_LABEL = ("#9aa0aa", False, 7) + + +@dataclass(slots=True) +class _Label: + """A pending label placement candidate.""" + + rank: int + x: float + y: float + text: str + color: RGB + bold: bool + min_zoom: int = 0 + + +@dataclass(slots=True) +class _Frame: + """Working state while composing one frame.""" + + canvas: MapCanvas + viewport: Viewport + labels: list[_Label] = field(default_factory=list) + + +def render_map( + viewport: Viewport, + tiles: dict[tuple[int, int, int], Optional[list[Layer]]], + markers: list[MapMarker], + *, + max_labels: int = 80, +) -> list[str]: + """Render a full map frame to truecolour ANSI lines. + + Args: + viewport: The view to render (its dot size sets the canvas size). + tiles: Decoded layers keyed by ``(z, x, y)``; ``None`` values are pending/absent. + markers: Mesh nodes to overlay. + max_labels: Cap on basemap labels placed, to keep the map readable. + + Returns: + One ANSI string per row, ready for the TUI frame or the console. + """ + canvas = MapCanvas(viewport.dot_w // 2, viewport.dot_h // 4) + frame = _Frame(canvas=canvas, viewport=viewport) + + for (z, x, y), layers in tiles.items(): + if layers: + _draw_tile(frame, layers, z, x, y) + + # Nodes reserve their cells first so basemap labels route around them. + _draw_nodes(canvas, viewport, markers) + + # Then place basemap labels by importance, honouring the zoom gate and collisions. + placed = 0 + for label in sorted(frame.labels, key=lambda l: l.rank): + if placed >= max_labels: + break + if viewport.zoom < label.min_zoom: + continue + if canvas.place_label(label.x, label.y, label.text, label.color, bold=label.bold): + placed += 1 + + return canvas.to_ansi_lines() + + +def _draw_tile(frame: _Frame, layers: list[Layer], z: int, x: int, y: int) -> None: + """Draw one decoded tile's geometry and collect its label candidates.""" + vp = frame.canvas + by_name = {layer.name: layer for layer in layers} + + def project(ring: list[tuple[int, int]], extent: int) -> list[tuple[float, float]]: + return [ + frame.viewport.feature_to_dot(x, y, z, extent, lx, ly) for lx, ly in ring + ] + + # Fills first (water, green space) so lines and labels sit on top. + for name in ("water", "landcover", "landuse", "park"): + layer = by_name.get(name) + if layer is None: + continue + for feat in layer.features: + if feat.geom_type != GEOM_POLYGON: + continue + if name == "water": + color, prio = _WATER_FILL + elif str(feat.get("class") or feat.get("subclass")) in _GREEN_CLASSES: + color, prio = _GREEN_FILL + else: + continue + vp.fill_polygon( + [project(r, layer.extent) for r in feat.rings], parse_hex(color), prio + ) + + # Waterways (rivers/streams) as lines. + waterway = by_name.get("waterway") + if waterway is not None: + for feat in waterway.features: + style = _WATERWAY_STYLE.get(str(feat.get("class"))) + if style is None or feat.geom_type != GEOM_LINE: + continue + color, prio = style + for ring in feat.rings: + if len(ring) >= 2: + vp.draw_line(project(ring, waterway.extent), parse_hex(color), prio) + if feat.name: + _add_line_label(frame, feat.rings, waterway.extent, z, x, y, feat.name, _WATER_LABEL) + + # Roads / rail. + transportation = by_name.get("transportation") + if transportation is not None: + for feat in transportation.features: + if feat.geom_type != GEOM_LINE: + continue + cls = str(feat.get("class") or "") + if cls in ("rail", "transit"): + color, prio = _RAIL + else: + color, prio = _ROAD_STYLE.get(cls, _ROAD_DEFAULT) + rgb = parse_hex(color) + for ring in feat.rings: + if len(ring) >= 2: + vp.draw_line(project(ring, transportation.extent), rgb, prio) + + # Boundaries (admin) as a faint hint. + boundary = by_name.get("boundary") + if boundary is not None: + for feat in boundary.features: + if feat.geom_type != GEOM_LINE: + continue + try: + if int(feat.get("admin_level", 99)) > 6: + continue + except (TypeError, ValueError): + continue + for ring in feat.rings: + if len(ring) >= 2: + vp.draw_line(project(ring, boundary.extent), parse_hex("#6d5f88"), 14) + + # Street-name labels (only kick in at high zoom via the label's min_zoom gate). + tname = by_name.get("transportation_name") + if tname is not None: + color, bold, rank = _STREET_LABEL + for feat in tname.features: + if feat.name and feat.rings: + _add_line_label( + frame, feat.rings, tname.extent, z, x, y, feat.name, + (color, bold, rank), min_zoom=15, + ) + + # Place labels. + place = by_name.get("place") + if place is not None: + for feat in place.features: + if not feat.name or not feat.rings or not feat.rings[0]: + continue + style = _PLACE_STYLE.get(str(feat.get("class"))) + if style is None: + continue + color, bold, rank, min_zoom = style + lx, ly = feat.rings[0][0] + dx, dy = frame.viewport.feature_to_dot(x, y, z, place.extent, lx, ly) + frame.labels.append( + _Label(rank, dx, dy, feat.name, parse_hex(color), bold, min_zoom) + ) + + # Water-body names. + water_name = by_name.get("water_name") + if water_name is not None: + color, bold, rank = _WATER_LABEL + for feat in water_name.features: + if feat.name and feat.rings and feat.rings[0]: + lx, ly = feat.rings[0][0] + dx, dy = frame.viewport.feature_to_dot(x, y, z, water_name.extent, lx, ly) + frame.labels.append(_Label(rank, dx, dy, feat.name, parse_hex(color), bold, 8)) + + +def _add_line_label( + frame: _Frame, + rings: list[list[tuple[int, int]]], + extent: int, + z: int, + x: int, + y: int, + text: str, + style: tuple[str, bool, int], + *, + min_zoom: int = 0, +) -> None: + """Queue a label at the midpoint of a line feature's longest part.""" + longest = max(rings, key=len) + lx, ly = longest[len(longest) // 2] + dx, dy = frame.viewport.feature_to_dot(x, y, z, extent, lx, ly) + color, bold, rank = style + frame.labels.append(_Label(rank, dx, dy, text, parse_hex(color), bold, min_zoom)) + + +def _draw_nodes(canvas: MapCanvas, viewport: Viewport, markers: list[MapMarker]) -> None: + """Overlay mesh nodes (lowest priority first, so self/repeaters land on top).""" + for marker in sorted(markers, key=lambda m: m._rank()): + x, y = viewport.lonlat_to_dot(marker.lat, marker.lon) + ix, iy = int(round(x)), int(round(y)) + if not (0 <= ix < viewport.dot_w and 0 <= iy < viewport.dot_h): + continue + if marker.is_self: + glyph, color = _SELF + elif marker.is_repeater: + glyph, color = _REPEATER + else: + glyph, color = _NODE + canvas.marker(ix, iy, glyph, parse_hex(color), label=marker.label) diff --git a/meshtools/ui/map_screen.py b/meshtools/ui/map_screen.py new file mode 100644 index 0000000..1e8f784 --- /dev/null +++ b/meshtools/ui/map_screen.py @@ -0,0 +1,217 @@ +"""The interactive full-screen map: a pannable, zoomable slippy map in the terminal. + +This drives the braille street map inside the TUI. It owns a :class:`~meshtools.core.geo. +Viewport` over the mesh's nodes, fetches the vector tiles covering it in the background (so +the UI never blocks on the network), and redraws via :func:`~meshtools.ui.map_render. +render_map`. Keys: + +* ``w`` / ``a`` / ``s`` / ``d`` (or the arrow keys) pan north / west / south / east, +* ``=`` / ``+`` zoom in, ``-`` / ``_`` zoom out, +* ``r`` recenters and refits to the nodes, +* ``Esc`` / ``q`` leaves the map. + +With no network (and no cached tiles) the basemap is simply absent and nodes are plotted on a +blank grid — the map still works, it just has no streets. +""" + +from __future__ import annotations + +import asyncio +import math +from typing import TYPE_CHECKING, Optional + +from ..core.geo import EARTH_RADIUS_KM, Viewport +from ..core.mvt import Layer +from ..services.basemap import BasemapSource +from .map_render import MapMarker, render_map +from .tui.screen import Screen + +if TYPE_CHECKING: + from ..context import AppContext + +#: Fraction of the view a single pan keypress moves. +_PAN_STEP = 0.30 + +#: How far past the tile source's max zoom the display may go (lower tiles are magnified). +_OVERZOOM = 2 + + +class MapScreen(Screen): + """A full-screen, keyboard-driven map of the mesh's located nodes over an OSM basemap.""" + + floating = False + + def __init__( + self, + session, # noqa: ANN001 - TuiSession, imported lazily to avoid a cycle + markers: list[MapMarker], + source: BasemapSource, + max_tile_zoom: int, + ) -> None: + """Create the map screen. + + Args: + session: The running TUI session (for size + repaint scheduling). + markers: The located mesh nodes to plot (must be non-empty). + source: The vector-tile source (already resolved/warmed). + max_tile_zoom: The source's max zoom, captured off the event loop at open time. + """ + super().__init__() + self.title = "mesh map" + self._session = session + self._markers = markers + self._source = source + self._max_tile_zoom = max_tile_zoom + self._viewport: Optional[Viewport] = None + self._size: tuple[int, int] = (0, 0) # (dot_w, dot_h) the viewport is built for + # Decoded tiles keyed by (z, x, y); a stored ``None`` means "fetched, empty/absent". + self._tiles: dict[tuple[int, int, int], Optional[list[Layer]]] = {} + self._pending: set[tuple[int, int, int]] = set() + + # --- rendering ----------------------------------------------------------- + + @property + def footer_hint(self) -> str: # type: ignore[override] + """Key hints plus a live tile-loading indicator.""" + base = "wasd/↑↓←→ pan · +/- zoom · r reset · Esc back" + if self._pending: + return f"{base} · [muted]loading {len(self._pending)} tiles…[/muted]" + if not self._source.available: + return f"{base} · [warn]offline — no basemap[/warn]" + return base + + def render_body(self, width: int) -> list[str]: + """Build (or resize) the viewport, ensure its tiles, and render the frame.""" + _, cell_h = self._session.base_body_size() + cell_w = width + dot_w, dot_h = cell_w * 2, cell_h * 4 + + if self._viewport is None: + self._viewport = Viewport.fit( + [(m.lat, m.lon) for m in self._markers], + dot_w, + dot_h, + max_zoom=self._max_tile_zoom, + ) + self._size = (dot_w, dot_h) + elif self._size != (dot_w, dot_h): + self._viewport = self._viewport.resized(dot_w, dot_h) + self._size = (dot_w, dot_h) + + self._ensure_tiles(self._viewport) + self.title = self._title(self._viewport) + tiles = {t: self._tiles.get(t) for t in self._viewport.tiles(self._max_tile_zoom)} + return render_map(self._viewport, tiles, self._markers) + + def _title(self, vp: Viewport) -> str: + """A compact status title: zoom, node count, and scale (metres per dot).""" + # Ground metres per braille dot at the view centre, for a rough sense of scale. + m_per_dot = ( + 2 * math.pi * EARTH_RADIUS_KM * 1000 + * math.cos(math.radians(vp.center_lat)) + / (256 * (2**vp.zoom)) + ) + scale = f"{m_per_dot * vp.dot_w:.0f} m across" if m_per_dot * vp.dot_w < 1000 else \ + f"{m_per_dot * vp.dot_w / 1000:.1f} km across" + return f"mesh map · z{vp.zoom} · {len(self._markers)} nodes · {scale}" + + # --- tiles --------------------------------------------------------------- + + def _ensure_tiles(self, vp: Viewport) -> None: + """Schedule background fetches for any visible tiles not yet loaded or pending.""" + if not self._source.available: # offline (resolved at open time) — nodes only + return + for t in vp.tiles(self._max_tile_zoom): + if t in self._tiles or t in self._pending: + continue + self._pending.add(t) + try: + asyncio.ensure_future(self._load(t)) + except RuntimeError: # pragma: no cover - no running loop (non-interactive) + self._pending.discard(t) + + async def _load(self, t: tuple[int, int, int]) -> None: + """Fetch+decode one tile off the event loop, then repaint.""" + try: + layers = await asyncio.to_thread(self._source.load_tile, *t) + except Exception: # noqa: BLE001 - a failed tile is just an absent one + layers = None + self._tiles[t] = layers + self._pending.discard(t) + self._session.invalidate() + + # --- input --------------------------------------------------------------- + + def handle(self, action: str, data: str = "") -> None: + """Pan, zoom, reset, or exit in response to a normalized key action.""" + vp = self._viewport + if action == "escape": + self.resolve(None) + return + if vp is None: + return + if action == "up": + self._viewport = vp.panned(0, -_PAN_STEP) + elif action == "down": + self._viewport = vp.panned(0, _PAN_STEP) + elif action == "left": + self._viewport = vp.panned(-_PAN_STEP, 0) + elif action == "right": + self._viewport = vp.panned(_PAN_STEP, 0) + elif action == "text": + self._handle_key(data, vp) + + def _handle_key(self, key: str, vp: Viewport) -> None: + """Handle a printable-key action (pan/zoom/reset/quit).""" + key = key.lower() + if key == "w": + self._viewport = vp.panned(0, -_PAN_STEP) + elif key == "s": + self._viewport = vp.panned(0, _PAN_STEP) + elif key == "a": + self._viewport = vp.panned(-_PAN_STEP, 0) + elif key == "d": + self._viewport = vp.panned(_PAN_STEP, 0) + elif key in ("=", "+"): + self._viewport = vp.zoomed(1, max_zoom=self._max_tile_zoom + _OVERZOOM) + elif key in ("-", "_"): + self._viewport = vp.zoomed(-1) + elif key == "r": + self._viewport = Viewport.fit( + [(m.lat, m.lon) for m in self._markers], + vp.dot_w, + vp.dot_h, + max_zoom=self._max_tile_zoom, + ) + elif key == "q": + self.resolve(None) + + +def basemap_source(ctx: "AppContext") -> BasemapSource: + """Build the vector-tile source backed by the app's on-disk tile cache.""" + return BasemapSource(ctx.settings.config_dir / "tilecache") + + +async def open_map(ctx: "AppContext", markers: list[MapMarker]) -> None: + """Open the interactive full-screen map over ``markers`` and run until dismissed. + + Warms the tile source off the event loop (so the first paint doesn't block on the + network), then pushes the :class:`MapScreen` and awaits its dismissal. + + Args: + ctx: Shared application context (must be in the interactive menu). + markers: The located mesh nodes to plot (non-empty). + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + from .surface import TuiUi + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("the interactive map is only available in the menu") + session = ctx.ui.session + source = basemap_source(ctx) + # Resolve the tile template/zoom in a worker thread so the UI thread never blocks. + max_zoom = await asyncio.to_thread(lambda: source.max_zoom) + screen = MapScreen(session, markers, source, max_zoom) + await session.run_screen(screen) diff --git a/meshtools/ui/mapcanvas.py b/meshtools/ui/mapcanvas.py new file mode 100644 index 0000000..6bcbf32 --- /dev/null +++ b/meshtools/ui/mapcanvas.py @@ -0,0 +1,252 @@ +"""A colored Unicode-braille canvas for the terminal map. + +Each character cell holds a 2×4 grid of braille dots, so the drawable resolution is twice +the columns by four times the rows. Streets, rivers and water are rasterized as braille +*dots*; place names, street names and node labels are written as *text* over whole cells; +node markers are single glyphs on top. + +A terminal cell can only show one colour, so each cell keeps the colour of the +highest-**priority** feature whose dots fall in it (a river drawn over water keeps the river's +blue). Text and markers form a separate overlay that always wins the cell, with greedy +collision avoidance so labels never overprint each other. The canvas renders straight to +truecolour ANSI lines, which is exactly what the TUI frame consumes. +""" + +from __future__ import annotations + +from typing import Optional + +#: Unicode braille pattern base; add a dot bitmask to get the glyph. +_BRAILLE_BASE = 0x2800 + +#: Dot bit for each (col, row) within a cell — the Unicode braille standard layout. +_DOT_BITS = ( + (0x01, 0x02, 0x04, 0x40), # left column, rows 0..3 + (0x08, 0x10, 0x20, 0x80), # right column, rows 0..3 +) + +RGB = tuple[int, int, int] + + +def parse_hex(color: str) -> RGB: + """Convert ``"#rrggbb"`` (or ``"rrggbb"``) to an ``(r, g, b)`` tuple.""" + c = color.lstrip("#") + return (int(c[0:2], 16), int(c[2:4], 16), int(c[4:6], 16)) + + +class MapCanvas: + """A colored braille raster with a text/marker overlay and label collision tracking.""" + + def __init__(self, cell_w: int, cell_h: int) -> None: + """Create a blank canvas. + + Args: + cell_w: Width in character cells. + cell_h: Height in character cells. + """ + self.cell_w = max(1, cell_w) + self.cell_h = max(1, cell_h) + self.dot_w = self.cell_w * 2 + self.dot_h = self.cell_h * 4 + self._bits = [[0] * self.cell_w for _ in range(self.cell_h)] + self._color: list[list[Optional[RGB]]] = [ + [None] * self.cell_w for _ in range(self.cell_h) + ] + self._prio = [[-1] * self.cell_w for _ in range(self.cell_h)] + # Overlay: (cx, cy) -> (char, rgb, bold). Occupied tracks cells claimed by labels / + # markers so later labels can avoid them. + self._overlay: dict[tuple[int, int], tuple[str, RGB, bool]] = {} + self._occupied: set[tuple[int, int]] = set() + + # -- primitives ------------------------------------------------------------- + + def plot(self, x: int, y: int, color: RGB, priority: int) -> None: + """Light the braille dot at dot ``(x, y)`` and colour its cell by ``priority``.""" + if not (0 <= x < self.dot_w and 0 <= y < self.dot_h): + return + cx, cy = x >> 1, y >> 2 + self._bits[cy][cx] |= _DOT_BITS[x & 1][y & 3] + if priority >= self._prio[cy][cx]: + self._prio[cy][cx] = priority + self._color[cy][cx] = color + + def draw_line( + self, points: list[tuple[float, float]], color: RGB, priority: int + ) -> None: + """Rasterize a polyline through ``points`` (dot coordinates) as braille dots.""" + for (x0, y0), (x1, y1) in zip(points, points[1:]): + self._segment(x0, y0, x1, y1, color, priority) + + def _segment( + self, x0: float, y0: float, x1: float, y1: float, color: RGB, priority: int + ) -> None: + """Bresenham a single segment, skipping ones wholly off the canvas.""" + # Quick reject: both endpoints beyond the same edge → nothing visible. + if (x0 < 0 and x1 < 0) or (x0 >= self.dot_w and x1 >= self.dot_w): + return + if (y0 < 0 and y1 < 0) or (y0 >= self.dot_h and y1 >= self.dot_h): + return + xa, ya, xb, yb = int(round(x0)), int(round(y0)), int(round(x1)), int(round(y1)) + dx, dy = abs(xb - xa), -abs(yb - ya) + sx = 1 if xa < xb else -1 + sy = 1 if ya < yb else -1 + err = dx + dy + while True: + self.plot(xa, ya, color, priority) + if xa == xb and ya == yb: + break + e2 = 2 * err + if e2 >= dy: + err += dy + xa += sx + if e2 <= dx: + err += dx + ya += sy + + def fill_polygon( + self, rings: list[list[tuple[float, float]]], color: RGB, priority: int + ) -> None: + """Even-odd scanline fill of a polygon (with holes) in dot space.""" + edges: list[tuple[float, float, float, float]] = [] + ys: list[float] = [] + for ring in rings: + for (x0, y0), (x1, y1) in zip(ring, ring[1:]): + if y0 != y1: + edges.append((x0, y0, x1, y1)) + ys.extend((y0, y1)) + if not edges: + return + y_start = max(0, int(min(ys))) + y_end = min(self.dot_h - 1, int(max(ys))) + for y in range(y_start, y_end + 1): + yc = y + 0.5 + xs: list[float] = [] + for x0, y0, x1, y1 in edges: + if (y0 <= yc < y1) or (y1 <= yc < y0): + xs.append(x0 + (yc - y0) * (x1 - x0) / (y1 - y0)) + xs.sort() + for i in range(0, len(xs) - 1, 2): + x_from = max(0, int(round(xs[i]))) + x_to = min(self.dot_w - 1, int(round(xs[i + 1]))) + for x in range(x_from, x_to + 1): + self.plot(x, y, color, priority) + + # -- overlay (markers + labels) -------------------------------------------- + + def marker( + self, + x: int, + y: int, + glyph: str, + color: RGB, + *, + label: str = "", + label_color: Optional[RGB] = None, + ) -> None: + """Place a marker glyph at dot ``(x, y)`` with an optional label beside it. + + Markers always draw (they are the point of the map). The label is placed to the + right when there's room, otherwise to the left, and its cells are reserved so + basemap labels avoid them. + """ + cx, cy = x >> 1, y >> 2 + if not (0 <= cx < self.cell_w and 0 <= cy < self.cell_h): + return + self._overlay[(cx, cy)] = (glyph, color, True) + self._occupied.add((cx, cy)) + if label: + lc = label_color or color + # Prefer the label to the right of the marker (one blank cell gap); fall back to + # the left when it would overflow — never overwriting the marker glyph itself. + if cx + 2 + len(label) <= self.cell_w: + self._place_run(cx + 2, cy, label, lc, bold=True) + else: + self._place_run(max(0, cx - 1 - len(label)), cy, label, lc, bold=True) + + def place_label( + self, x: float, y: float, text: str, color: RGB, *, bold: bool = False + ) -> bool: + """Write a centered basemap label at dot ``(x, y)`` if it fits without collision. + + Args: + x: Anchor dot x. + y: Anchor dot y. + text: Label text. + color: Text colour. + bold: Whether to embolden (used for the most important places). + + Returns: + ``True`` if placed, ``False`` if it fell off-canvas or overlapped existing text. + """ + cx = int(x) >> 1 + cy = int(y) >> 2 + start = cx - len(text) // 2 + return self._place_run(start, cy, text, color, bold=bold, checked=True) + + def _place_run( + self, + start_cx: int, + cy: int, + text: str, + color: RGB, + *, + bold: bool, + checked: bool = False, + ) -> bool: + """Write ``text`` starting at cell ``(start_cx, cy)``. + + With ``checked`` the run is skipped entirely if any target cell (or a one-cell + horizontal margin) is already claimed or off-canvas; otherwise it is forced and + simply clipped to the canvas. Returns whether anything was placed. + """ + if not (0 <= cy < self.cell_h): + return False + cells = range(start_cx, start_cx + len(text)) + if checked: + if start_cx < 0 or start_cx + len(text) > self.cell_w: + return False + margin = range(start_cx - 1, start_cx + len(text) + 1) + if any((mx, cy) in self._occupied for mx in margin): + return False + placed = False + for offset, ch in enumerate(text): + mx = start_cx + offset + if 0 <= mx < self.cell_w: + self._overlay[(mx, cy)] = (ch, color, bold) + self._occupied.add((mx, cy)) + placed = True + return placed + + # -- output ----------------------------------------------------------------- + + def to_ansi_lines(self) -> list[str]: + """Render the canvas to one truecolour ANSI string per row.""" + reset = "\x1b[0m" + lines: list[str] = [] + for cy in range(self.cell_h): + parts: list[str] = [] + cur: Optional[tuple[RGB, bool]] = None + for cx in range(self.cell_w): + overlay = self._overlay.get((cx, cy)) + if overlay is not None: + ch, color, bold = overlay + elif self._bits[cy][cx]: + ch = chr(_BRAILLE_BASE + self._bits[cy][cx]) + color = self._color[cy][cx] or (128, 128, 128) + bold = False + else: + if cur is not None: + parts.append(reset) + cur = None + parts.append(" ") + continue + style = (color, bold) + if style != cur: + r, g, b = color + parts.append(f"\x1b[0m\x1b[38;2;{r};{g};{b}m" + ("\x1b[1m" if bold else "")) + cur = style + parts.append(ch) + if cur is not None: + parts.append(reset) + lines.append("".join(parts)) + return lines diff --git a/meshtools/ui/tui/session.py b/meshtools/ui/tui/session.py index 9d21a34..cd3777a 100644 --- a/meshtools/ui/tui/session.py +++ b/meshtools/ui/tui/session.py @@ -251,6 +251,23 @@ class TuiSession: size = self._app.output.get_size() # type: ignore[union-attr] return max(20, size.columns), max(6, size.rows) + def base_body_size(self) -> tuple[int, int]: + """Return the ``(width, height)`` in cells available to the base screen's body. + + Mirrors the layout math in :func:`~meshtools.ui.tui.frame.compose_base` so a + full-screen screen (e.g. the map) can size its own content to fill the frame exactly, + without waiting a repaint to learn its height. + + Returns: + The inner content width and the body viewport height, both in character cells. + """ + from .render import render_lines + + cols, rows = self._size() + header_h = len(render_lines(self._header(), cols, no_wrap=True)) + viewport = max(1, rows - header_h - 1 - 2) # minus footer(1) and panel border(2) + return cols - 4, viewport + def _base_screen(self) -> Optional[Screen]: """The screen drawn as the background (parent of a floating dialog, else the top).""" if not self._stack: diff --git a/tests/fixtures/tile_14_4843_5861.mvt b/tests/fixtures/tile_14_4843_5861.mvt new file mode 100644 index 0000000..0d1add6 Binary files /dev/null and b/tests/fixtures/tile_14_4843_5861.mvt differ diff --git a/tests/test_map.py b/tests/test_map.py new file mode 100644 index 0000000..d2ce978 --- /dev/null +++ b/tests/test_map.py @@ -0,0 +1,397 @@ +"""Tests for the node map: MVT decoding, projection, the braille canvas, and the tool. + +The vector-tile decoder and geometry are exercised against a real cached tile fixture +(``tests/fixtures/tile_14_4843_5861.mvt``, central Montréal) so the parsing is verified end +to end without the network; the projection, canvas, and compositor are pure; the tool runs +against the :class:`MockDevice` simulator with the basemap disabled (no network in tests). +""" + +from __future__ import annotations + +import io +import re +from datetime import timedelta +from pathlib import Path + +import pytest +from rich.console import Console + +from meshtools.core.geo import BBox, Viewport, haversine_km, lonlat_to_world, world_to_lonlat +from meshtools.core.models import ( + NODE_TYPE_CHAT, + NODE_TYPE_REPEATER, + Observation, + utcnow, +) +from meshtools.core.mvt import GEOM_LINE, GEOM_POLYGON, decode_tile +from meshtools.tools.map import MapTool +from meshtools.ui.mapcanvas import MapCanvas, parse_hex + +_FIXTURE = Path(__file__).parent / "fixtures" / "tile_14_4843_5861.mvt" +_ANSI = re.compile(r"\x1b\[[0-9;]*m") + + +def _plain(lines: list[str]) -> str: + """Strip ANSI colour from rendered lines for content assertions.""" + return "\n".join(_ANSI.sub("", line) for line in lines) + + +# -- MVT decoding ------------------------------------------------------------- + + +def test_decode_tile_reads_expected_layers() -> None: + """The real fixture decodes into the OpenMapTiles layers the map relies on.""" + layers = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes())} + for expected in ("transportation", "transportation_name", "water", "waterway", "place"): + assert expected in layers, expected + assert layers["transportation"].extent == 4096 + assert layers["transportation"].features # roads were decoded + + +def test_decode_tile_resolves_names_and_geometry() -> None: + """Street and place features carry resolved names and non-trivial geometry.""" + layers = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes())} + streets = [f.name for f in layers["transportation_name"].features if f.name] + assert any("Rue" in s or "Avenue" in s or "Av." in s for s in streets) + + places = {f.name: f.get("class") for f in layers["place"].features if f.name} + assert "Montréal" in places and places["Montréal"] == "city" + + # A road is a line with at least two points; water is a polygon. + assert any( + f.geom_type == GEOM_LINE and len(f.rings[0]) >= 2 + for f in layers["transportation"].features + ) + assert any(f.geom_type == GEOM_POLYGON for f in layers["water"].features) + + +def test_decode_tile_handles_empty_input() -> None: + """An empty byte string decodes to no layers rather than raising.""" + assert decode_tile(b"") == [] + + +# -- projection / viewport ---------------------------------------------------- + + +def test_world_projection_round_trips() -> None: + """lon/lat -> world pixels -> lon/lat recovers the original coordinate.""" + for lat, lon, z in [(45.5, -73.6, 14), (0.0, 0.0, 3), (-33.9, 151.2, 12)]: + wx, wy = lonlat_to_world(lat, lon, z) + rlat, rlon = world_to_lonlat(wx, wy, z) + assert rlat == pytest.approx(lat, abs=1e-6) + assert rlon == pytest.approx(lon, abs=1e-6) + + +def test_haversine_known_distance() -> None: + """One degree of longitude at the equator is ~111 km; identical points are 0.""" + assert haversine_km(0.0, 0.0, 0.0, 1.0) == pytest.approx(111.19, abs=0.5) + assert haversine_km(45.0, -73.0, 45.0, -73.0) == 0.0 + + +def test_viewport_center_projects_to_the_middle() -> None: + """The viewport's centre lands at the middle dot; north is up.""" + vp = Viewport(45.5, -73.6, 13, 200, 120) + cx, cy = vp.lonlat_to_dot(45.5, -73.6) + assert cx == pytest.approx(100, abs=0.5) and cy == pytest.approx(60, abs=0.5) + # A point due north projects above the centre (smaller y). + _, ny = vp.lonlat_to_dot(45.6, -73.6) + assert ny < cy + + +def test_viewport_fit_frames_all_points() -> None: + """A fitted viewport is centred on the points and zoomed so they share a tile scale.""" + pts = [(45.5019, -73.5674), (45.4768, -73.5990)] + vp = Viewport.fit(pts, 200, 120, max_zoom=14) + assert 8 <= vp.zoom <= 14 + for lat, lon in pts: + x, y = vp.lonlat_to_dot(lat, lon) + assert 0 <= x < vp.dot_w and 0 <= y < vp.dot_h # every point is on-canvas + + +def test_viewport_zoom_and_pan() -> None: + """Zooming and panning return new viewports; zoom clamps to its bounds.""" + vp = Viewport(45.5, -73.6, 10, 200, 120) + assert vp.zoomed(2).zoom == 12 + assert vp.zoomed(50, max_zoom=16).zoom == 16 # clamped + assert vp.zoomed(-50, min_zoom=3).zoom == 3 + east = vp.panned(0.5, 0) + assert east.center_lon > vp.center_lon # panning east increases longitude + north = vp.panned(0, -0.5) + assert north.center_lat > vp.center_lat # panning north increases latitude + + +def test_viewport_tiles_cover_the_view() -> None: + """The viewport reports the handful of tiles overlapping it at the tile zoom.""" + vp = Viewport(45.5019, -73.5674, 14, 200, 120) + tiles = vp.tiles(14) + assert tiles and all(z == 14 for z, _, _ in tiles) + # The fixture tile for this centre must be among them. + assert (14, 4843, 5861) in tiles + + +# -- braille canvas ----------------------------------------------------------- + + +def test_canvas_plots_dots_as_braille() -> None: + """A single plotted dot renders as a braille glyph in the top-left cell.""" + canvas = MapCanvas(4, 2) + canvas.plot(0, 0, (255, 0, 0), 1) + out = _plain(canvas.to_ansi_lines()) + assert out.splitlines()[0][0] == "⠁" # braille dot-1 + + +def test_canvas_priority_decides_cell_colour() -> None: + """When dots from two features share a cell, the higher priority sets its colour.""" + canvas = MapCanvas(1, 1) + canvas.plot(0, 0, (10, 20, 30), 1) # low priority + canvas.plot(1, 0, (200, 100, 50), 9) # high priority, same cell + ansi = "".join(canvas.to_ansi_lines()) + assert "38;2;200;100;50" in ansi # the high-priority colour won + assert "38;2;10;20;30" not in ansi + + +def test_canvas_marker_and_label_do_not_overprint() -> None: + """A marker keeps its glyph and places its label in a neighbouring cell.""" + canvas = MapCanvas(20, 3) + canvas.marker(4, 4, "★", (255, 255, 255), label="me") + out = _plain(canvas.to_ansi_lines()) + assert "★" in out and "me" in out + row = out.splitlines()[1] + assert "★me" not in row # a gap sits between the marker and its label + + +def test_canvas_label_collision_is_avoided() -> None: + """A checked label is skipped when it would overlap an already-placed one.""" + canvas = MapCanvas(20, 1) + assert canvas.place_label(10, 0, "First", (200, 200, 200)) + assert not canvas.place_label(10, 0, "Second", (200, 200, 200)) # overlaps → skipped + + +def test_parse_hex() -> None: + """Hex colours parse to RGB triples, with or without the leading hash.""" + assert parse_hex("#38bdf8") == (0x38, 0xBD, 0xF8) + assert parse_hex("ffffff") == (255, 255, 255) + + +# -- compositor over a real tile ---------------------------------------------- + + +def test_render_map_draws_basemap_streets_and_labels() -> None: + """Rendering the real fixture yields braille geometry and a place label.""" + from meshtools.ui.map_render import MapMarker, render_map + + vp = Viewport(45.5019, -73.5674, 14, 180, 120) + layers = decode_tile(_FIXTURE.read_bytes()) + tiles = {(14, 4843, 5861): layers} + markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)] + out = _plain(render_map(vp, tiles, markers)) + assert any(0x2800 <= ord(ch) <= 0x28FF for ch in out) # braille was drawn + assert "▲" in out and "Yagi" in out # the repeater marker + label + assert "Montréal" in out # a place label from the tile + + +def test_render_map_prioritises_repeater_glyph() -> None: + """A repeater and a node at the same spot resolve to the repeater's marker on top.""" + from meshtools.ui.map_render import MapMarker, render_map + + vp = Viewport(45.50, -73.57, 14, 60, 40) + markers = [ + MapMarker("node", 45.50, -73.57), + MapMarker("rptr", 45.50, -73.57, is_repeater=True), + ] + out = _plain(render_map(vp, {}, markers)) + assert "▲" in out and "●" not in out + + +def test_render_map_works_without_basemap() -> None: + """With no tiles the nodes still render on a blank grid (the offline fallback).""" + from meshtools.ui.map_render import MapMarker, render_map + + vp = Viewport.fit([(45.5, -73.6), (45.4, -73.5)], 120, 80, max_zoom=14) + markers = [ + MapMarker("A", 45.5, -73.6, is_self=True), + MapMarker("B", 45.4, -73.5, is_repeater=True), + ] + out = _plain(render_map(vp, {}, markers)) + assert "★" in out and "▲" in out + + +# -- basemap source (offline behaviour) --------------------------------------- + + +def test_basemap_source_offline_is_graceful(tmp_path: Path) -> None: + """An unreachable source reports unavailable and returns no tiles, never raising.""" + from meshtools.services.basemap import BasemapSource + + src = BasemapSource(tmp_path / "cache", tilejson_url="http://127.0.0.1:1/none", timeout=0.2) + assert src.available is False + assert src.max_zoom == 14 # sensible default when the TileJSON can't be fetched + assert src.load_tile(14, 4843, 5861) is None + + +def test_basemap_source_reads_cached_tile(tmp_path: Path) -> None: + """A tile already on disk is decoded without any network access.""" + from meshtools.services.basemap import BasemapSource + + cache = tmp_path / "cache" + dest = cache / "tiles" / "14" / "4843" + dest.mkdir(parents=True) + (dest / "5861.pbf").write_bytes(_FIXTURE.read_bytes()) + src = BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2) + layers = src.load_tile(14, 4843, 5861) + assert layers is not None + assert any(layer.name == "transportation" for layer in layers) + + +# -- the tool ----------------------------------------------------------------- + + +@pytest.fixture() +def ctx(tmp_path: Path): + """A mock-backed application context with the plain (console) UI surface.""" + from meshtools.context import AppContext + from meshtools.core.admin_store import AdminStore + from meshtools.core.config import Settings + from meshtools.core.device_store import DeviceStore + from meshtools.persistence.repository import Repository + from meshtools.ui.theme import MESH_THEME + + settings = Settings(config_dir=tmp_path, db_path=tmp_path / "map.db") + context = AppContext( + console=Console(file=io.StringIO(), theme=MESH_THEME), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(tmp_path / "devices.json"), + admin_store=AdminStore(tmp_path / "admin.json"), + mock=True, + ) + yield context + context.repo.close() + + +def _seed_located_nodes(repo) -> None: + """Record a repeater and a leaf node (each located) plus one node with no fix.""" + run_id = repo.start_run("monitor", {}) + now = utcnow() + repo.record_observation( + run_id, + Observation(node="a1", name="Yagi", node_type=NODE_TYPE_REPEATER, snr=6.0, + lat=45.55, lon=-73.55, observed_at=now), + ) + repo.record_observation( + run_id, + Observation(node="d4", name="Alice", node_type=NODE_TYPE_CHAT, snr=3.0, + lat=45.40, lon=-73.60, observed_at=now + timedelta(seconds=1)), + ) + repo.record_observation( + run_id, Observation(node="c3", name="Bot", node_type=NODE_TYPE_CHAT, snr=1.0) + ) + + +async def test_map_tool_static_render_counts_repeaters(ctx) -> None: + """The CLI path renders located nodes (basemap off) and reports the breakdown.""" + _seed_located_nodes(ctx.repo) + result = await MapTool().run(ctx, {"static": True, "basemap": False}) + assert result.summary["located"] == 2 # the mock companion has no fix (0/0) + assert result.summary["repeaters"] == 1 + assert result.summary["nodes"] == 1 + assert result.summary["self_located"] is False + # Something was rendered into the plain UI surface. + assert ctx.console.file.getvalue() != "" + + +async def test_map_tool_reports_nothing_to_plot(ctx) -> None: + """With no located history the tool returns cleanly instead of rendering.""" + result = await MapTool().run(ctx, {"static": True, "basemap": False}) + assert result.summary == {"located": 0} + + +class _StubSession: + """Minimal stand-in for :class:`TuiSession` for driving :class:`MapScreen`.""" + + def __init__(self, cols: int, rows: int) -> None: + self._cols, self._rows = cols, rows + self.invalidated = 0 + + def base_body_size(self) -> tuple[int, int]: + return self._cols, self._rows + + def invalidate(self) -> None: + self.invalidated += 1 + + +class _StubSource: + """An always-offline tile source, so the screen renders nodes only (no network).""" + + available = False + max_zoom = 14 + + def load_tile(self, z: int, x: int, y: int): + return None + + +def test_map_screen_renders_pans_zooms_and_resets() -> None: + """The interactive screen fills its body, and wasd/zoom/reset move the viewport.""" + from meshtools.core.geo import Viewport + from meshtools.ui.map_render import MapMarker + from meshtools.ui.map_screen import MapScreen + + session = _StubSession(80, 24) + markers = [ + MapMarker("A", 45.50, -73.60, is_repeater=True), + MapMarker("B", 45.40, -73.50), + ] + screen = MapScreen(session, markers, _StubSource(), 14) + + lines = screen.render_body(80) + assert len(lines) == 24 # fills the body height reported by the session + start = screen._viewport + + screen.handle("text", "d") # pan east → longitude increases + assert screen._viewport.center_lon > start.center_lon + screen.handle("text", "w") # pan north → latitude increases + assert screen._viewport.center_lat > start.center_lat + + z = screen._viewport.zoom + screen.handle("text", "=") + assert screen._viewport.zoom == z + 1 + screen.handle("text", "-") + assert screen._viewport.zoom == z + + screen.handle("text", "r") # reset refits to the nodes + refit = Viewport.fit([(m.lat, m.lon) for m in markers], start.dot_w, start.dot_h, max_zoom=14) + assert screen._viewport.zoom == refit.zoom + assert screen._viewport.center_lat == pytest.approx(refit.center_lat) + + # Offline source: nodes still render and both markers are present. + assert "▲" in _plain(screen.render_body(80)) and "●" in _plain(screen.render_body(80)) + assert "offline" in screen.footer_hint + + +def test_map_screen_escape_dismisses() -> None: + """Esc resolves the screen's future with None (backs out to the menu).""" + import asyncio + + from meshtools.ui.map_render import MapMarker + from meshtools.ui.map_screen import MapScreen + + screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14) + + async def drive() -> object: + screen.future = asyncio.get_running_loop().create_future() + screen.handle("escape") + return await screen.future + + assert asyncio.run(drive()) is None + + +def test_map_observation_round_trips_node_type(ctx) -> None: + """A stored observation's advert type survives persistence and aggregation.""" + run_id = ctx.repo.start_run("monitor", {}) + ctx.repo.record_observation( + run_id, + Observation(node="a1", name="Yagi", node_type=NODE_TYPE_REPEATER, snr=6.0, + lat=45.5, lon=-73.5), + ) + node = next(n for n in ctx.repo.heard_nodes() if n.node == "a1") + assert node.is_repeater and node.node_type == NODE_TYPE_REPEATER