Jean-Pierre MartineauandClaude Opus 5
07a5c82945
Map. Enter framed the matches, which is only ever half of what finishing a find means. Typing a query dims everything it doesn't match, and the common finish is "yes, that one -- now let me look around it": the query has done its job and only the dimming is in the way. So Enter drops the query and leaves the view where the reader put it, and ^Enter is the other finish -- take me to them -- keeping the query, because a frame is a place to arrive and framing tighter from there is one more press. ^Enter is not a keycode any terminal has to offer: Enter *is* ^M, so a console without modifyOtherKeys has nothing left to modify. What most do send is LF, which prompt_toolkit spells ControlJ and nothing here was using, so it is bound; the right-Ctrl rescue covers the rest; and where a console spells neither, the F3 Frame chip is the affordance -- which is the platform where that was already the whole story. Mesh walk. The breadcrumb elided its head and snapped right, so a long walk had a beginning that could not be read at all. Left and right now scroll it a hop at a time, and a scrolled line grows its own trailing elision: the mark means the same thing on either side, more walk that way. It stops where the head comes into view rather than running the line off the edge, and any change to the trail rewinds it -- the focus is what the next move is read from. The hint trades "type to find" for the arrows exactly while the trail overflows: typing announces itself the moment a letter lands, the scroll had no way to. And the find now narrows the canvas fan, not just the list. The east of the picture is the ways onward, which is what the query is asking about; the focus and the node we came from are the walk itself, so they hold through whatever is typed and the graph keeps its bearings while the choices thin. Chat. The transcript's sections are days and its section jump was already bound, so it takes the free left pair the same way a grouped list's does -- Day up outside Day down, the lane's handedness, ctrl_pageup/ctrl_pagedown -- lit only when there is a second day to step to. Read off the two end messages rather than by re-dating the whole transcript every paint. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1802 lines
75 KiB
Python
1802 lines
75 KiB
Python
"""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 os
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from rich.cells import cell_len
|
||
from rich.console import Console
|
||
|
||
from meshterm.core.geo import BBox, Viewport, haversine_km, lonlat_to_world, world_to_lonlat
|
||
from meshterm.core.models import (
|
||
NODE_TYPE_REPEATER,
|
||
Observation,
|
||
utcnow,
|
||
)
|
||
from meshterm.core.mvt import GEOM_LINE, GEOM_POLYGON, Layer, decode_tile
|
||
from meshterm.tools.map import MapTool
|
||
from meshterm.ui.mapcanvas import MapCanvas, parse_hex
|
||
|
||
_FIXTURE = Path(__file__).parent / "fixtures" / "tile_14_4843_5861.mvt"
|
||
from tests.conftest import plain as _plain # THE strip-and-join screen reader
|
||
|
||
|
||
# -- 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"") == []
|
||
|
||
|
||
def test_decode_tile_filter_keeps_the_whole_tile_visible() -> None:
|
||
"""A narrowed decode still names every layer — it only skips their features.
|
||
|
||
The tile source decides whether a cache entry is real by whether it parsed into
|
||
layers, so narrowing must never make a tile look like it holds nothing.
|
||
"""
|
||
raw = _FIXTURE.read_bytes()
|
||
full = decode_tile(raw)
|
||
narrow = decode_tile(raw, layers={"water"})
|
||
|
||
assert [layer.name for layer in narrow] == [layer.name for layer in full]
|
||
assert all(layer.extent == 4096 for layer in narrow)
|
||
picked = {layer.name: layer for layer in narrow}
|
||
assert picked["water"].features # the one we asked for was decoded
|
||
assert not picked["building"].features # the rest were skipped, not dropped
|
||
|
||
|
||
def test_decode_tile_filter_is_invisible_to_the_renderer() -> None:
|
||
"""Narrowing to ``DRAWN_LAYERS`` draws exactly the map a full decode draws.
|
||
|
||
This is the anti-drift guard: if ``_draw_tile`` grows a lookup for a layer that
|
||
isn't in ``DRAWN_LAYERS``, that layer arrives featureless and the rendered output
|
||
diverges here.
|
||
"""
|
||
from meshterm.ui.map_render import DRAWN_LAYERS, MapMarker, render_map
|
||
|
||
raw = _FIXTURE.read_bytes()
|
||
vp = Viewport(45.5019, -73.5674, 14, 180, 120)
|
||
markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)]
|
||
|
||
def drawn(layers):
|
||
return render_map(vp, {(14, 4843, 5861): layers}, markers)
|
||
|
||
assert drawn(decode_tile(raw, layers=DRAWN_LAYERS)) == drawn(decode_tile(raw))
|
||
|
||
|
||
def test_decode_tile_filter_skips_the_layers_the_map_never_draws() -> None:
|
||
"""The layers left undecoded are the ones that made a tile expensive."""
|
||
from meshterm.ui.map_render import DRAWN_LAYERS
|
||
|
||
narrow = {layer.name: layer for layer in decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)}
|
||
for skipped in ("housenumber", "poi", "mountain_peak"):
|
||
assert skipped in narrow, f"{skipped} should still be named"
|
||
assert not narrow[skipped].features, f"{skipped} should not have been decoded"
|
||
|
||
|
||
# -- buildings ----------------------------------------------------------------
|
||
|
||
|
||
def _building_dots(zoom: int) -> int:
|
||
"""Lit cells in a render of the fixture's buildings alone, at ``zoom``."""
|
||
from meshterm.ui.map_render import render_map
|
||
|
||
layers = decode_tile(_FIXTURE.read_bytes(), layers={"building"})
|
||
vp = Viewport(45.4995, -73.5690, zoom, 53 * 2, 26 * 4)
|
||
out = _plain(render_map(vp, {(14, 4843, 5861): layers}, []))
|
||
return sum(1 for ch in out if ch not in " \n")
|
||
|
||
|
||
def test_buildings_wait_for_the_zoom_that_can_hold_them() -> None:
|
||
"""Below the gate a footprint is about two dots across, so the layer is a haze."""
|
||
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM
|
||
|
||
assert _building_dots(_BUILDING_MIN_ZOOM) > 100
|
||
assert _building_dots(_BUILDING_MIN_ZOOM - 1) == 0
|
||
|
||
|
||
def test_building_fill_is_stippled_so_the_streets_survive_it() -> None:
|
||
"""A braille dot is one bit: a *solid* fill would erase the street grid it covers.
|
||
|
||
The stipple is what keeps a building a shade rather than an eraser — every cell it
|
||
touches keeps free dots for a road to be drawn through.
|
||
"""
|
||
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM, render_map
|
||
|
||
layers = decode_tile(_FIXTURE.read_bytes(), layers={"building"})
|
||
vp = Viewport(45.4995, -73.5690, _BUILDING_MIN_ZOOM, 53 * 2, 26 * 4)
|
||
out = _plain(render_map(vp, {(14, 4843, 5861): layers}, []))
|
||
|
||
assert "⣿" not in out, "a fully lit cell means the fill left a road nowhere to go"
|
||
|
||
|
||
def test_offscreen_buildings_do_not_change_the_frame() -> None:
|
||
"""The bbox reject is an optimisation, so it must be invisible in the output.
|
||
|
||
A downtown tile carries thousands of footprints and a zoomed-in view holds a few
|
||
dozen; each one is rejected on its own tile-local bounds before a single point of it
|
||
is projected. Adding a footprint the view cannot reach must render byte-identically.
|
||
"""
|
||
from meshterm.core.mvt import Feature
|
||
from meshterm.ui.map_render import _BUILDING_MIN_ZOOM, render_map
|
||
|
||
vp = Viewport(45.4995, -73.5690, _BUILDING_MIN_ZOOM, 53 * 2, 26 * 4)
|
||
cx, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
||
here = [(cx - 40, cy - 40), (cx + 40, cy - 40), (cx + 40, cy + 40),
|
||
(cx - 40, cy + 40), (cx - 40, cy - 40)]
|
||
far = [(10, 10), (90, 10), (90, 90), (10, 90), (10, 10)]
|
||
|
||
def frame(rings):
|
||
layer = Layer(name="building", extent=4096,
|
||
features=[Feature(geom_type=GEOM_POLYGON, rings=rings, tags={})])
|
||
return render_map(vp, {(14, 4843, 5861): [layer]}, [])
|
||
|
||
assert _plain(frame([here])).strip(), "the in-view footprint should draw something"
|
||
assert frame([here]) == frame([here, far])
|
||
|
||
|
||
# -- 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_fit_fraction_ignores_outliers() -> None:
|
||
"""Framing half the nodes zooms to the dense core, letting far outliers fall off-canvas."""
|
||
# A tight downtown cluster of five nodes (~50 m across) plus one distant outlier.
|
||
core = [
|
||
(45.5000, -73.5600), (45.5003, -73.5602), (45.4998, -73.5598),
|
||
(45.5001, -73.5599), (45.4999, -73.5601),
|
||
]
|
||
outlier = (46.80, -71.20)
|
||
pts = core + [outlier]
|
||
|
||
full = Viewport.fit(pts, 200, 120, max_zoom=16)
|
||
half = Viewport.fit(pts, 200, 120, max_zoom=16, fraction=0.5)
|
||
|
||
# The core-only frame is zoomed in tighter than the frame that must hold the outlier.
|
||
assert half.zoom > full.zoom
|
||
# Every clustered node stays on-canvas; the outlier is pushed off.
|
||
for lat, lon in core:
|
||
x, y = half.lonlat_to_dot(lat, lon)
|
||
assert 0 <= x < half.dot_w and 0 <= y < half.dot_h
|
||
ox, oy = half.lonlat_to_dot(*outlier)
|
||
assert not (0 <= ox < half.dot_w and 0 <= oy < half.dot_h)
|
||
|
||
|
||
def test_viewport_fit_fraction_keeps_small_sets_whole() -> None:
|
||
"""With only two nodes there is no core to isolate — both are still framed."""
|
||
pts = [(45.5019, -73.5674), (45.4768, -73.5990)]
|
||
half = Viewport.fit(pts, 200, 120, max_zoom=14, fraction=0.5)
|
||
full = Viewport.fit(pts, 200, 120, max_zoom=14)
|
||
assert (half.center_lat, half.center_lon, half.zoom) == (
|
||
full.center_lat, full.center_lon, full.zoom
|
||
)
|
||
|
||
|
||
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))
|
||
assert canvas.marker_label(4, 4, "me", (255, 255, 255))
|
||
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_canvas_labels_keep_a_vertical_gap() -> None:
|
||
"""A checked label is skipped when it would sit flush above/below existing text."""
|
||
canvas = MapCanvas(20, 3)
|
||
assert canvas.place_label(20, 4, "Row1", (200, 200, 200)) # dot y=4 → cell row 1
|
||
# Directly below (cell row 2) with overlapping columns: rejected for lack of a gap.
|
||
assert not canvas.place_label(20, 8, "Row2", (200, 200, 200))
|
||
# Same row but clear of the horizontal span still fits.
|
||
assert canvas.place_label(2, 4, "Far", (200, 200, 200))
|
||
|
||
|
||
def _dot_colors(lines: list[str]) -> set[tuple[int, int, int]]:
|
||
"""Every truecolour a braille run was drawn in (the ground's colours, not the text's)."""
|
||
return {
|
||
(int(r), int(g), int(b))
|
||
for r, g, b in re.findall(
|
||
r"38;2;(\d+);(\d+);(\d+)m(?:\x1b\[1m)?[⠀-⣿]", "".join(lines)
|
||
)
|
||
}
|
||
|
||
|
||
def test_canvas_paste_raster_offsets_fades_and_yields_the_cell() -> None:
|
||
"""A pasted raster lands where the caller's index tables put it, dimmed, underneath.
|
||
|
||
``-1`` is a cell with no source — the ground a pan has just moved onto — and the
|
||
pasted cell keeps the empty-cell priority so anything drawn afterwards wins it.
|
||
"""
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
|
||
src = MapCanvas(3, 1)
|
||
src.plot(0, 0, (200, 100, 50), 5) # cell 0
|
||
src.plot(4, 0, (60, 120, 240), 5) # cell 2
|
||
|
||
canvas = MapCanvas(3, 1)
|
||
canvas.paste_raster(src.raster(), [-1, 0, 2], [0], fade=0.5)
|
||
out = "".join(canvas.to_ansi_lines())
|
||
assert _plain([out]) == " ⠁⠁" # shifted one cell right; the gap has no source
|
||
assert _dot_colors([out]) == {(100, 50, 25), (30, 60, 120)}
|
||
|
||
canvas.plot(2, 0, (10, 20, 30), 0) # a real feature, drawn after, takes the cell
|
||
assert _dot_colors(["".join(canvas.to_ansi_lines())]) >= {(10, 20, 30)}
|
||
|
||
|
||
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 meshterm.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 meshterm.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 _street_labels_placed(cols: int, rows: int, zoom: int) -> int:
|
||
"""How many street names actually reach the canvas at this size and zoom."""
|
||
from meshterm.ui.map_render import DRAWN_LAYERS, _draw_tile, _Frame
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
|
||
layers = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
||
vp = Viewport(45.5019, -73.5674, zoom, cols * 2, (rows - 4) * 4)
|
||
canvas = MapCanvas(vp.dot_w // 2, vp.dot_h // 4)
|
||
frame = _Frame(canvas=canvas, viewport=vp)
|
||
_draw_tile(frame, layers, 14, 4843, 5861)
|
||
|
||
placed = 0
|
||
for label in sorted(frame.labels, key=lambda l: l.rank):
|
||
if vp.zoom < label.min_zoom:
|
||
continue
|
||
for ax, ay in ((label.x, label.y), *label.alts):
|
||
if canvas.place_label(ax, ay, label.text, label.color, bold=label.bold):
|
||
placed += label.min_zoom == 15 # the street-label gate identifies them
|
||
break
|
||
return placed
|
||
|
||
|
||
def test_street_labels_get_denser_as_you_zoom_in() -> None:
|
||
"""Zooming toward a street must not make its name less likely to appear.
|
||
|
||
Anchoring a label to the feature's own midpoint pinned it to a fixed geographic
|
||
point, so the tighter the view the less often that point was still on screen — the
|
||
fixture's 265 named streets yielded 2 labels at zoom 16 and 10 at zoom 15.
|
||
"""
|
||
assert _street_labels_placed(53, 26, 16) >= 5
|
||
assert _street_labels_placed(53, 26, 15) >= 10
|
||
# A wider terminal sees more of the same ground, so it may name more -- never fewer.
|
||
assert _street_labels_placed(72, 24, 16) >= _street_labels_placed(53, 26, 16)
|
||
|
||
|
||
def _tile_local_of_view_centre(vp: Viewport, tz: int, tx: int, ty: int, extent: int = 4096):
|
||
"""The tile-local point that lands in the middle of this viewport (inverts feature_to_dot)."""
|
||
from meshterm.core.geo import TILE_PX
|
||
|
||
ox, oy = vp.origin_world
|
||
scale = 2.0 ** (vp.zoom - tz)
|
||
lx = extent * ((vp.dot_w / 2 + ox) / (TILE_PX * scale) - tx)
|
||
ly = extent * ((vp.dot_h / 2 + oy) / (TILE_PX * scale) - ty)
|
||
return lx, ly
|
||
|
||
|
||
def test_street_name_is_drawn_only_once() -> None:
|
||
"""OSM splits a long street into several features; the map names it once."""
|
||
from meshterm.ui.map_render import DRAWN_LAYERS, MapMarker, render_map
|
||
|
||
layers = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
||
vp = Viewport(45.5019, -73.5674, 16, 53 * 2, 22 * 4)
|
||
out = _plain(render_map(vp, {(14, 4843, 5861): layers},
|
||
[MapMarker("YUL", 45.5040, -73.5700, is_repeater=True)]))
|
||
|
||
# René-Lévesque arrives as several segments and used to be drawn twice on one screen.
|
||
assert out.count("René-Lévesque") <= 1, "a street was named more than once"
|
||
|
||
|
||
def test_line_label_anchors_on_the_visible_stretch() -> None:
|
||
"""A street crossing the view is named even with both its endpoints off screen."""
|
||
from meshterm.ui.map_render import _Frame, _add_line_label
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
|
||
vp = Viewport(45.5019, -73.5674, 14, 120, 80)
|
||
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
||
_, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
||
# A line spanning the whole tile at the view's latitude: its own midpoint is far away,
|
||
# but it crosses the canvas, so the clip must find it.
|
||
_add_line_label(frame, [[(0, cy), (4096, cy)]], 4096, 14, 4843, 5861, "Rue Long",
|
||
("#9aa0aa", False, 7))
|
||
|
||
assert len(frame.labels) == 1
|
||
label = frame.labels[0]
|
||
assert 0 <= label.x <= vp.dot_w and 0 <= label.y <= vp.dot_h
|
||
|
||
|
||
def test_line_label_off_screen_is_not_queued() -> None:
|
||
"""A feature with nothing on screen costs no label slot at all."""
|
||
from meshterm.ui.map_render import _Frame, _add_line_label
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
|
||
vp = Viewport(45.5019, -73.5674, 16, 120, 80)
|
||
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
||
# A short line in the far corner of the tile, well outside a zoom-16 window.
|
||
_add_line_label(frame, [[(0, 0), (8, 8)]], 4096, 14, 4843, 5861, "Nowhere",
|
||
("#9aa0aa", False, 7))
|
||
|
||
assert frame.labels == []
|
||
|
||
|
||
def test_line_label_offers_alternates_along_the_visible_run() -> None:
|
||
"""A crowded first choice falls back further along the street rather than vanishing."""
|
||
from meshterm.ui.map_render import _Frame, _add_line_label
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
|
||
vp = Viewport(45.5019, -73.5674, 14, 120, 80)
|
||
frame = _Frame(canvas=MapCanvas(60, 20), viewport=vp)
|
||
cx, cy = _tile_local_of_view_centre(vp, 14, 4843, 5861)
|
||
# A zig-zag across the view gives several visible pieces to choose between.
|
||
ring = [(cx - 120 + i * 40, cy - 40 + (i % 2) * 80) for i in range(8)]
|
||
_add_line_label(frame, [ring], 4096, 14, 4843, 5861, "Rue Zig", ("#9aa0aa", False, 7))
|
||
|
||
assert frame.labels and frame.labels[0].alts, "no fallback anchors offered"
|
||
assert all(a != (frame.labels[0].x, frame.labels[0].y) for a in frame.labels[0].alts)
|
||
|
||
|
||
def test_render_map_drops_crowded_labels_favouring_repeaters() -> None:
|
||
"""When labels can't all fit, the repeater's wins and a crowded node's is dropped."""
|
||
from meshterm.ui.map_render import MapMarker, render_map
|
||
|
||
# Three nodes stacked on one spot: only the two sides (left/right) can hold a label,
|
||
# so one of the three must show as a bare marker — and the repeater must not be it.
|
||
vp = Viewport(45.50, -73.57, 14, 40, 8)
|
||
markers = [
|
||
MapMarker("NODEONE", 45.50, -73.57),
|
||
MapMarker("NODETWO", 45.50, -73.57),
|
||
MapMarker("REPEATER", 45.50, -73.57, is_repeater=True),
|
||
]
|
||
out = _plain(render_map(vp, {}, markers))
|
||
assert "▲" in out # the repeater's glyph is on top
|
||
assert "REPEATER" in out # and it keeps its label (placed first)
|
||
# Only one of the two leaf nodes could fit a label; the other is a bare marker.
|
||
assert ("NODEONE" in out) != ("NODETWO" in out)
|
||
|
||
|
||
def _glyph_color(lines: list[str], glyph: str) -> tuple[int, int, int]:
|
||
"""Extract the truecolour ``(r, g, b)`` the given glyph was rendered with."""
|
||
m = re.search(
|
||
r"38;2;(\d+);(\d+);(\d+)m(?:\x1b\[1m)?" + re.escape(glyph), "".join(lines)
|
||
)
|
||
assert m, f"glyph {glyph!r} not found with a colour"
|
||
return int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||
|
||
|
||
def test_render_map_brightens_piled_markers() -> None:
|
||
"""A cell many nodes share renders its glyph brighter than a lone node's."""
|
||
from meshterm.ui.map_render import _NODE, MapMarker, render_map
|
||
|
||
base = parse_hex(_NODE[1])
|
||
vp = Viewport(45.50, -73.57, 14, 60, 40)
|
||
|
||
lone = render_map(vp, {}, [MapMarker("solo", 45.50, -73.57)])
|
||
assert _glyph_color(lone, "●") == base # a single node keeps its base colour
|
||
|
||
crowd = [MapMarker(f"n{i}", 45.50, -73.57) for i in range(8)]
|
||
piled = _glyph_color(render_map(vp, {}, crowd), "●")
|
||
# Washed toward white: every channel is brighter than the base cyan.
|
||
assert all(p > b for p, b in zip(piled, base))
|
||
assert piled != base
|
||
|
||
|
||
def test_render_map_works_without_basemap() -> None:
|
||
"""With no tiles the nodes still render on a blank grid (the offline fallback)."""
|
||
from meshterm.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
|
||
|
||
|
||
# -- the ghost ground ---------------------------------------------------------
|
||
|
||
|
||
def _dot_rows(lines: list[str]) -> list[str]:
|
||
"""Each rendered row reduced to its braille dots (labels and markers blanked out)."""
|
||
return [
|
||
"".join(ch if 0x2800 <= ord(ch) <= 0x28FF else " " for ch in row)
|
||
for row in _plain(lines).split("\n")
|
||
]
|
||
|
||
|
||
def _ghosted(zoom: int = 14, *, dlon: float = 0.0, dlat: float = 0.0, dz: int = 0):
|
||
"""Draw the fixture, then paint a moved view standing on that frame's ground."""
|
||
from meshterm.ui.map_render import MapMarker, render_ground, render_map
|
||
|
||
vp = Viewport(45.5019, -73.5674, zoom, 106, 104)
|
||
tiles = {(14, 4843, 5861): decode_tile(_FIXTURE.read_bytes())}
|
||
markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)]
|
||
lines, ghost = render_ground(vp, tiles, markers)
|
||
moved = Viewport(vp.center_lat + dlat, vp.center_lon + dlon, zoom + dz, 106, 104)
|
||
return lines, render_map(moved, {}, markers, ghost=ghost)
|
||
|
||
|
||
def test_a_panned_view_stands_on_the_ground_it_just_had() -> None:
|
||
"""Panning slides the last frame's streets across rather than blanking to black.
|
||
|
||
Rasterizing takes far too long to sit on a keypress, so the paint that answers a pan
|
||
has no ground of its own. Without the ghost it drew the markers on an empty canvas —
|
||
the map going black between every step and flashing back when the frame landed.
|
||
"""
|
||
from meshterm.ui.map_render import MapMarker, render_map
|
||
|
||
vp = Viewport(45.5019, -73.5674, 14, 106, 104)
|
||
markers = [MapMarker("Yagi", 45.5019, -73.5674, is_repeater=True)]
|
||
bare = sum(1 for ch in _plain(render_map(vp, {}, markers)) if ch.strip())
|
||
|
||
drawn, moved = _ghosted(dlon=0.004)
|
||
lit = sum(1 for ch in _plain(moved) if ch.strip())
|
||
assert lit > bare * 10, "the panned paint is as empty as one with no ghost at all"
|
||
assert lit < sum(1 for ch in _plain(drawn) if ch.strip()), "nothing was left behind"
|
||
|
||
|
||
def test_the_ghost_ground_lands_where_the_pan_put_it() -> None:
|
||
"""The reused dots move by exactly the cells the view moved, to within one cell."""
|
||
drawn, moved = _ghosted(dlon=0.004)
|
||
# 0.004° of longitude at zoom 14: dots per degree = 256 * 2^14 / 360.
|
||
shift = round(0.004 * 256 * (2**14) / 360 / 2) # → cells (2 dots wide)
|
||
|
||
before, after = _dot_rows(drawn), _dot_rows(moved)
|
||
assert len(before) == len(after)
|
||
matched = sum(
|
||
1
|
||
for old, new in zip(before, after)
|
||
if old[shift:].rstrip() and new.rstrip() == old[shift:].rstrip()
|
||
)
|
||
assert matched > len(before) // 2, "the ground did not slide with the view"
|
||
|
||
|
||
def test_the_ghost_ground_carries_no_stale_text() -> None:
|
||
"""Only the dots come along: a label pinned to old ground would name the wrong place."""
|
||
drawn, moved = _ghosted(dlon=0.004)
|
||
assert "Montréal" in _plain(drawn)
|
||
assert "Montréal" not in _plain(moved)
|
||
assert "Yagi" in _plain(moved) # the markers are redrawn where they really are
|
||
|
||
|
||
def test_the_ghost_ground_dims_while_its_replacement_is_drawn() -> None:
|
||
"""Reused ground reads as provisional — it is stale, and short of the leading edge."""
|
||
from meshterm.ui.map_render import _GHOST_FADE
|
||
|
||
assert 0 < _GHOST_FADE < 1 # the desktop default; the 16-slot console binds it to 1.0
|
||
drawn, moved = _ghosted(dlon=0.004)
|
||
faded = {
|
||
tuple(round(c * _GHOST_FADE) for c in rgb) for rgb in _dot_colors(drawn)
|
||
}
|
||
ghost = _dot_colors(moved)
|
||
assert ghost and ghost <= faded, "the reused ground is not the ground we drew"
|
||
assert not ghost & _dot_colors(drawn), "it came through at full strength"
|
||
|
||
|
||
def test_the_ghost_ground_is_dropped_once_it_says_nothing() -> None:
|
||
"""A view that has left the old frame behind gets the honest empty canvas."""
|
||
_, far = _ghosted(dlon=4.0) # panned clean off the ground we had
|
||
assert not any(row.strip() for row in _dot_rows(far))
|
||
_, deep = _ghosted(dz=3) # zoomed past what a cell-coarse stand-in can say
|
||
assert not any(row.strip() for row in _dot_rows(deep))
|
||
_, near = _ghosted(dz=1) # one step is still a readable preview
|
||
assert any(row.strip() for row in _dot_rows(near))
|
||
|
||
|
||
# -- 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 meshterm.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 meshterm.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)
|
||
|
||
|
||
def _offline_source(cache: Path):
|
||
"""A source that can't reach anything — every fetch is silence, not an empty tile."""
|
||
from meshterm.services.basemap import BasemapSource
|
||
|
||
return BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2)
|
||
|
||
|
||
def test_basemap_source_never_caches_an_unanswered_fetch(tmp_path: Path) -> None:
|
||
"""A timeout writes nothing: a blank square now must not become a blank square for ever."""
|
||
from meshterm.services import basemap as basemap_mod
|
||
|
||
cache = tmp_path / "cache"
|
||
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
||
assert list(cache.rglob("*.pbf")) == []
|
||
|
||
# The same, with the source resolved and the *fetch* the thing that fails — the flaky
|
||
# link the PicoCalc lives on. The tile stays unknown, so the next pan asks again.
|
||
src = _offline_source(cache)
|
||
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf"
|
||
src._resolved = True
|
||
calls: list[str] = []
|
||
|
||
def _get(url: str):
|
||
calls.append(url)
|
||
return basemap_mod._Response(False, b"")
|
||
|
||
src._http_get = _get # type: ignore[method-assign]
|
||
assert src.load_tile(14, 4843, 5861) is None
|
||
assert src.load_tile(14, 4843, 5861) is None
|
||
assert len(calls) == 2 # retried, not written off as empty
|
||
assert list(cache.rglob("*.pbf")) == []
|
||
|
||
|
||
def test_basemap_source_prunes_a_blank_cached_tile(tmp_path: Path) -> None:
|
||
"""A zero-byte entry (an older build's failure marker) is dropped rather than drawn."""
|
||
cache = tmp_path / "cache"
|
||
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
||
tile.parent.mkdir(parents=True)
|
||
tile.write_bytes(b"")
|
||
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
||
assert not tile.exists() # pruned, so a session with network re-fetches it
|
||
|
||
|
||
def test_basemap_source_prunes_a_corrupt_cached_tile(tmp_path: Path) -> None:
|
||
"""Bytes cut short mid-write don't decode, so they're pruned instead of kept blank."""
|
||
cache = tmp_path / "cache"
|
||
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
||
tile.parent.mkdir(parents=True)
|
||
tile.write_bytes(_FIXTURE.read_bytes()[:200])
|
||
assert _offline_source(cache).load_tile(14, 4843, 5861) is None
|
||
assert not tile.exists()
|
||
|
||
|
||
def test_basemap_source_caches_only_tiles_with_content(tmp_path: Path) -> None:
|
||
"""The disk cache holds decodable geometry — never an empty or unparseable answer."""
|
||
from meshterm.services import basemap as basemap_mod
|
||
|
||
cache = tmp_path / "cache"
|
||
src = _offline_source(cache)
|
||
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf" # skip TileJSON resolution
|
||
src._resolved = True
|
||
|
||
bodies = iter([b"", b"not a vector tile at all", _FIXTURE.read_bytes()])
|
||
src._http_get = lambda url: basemap_mod._Response(True, next(bodies)) # type: ignore[method-assign]
|
||
|
||
assert src.load_tile(14, 1, 1) is None # answered "empty"
|
||
assert src.load_tile(14, 2, 2) is None # answered with junk
|
||
assert list(cache.rglob("*.pbf")) == []
|
||
layers = src.load_tile(14, 4843, 5861) # answered with a real tile
|
||
assert layers is not None
|
||
assert (cache / "tiles" / "14" / "4843" / "5861.pbf").exists()
|
||
|
||
|
||
def test_basemap_source_keeps_a_tile_whose_layers_are_all_undrawn(tmp_path: Path) -> None:
|
||
"""Narrowing the decode must not make a good cached tile look blank and get pruned.
|
||
|
||
A source told to decode a layer the tile doesn't carry sees no features at all. If
|
||
that were read as "nothing here", the entry would be deleted and re-downloaded on
|
||
every session — the cache paying for a rendering decision.
|
||
"""
|
||
from meshterm.services.basemap import BasemapSource
|
||
|
||
cache = tmp_path / "cache"
|
||
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
||
tile.parent.mkdir(parents=True)
|
||
tile.write_bytes(_FIXTURE.read_bytes())
|
||
|
||
src = BasemapSource(
|
||
cache,
|
||
tilejson_url="http://127.0.0.1:1/none",
|
||
timeout=0.2,
|
||
layers=frozenset({"a_layer_this_tile_does_not_have"}),
|
||
)
|
||
layers = src.load_tile(14, 4843, 5861)
|
||
|
||
assert layers, "a real tile must still read as real"
|
||
assert not any(layer.features for layer in layers) # nothing was decoded
|
||
assert tile.exists(), "the cache entry must survive"
|
||
|
||
|
||
def test_decoded_layers_survive_a_round_trip() -> None:
|
||
"""The sidecar encoding restores exactly what the decoder produced."""
|
||
from meshterm.core.mvt import dumps_layers, loads_layers
|
||
from meshterm.ui.map_render import DRAWN_LAYERS
|
||
|
||
original = decode_tile(_FIXTURE.read_bytes(), layers=DRAWN_LAYERS)
|
||
restored = loads_layers(dumps_layers(original, stamp="x"), stamp="x")
|
||
|
||
assert restored is not None
|
||
assert [(l.name, l.extent) for l in restored] == [(l.name, l.extent) for l in original]
|
||
for before, after in zip(original, restored):
|
||
assert after.features == before.features
|
||
|
||
|
||
def test_decoded_layers_refuse_a_blob_from_another_layer_set() -> None:
|
||
"""A narrowed blob must never be served to a caller that wants more of the tile."""
|
||
from meshterm.core.mvt import dumps_layers, loads_layers
|
||
|
||
blob = dumps_layers(decode_tile(_FIXTURE.read_bytes(), layers={"water"}), stamp="water")
|
||
assert loads_layers(blob, stamp="water") is not None
|
||
assert loads_layers(blob, stamp="water,place") is None
|
||
|
||
|
||
def test_decoded_layers_treat_damage_as_a_miss() -> None:
|
||
"""A truncated or foreign sidecar is a cache miss, never an exception."""
|
||
from meshterm.core.mvt import dumps_layers, loads_layers
|
||
|
||
blob = dumps_layers(decode_tile(_FIXTURE.read_bytes(), layers={"water"}), stamp="s")
|
||
assert loads_layers(blob[: len(blob) // 2], stamp="s") is None
|
||
assert loads_layers(b"", stamp="s") is None
|
||
assert loads_layers(b"not marshal at all", stamp="s") is None
|
||
|
||
|
||
def test_basemap_source_writes_and_reuses_a_decoded_sidecar(tmp_path: Path) -> None:
|
||
"""The second load parses nothing: it comes back from the sidecar."""
|
||
cache = tmp_path / "cache"
|
||
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
||
tile.parent.mkdir(parents=True)
|
||
tile.write_bytes(_FIXTURE.read_bytes())
|
||
|
||
src = _offline_source(cache)
|
||
first = src.load_tile(14, 4843, 5861)
|
||
sidecar = cache / "decoded" / "14" / "4843" / "5861.bin"
|
||
assert first is not None
|
||
assert sidecar.exists(), "decoding should have been written down"
|
||
|
||
# Break the raw tile: a second load that still works can only have used the sidecar.
|
||
tile.write_bytes(b"junk that cannot decode")
|
||
second = src.load_tile(14, 4843, 5861)
|
||
assert second is not None
|
||
assert [(l.name, len(l.features)) for l in second] == [
|
||
(l.name, len(l.features)) for l in first
|
||
]
|
||
|
||
|
||
def test_basemap_source_ignores_a_sidecar_from_another_layer_set(tmp_path: Path) -> None:
|
||
"""Widening what the map draws re-decodes rather than serving the narrower blob."""
|
||
from meshterm.services.basemap import BasemapSource
|
||
|
||
cache = tmp_path / "cache"
|
||
tile = cache / "tiles" / "14" / "4843" / "5861.pbf"
|
||
tile.parent.mkdir(parents=True)
|
||
tile.write_bytes(_FIXTURE.read_bytes())
|
||
|
||
def source(layers):
|
||
return BasemapSource(
|
||
cache, tilejson_url="http://127.0.0.1:1/none", timeout=0.2, layers=layers
|
||
)
|
||
|
||
narrow = source(frozenset({"water"}))
|
||
assert narrow.load_tile(14, 4843, 5861) is not None
|
||
|
||
wider = source(frozenset({"water", "place"}))
|
||
layers = {layer.name: layer for layer in wider.load_tile(14, 4843, 5861) or []}
|
||
assert layers["place"].features, "the wider set must have been decoded afresh"
|
||
|
||
|
||
def test_basemap_source_prunes_the_decoded_cache_to_its_budget(tmp_path: Path) -> None:
|
||
"""The sidecar half of the cache is bounded; the raw tiles it derives from are not."""
|
||
from meshterm.services.basemap import BasemapSource
|
||
|
||
cache = tmp_path / "cache"
|
||
decoded = cache / "decoded" / "14" / "1"
|
||
decoded.mkdir(parents=True)
|
||
for i in range(6):
|
||
blob = decoded / f"{i}.bin"
|
||
blob.write_bytes(b"x" * 1000)
|
||
os.utime(blob, (i, i)) # oldest first
|
||
|
||
# 6 KB present, 3 KB allowed: the three oldest go, the three newest stay.
|
||
src = BasemapSource(cache, tilejson_url="http://127.0.0.1:1/none", max_decoded_bytes=3000)
|
||
src._prune_decoded()
|
||
|
||
left = sorted(p.stem for p in decoded.glob("*.bin"))
|
||
assert left == ["3", "4", "5"], "the oldest sidecars go first"
|
||
|
||
|
||
def test_basemap_source_remembers_a_blank_tile_for_the_session(tmp_path: Path) -> None:
|
||
"""An answered-empty tile isn't re-requested this session (but isn't written down)."""
|
||
from meshterm.services import basemap as basemap_mod
|
||
|
||
src = _offline_source(tmp_path / "cache")
|
||
src._template = "http://tiles.invalid/{z}/{x}/{y}.pbf"
|
||
src._resolved = True
|
||
calls: list[str] = []
|
||
|
||
def _get(url: str):
|
||
calls.append(url)
|
||
return basemap_mod._Response(True, b"")
|
||
|
||
src._http_get = _get # type: ignore[method-assign]
|
||
assert src.load_tile(14, 1, 1) is None
|
||
assert src.load_tile(14, 1, 1) is None
|
||
assert len(calls) == 1
|
||
|
||
|
||
def test_basemap_http_get_separates_an_answer_from_silence(tmp_path: Path) -> None:
|
||
"""404 is an answer; a 500, and a body short of Content-Length, are not."""
|
||
import urllib.error
|
||
|
||
from meshterm.services.basemap import BasemapSource
|
||
|
||
src = BasemapSource(tmp_path / "cache")
|
||
|
||
class _Resp:
|
||
"""The slice of an ``http.client.HTTPResponse`` the fetcher touches."""
|
||
|
||
def __init__(self, body: bytes, declared: str | None) -> None:
|
||
self._body, self.headers = body, {"Content-Length": declared}
|
||
|
||
def read(self) -> bytes:
|
||
return self._body
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *_exc) -> bool:
|
||
return False
|
||
|
||
def _fake(result):
|
||
def _urlopen(_req, timeout=None): # noqa: ANN001
|
||
if isinstance(result, Exception):
|
||
raise result
|
||
return result
|
||
|
||
return _urlopen
|
||
|
||
def _get(result) -> tuple[bool, bytes]:
|
||
import urllib.request
|
||
|
||
saved = urllib.request.urlopen
|
||
urllib.request.urlopen = _fake(result) # type: ignore[assignment]
|
||
try:
|
||
return tuple(src._http_get("http://tiles.invalid/t"))
|
||
finally:
|
||
urllib.request.urlopen = saved # type: ignore[assignment]
|
||
|
||
err = lambda code: urllib.error.HTTPError("u", code, "", {}, None) # noqa: E731
|
||
assert _get(_Resp(b"tile-bytes", "10")) == (True, b"tile-bytes")
|
||
assert _get(_Resp(b"tile-b", "10")) == (False, b"") # truncated read
|
||
assert _get(_Resp(b"anything", None)) == (True, b"anything") # no declared length
|
||
assert _get(err(404)) == (True, b"") # "no tile here" — an answer
|
||
assert _get(err(500)) == (False, b"") # server fault — no answer
|
||
assert _get(err(429)) == (False, b"") # throttled — no answer
|
||
assert _get(TimeoutError("timed out")) == (False, b"")
|
||
|
||
|
||
# -- the tool -----------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture()
|
||
def ctx(tmp_path: Path):
|
||
"""A mock-backed application context with the plain (console) UI surface."""
|
||
from meshterm.context import AppContext
|
||
from meshterm.core.admin_store import AdminStore
|
||
from meshterm.core.config import Settings
|
||
from meshterm.core.device_store import DeviceStore
|
||
from meshterm.persistence.repository import Repository
|
||
from meshterm.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()
|
||
|
||
|
||
async def test_map_tool_static_render_plots_contacts(ctx) -> None:
|
||
"""The CLI path plots the device's located contacts and reports the breakdown.
|
||
|
||
The mock companion's contacts carry two located repeaters and one located leaf node
|
||
(a fourth, Alice, has no fix), so the contact list — not the observation history — is
|
||
what fills the map.
|
||
"""
|
||
# Seed an observation whose key matches a located contact, so its reception detail is
|
||
# merged onto that contact's marker.
|
||
run_id = ctx.repo.start_run("monitor", {})
|
||
ctx.repo.record_observation(
|
||
run_id,
|
||
Observation(node="a1b2c3d4", name="Yagi-Repeater", node_type=NODE_TYPE_REPEATER,
|
||
snr=6.0, observed_at=utcnow()),
|
||
)
|
||
result = await MapTool().run(ctx, {"static": True, "basemap": False})
|
||
assert result.summary["located"] == 3
|
||
assert result.summary["repeaters"] == 2
|
||
assert result.summary["nodes"] == 1
|
||
assert result.summary["self_located"] is False
|
||
out = ctx.console.file.getvalue()
|
||
assert out != ""
|
||
assert "1 pkts" in out # the seeded observation's detail merged onto the contact
|
||
|
||
|
||
def test_usable_fix_rejects_out_of_range_and_null_island() -> None:
|
||
"""A fix is usable only in valid lat/lon range and away from the 0/0 no-GPS sentinel."""
|
||
from meshterm.core.geo import usable_fix
|
||
|
||
assert usable_fix(45.5, -73.6) is True
|
||
assert usable_fix(90.0, 180.0) is True and usable_fix(-90.0, -180.0) is True # extremes
|
||
assert usable_fix(0.0, 0.0) is False # null island (a no-GPS companion reports this)
|
||
assert usable_fix(-97.0, -1041.97) is False # out of range (a real advert seen in the wild)
|
||
assert usable_fix(45.0, 181.0) is False and usable_fix(-91.0, -73.0) is False
|
||
|
||
|
||
async def test_gather_markers_drops_out_of_range_fix(ctx) -> None:
|
||
"""A node advertising nonsense coordinates is never plotted, only the real one is.
|
||
|
||
Some firmware reports an out-of-range fix (a companion has advertised ``lat -97,
|
||
lon -1042``); projecting it flings the view off the world, leaving the map an all-black
|
||
off-world frame. It is treated as no fix at all, the same guard the 0/0 null island gets.
|
||
"""
|
||
from meshterm.tools.map import gather_markers
|
||
|
||
run_id = ctx.repo.start_run("monitor", {})
|
||
ctx.repo.record_observation(run_id, Observation(
|
||
node="beefbeefbeef", name="BadFix", node_type=NODE_TYPE_REPEATER,
|
||
lat=-97.0, lon=-1041.97, observed_at=utcnow()))
|
||
ctx.repo.record_observation(run_id, Observation(
|
||
node="cafecafecafe", name="GoodFix", node_type=NODE_TYPE_REPEATER,
|
||
lat=45.5, lon=-73.6, observed_at=utcnow()))
|
||
labels = {m.label for m in await gather_markers(ctx)}
|
||
assert "GoodFix" in labels
|
||
assert "BadFix" not in labels
|
||
|
||
|
||
def test_repository_round_trips_map_view(ctx) -> None:
|
||
"""The saved map viewport persists and reads back; absent by default."""
|
||
assert ctx.repo.get_map_view() is None
|
||
ctx.repo.set_map_view(45.51, -73.57, 13)
|
||
lat, lon, zoom = ctx.repo.get_map_view()
|
||
assert (lat, lon) == pytest.approx((45.51, -73.57))
|
||
assert zoom == 13
|
||
# A later save overwrites the single stored view.
|
||
ctx.repo.set_map_view(40.0, -74.0, 9)
|
||
assert ctx.repo.get_map_view() == pytest.approx((40.0, -74.0, 9))
|
||
|
||
|
||
async def test_map_tool_reports_nothing_to_plot(ctx, monkeypatch) -> None:
|
||
"""With no located contacts and no located history the tool returns cleanly."""
|
||
async def _no_contacts(_ctx):
|
||
return []
|
||
|
||
monkeypatch.setattr("meshterm.tools.map._contacts", _no_contacts)
|
||
result = await MapTool().run(ctx, {"static": True, "basemap": False})
|
||
assert result.summary == {"located": 0}
|
||
|
||
|
||
async def test_gather_markers_drops_null_island_fixes(ctx, monkeypatch) -> None:
|
||
"""A node advertising a 0/0 fix (no GPS lock) is not plotted at null island.
|
||
|
||
Framing such a marker (filter to it, press Enter) would drop the map onto the empty
|
||
mid-Atlantic — a screen of solid water fill — so a both-near-zero fix is treated as no
|
||
fix at all, exactly as our own node's marker already is.
|
||
"""
|
||
from meshterm.core.models import Contact
|
||
from meshterm.tools.map import gather_markers
|
||
|
||
async def _contacts(_ctx):
|
||
return [
|
||
Contact(name="Real", public_key="aa" * 32, lat=45.50, lon=-73.57),
|
||
Contact(name="NoFix", public_key="bb" * 32, lat=0.0, lon=0.0),
|
||
]
|
||
|
||
monkeypatch.setattr("meshterm.tools.map._contacts", _contacts)
|
||
labels = [m.label for m in await gather_markers(ctx)]
|
||
assert "Real" in labels
|
||
assert "NoFix" not in labels # the null-island node is dropped
|
||
|
||
|
||
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 _loaded_tile() -> list[Layer]:
|
||
"""A stand-in for a decoded tile: truthy, which is all the budget cares about."""
|
||
return [Layer(name="water", extent=4096)]
|
||
|
||
|
||
def _map_screen_with_tiles(count: int) -> "MapScreen":
|
||
"""A rendered map screen carrying ``count`` tiles' worth of pan history."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14)
|
||
screen.render_body(80)
|
||
for i in range(count):
|
||
screen._tiles[(14, 9000 + i, 9000 + i)] = _loaded_tile()
|
||
return screen
|
||
|
||
|
||
def _budget(screen) -> int: # noqa: ANN001 - the screen's own arithmetic, mirrored
|
||
from meshterm.ui.map_screen import _MIN_TILE_CACHE, _TILE_CACHE_SCREENS
|
||
|
||
in_view = len(screen._viewport.tiles(14))
|
||
return max(_MIN_TILE_CACHE, in_view * _TILE_CACHE_SCREENS)
|
||
|
||
|
||
def test_map_tile_cache_stays_bounded_while_panning() -> None:
|
||
"""Panning far doesn't accumulate decoded tiles without limit.
|
||
|
||
One decoded tile is ~0.9 MB on the PicoCalc, which has ~100 MB in total, so an
|
||
unbounded cache turns a long pan into an out-of-memory kill.
|
||
"""
|
||
screen = _map_screen_with_tiles(200)
|
||
for _ in range(12):
|
||
screen.handle("right")
|
||
screen.render_body(80)
|
||
|
||
held = sum(1 for layers in screen._tiles.values() if layers)
|
||
assert held <= _budget(screen), f"{held} decoded tiles held, over budget"
|
||
|
||
|
||
def test_map_tile_cache_never_drops_what_is_on_screen() -> None:
|
||
"""Whatever the view needs survives the trim — eviction starts at the far end."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14)
|
||
screen.render_body(80)
|
||
visible = list(screen._viewport.tiles(14))
|
||
for t in visible:
|
||
screen._tiles[t] = _loaded_tile()
|
||
for i in range(200): # a long pan's worth of history arriving after them
|
||
screen._tiles[(14, 9000 + i, 9000 + i)] = _loaded_tile()
|
||
|
||
screen.render_body(80)
|
||
|
||
for t in visible:
|
||
assert screen._tiles.get(t), f"{t} is on screen and must not have been evicted"
|
||
|
||
|
||
def test_map_tile_cache_keeps_the_absent_tile_markers() -> None:
|
||
"""A ``None`` is the memory of having asked; it costs a slot, so it is never evicted.
|
||
|
||
Dropping one would buy nothing back and cost a re-request on the very next repaint.
|
||
"""
|
||
screen = _map_screen_with_tiles(200)
|
||
for i in range(30):
|
||
screen._tiles[(14, 100 + i, 100)] = None
|
||
|
||
screen.render_body(80)
|
||
|
||
absent = [k for k, v in screen._tiles.items() if v is None]
|
||
assert len(absent) == 30, "absent-tile markers were evicted along with the geometry"
|
||
|
||
|
||
class _CapturingSession(_StubSession):
|
||
"""Drives :func:`open_map` headless: captures the pushed screen, renders it once, then
|
||
replays a canned key sequence — so a test can assert on where the map opened and what
|
||
it did (or didn't) persist as the view moves."""
|
||
|
||
def __init__(self, cols: int, rows: int, keys: tuple[str, ...] = ()) -> None:
|
||
super().__init__(cols, rows)
|
||
self.keys = keys
|
||
self.screen = None
|
||
self.repainted = False
|
||
|
||
async def run_screen(self, screen): # noqa: ANN001, ANN201
|
||
self.screen = screen
|
||
screen.render_body(self._cols)
|
||
for key in self.keys:
|
||
screen.handle(key)
|
||
return None
|
||
|
||
def request_full_repaint(self) -> None:
|
||
self.repainted = True
|
||
|
||
|
||
def test_map_screen_renders_pans_zooms_and_resets() -> None:
|
||
"""The interactive screen fills its body, and wasd/zoom/reset move the viewport."""
|
||
from meshterm.core.geo import Viewport
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.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("right") # pan east → longitude increases
|
||
assert screen._viewport.center_lon > start.center_lon
|
||
screen.handle("up") # pan north → latitude increases
|
||
assert screen._viewport.center_lat > start.center_lat
|
||
|
||
z = screen._viewport.zoom
|
||
screen.handle("pageup")
|
||
assert screen._viewport.zoom == z + 1
|
||
screen.handle("pagedown")
|
||
assert screen._viewport.zoom == z
|
||
|
||
screen.handle("home") # 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)
|
||
|
||
# Letters never pan or zoom — they feed the find filter instead.
|
||
view = screen._viewport
|
||
screen.handle("text", "d")
|
||
assert screen._viewport is view and screen._filter == "d"
|
||
screen.handle("backspace")
|
||
|
||
# Offline source: nodes still render and both markers are present.
|
||
assert "▲" in _plain(screen.render_body(80)) and "●" in _plain(screen.render_body(80))
|
||
# The basemap's state is a status atom, so it rides the title's `·` chain (standing in
|
||
# for the scale) and leaves the footer a pure key line inside its 72-cell budget.
|
||
assert screen.title.endswith("· offline")
|
||
assert "offline" not in screen.footer_hint
|
||
assert cell_len(screen.footer_hint) <= 72
|
||
|
||
|
||
def test_map_screen_shift_pans_by_a_single_cell() -> None:
|
||
"""Holding Shift with an arrow pans finely, by one character cell."""
|
||
from meshterm.core.geo import lonlat_to_world, world_to_lonlat
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [MapMarker("A", 45.50, -73.60), MapMarker("B", 45.40, -73.50)]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
||
screen.render_body(80)
|
||
start = screen._viewport
|
||
|
||
# A coarse east pan (30% of the view) moves much further than a fine one.
|
||
screen.handle("right")
|
||
coarse = screen._viewport.center_lon - start.center_lon
|
||
|
||
screen._viewport = start
|
||
screen.handle("shift_right")
|
||
fine = screen._viewport.center_lon - start.center_lon
|
||
assert 0 < fine < coarse
|
||
|
||
# The fine step is exactly one cell (2 dots) east at the current zoom.
|
||
cx, cy = lonlat_to_world(start.center_lat, start.center_lon, start.zoom)
|
||
_, expected_lon = world_to_lonlat(cx + 2, cy, start.zoom)
|
||
assert screen._viewport.center_lon == pytest.approx(expected_lon)
|
||
|
||
|
||
def test_map_fkey_lane_names_its_three_destinations_and_the_zoom() -> None:
|
||
"""The map repurposes the nav keys, so its lane labels them — and offers no dead key."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
from meshterm.ui.tui.fkeys import action_for
|
||
|
||
screen = MapScreen(_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14)
|
||
lane = screen.fkey_lane
|
||
|
||
# PgUp/PgDn zoom here and Home reframes — the shared paging labels would all be lies,
|
||
# and "end" is bound to nothing at all, so no jump rides the zoom rocker's Shift bank.
|
||
# The zoom pair rises to the right, like every right-hand directional pair.
|
||
assert [pair.label if pair else None for pair in lane] == [
|
||
"Region", "You", "Frame", "Zoom -", "Zoom +",
|
||
]
|
||
assert action_for(lane, 1) == "home" and action_for(lane, 5) == "pageup"
|
||
assert action_for(lane, 2) == "locate" and action_for(lane, 3) == "frame"
|
||
# Two slots carry a Shift half along their own axis: You + homes in on us zoomed,
|
||
# Clear drops the find query. The other three Shift keys stay unbound.
|
||
assert action_for(lane, 7) == "locate_zoom" and action_for(lane, 8) == "clear_find"
|
||
assert all(action_for(lane, n) is None for n in (6, 9, 10))
|
||
# Region and the zoom always act; the other two only where they'd land somewhere —
|
||
# and each Shift half gates with its own axis (no self marker, no query typed).
|
||
assert [pair.enabled for pair in lane] == [True, False, False, True, True]
|
||
assert lane[1].opp_enabled is False and lane[2].opp_enabled is False
|
||
|
||
|
||
def test_map_locate_recenters_on_our_own_node_at_the_current_zoom() -> None:
|
||
"""``You`` moves the view to us and changes nothing else — not even the zoom."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [
|
||
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
||
MapMarker("Homestead", 45.40, -73.50, is_self=True),
|
||
]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
||
screen.render_body(80)
|
||
screen.handle("pageup") # zoom in one step, so the recentre has a zoom to preserve
|
||
zoom = screen._viewport.zoom
|
||
screen.handle("left") # and pan away from wherever we are
|
||
|
||
assert screen.fkey_lane[1].enabled is True # we're on the map, so You can act
|
||
screen.handle("locate")
|
||
assert screen._viewport.center_lat == pytest.approx(45.40)
|
||
assert screen._viewport.center_lon == pytest.approx(-73.50)
|
||
assert screen._viewport.zoom == zoom
|
||
|
||
# A mesh we aren't located in has nowhere to go: the chip dims and the key no-ops.
|
||
away = MapScreen(_StubSession(80, 24), markers[:1], _StubSource(), 14)
|
||
away.render_body(80)
|
||
before = away._viewport
|
||
away.handle("locate")
|
||
assert away.fkey_lane[1].enabled is False and away._viewport is before
|
||
|
||
|
||
def test_map_echoes_the_find_query_in_the_body_only_where_the_footer_is_gone() -> None:
|
||
"""The PicoCalc draws no hint line, so the query it carries moves into the body."""
|
||
from meshterm.platforms import PICOCALC, REGULAR, set_platform
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True)]
|
||
|
||
try:
|
||
set_platform(REGULAR)
|
||
desktop = MapScreen(_StubSession(72, 20), markers, _StubSource(), 14)
|
||
assert len(desktop.render_body(72)) == 20
|
||
for ch in "yul":
|
||
desktop.handle("text", ch)
|
||
# Unchanged: the footer already shows it, so the canvas keeps the whole body.
|
||
assert "/yul" not in _plain(desktop.render_body(72))
|
||
assert len(desktop.render_body(72)) == 20
|
||
assert "find: yul" in desktop.footer_hint
|
||
|
||
set_platform(PICOCALC)
|
||
device = MapScreen(_StubSession(53, 23), markers, _StubSource(), 14)
|
||
assert len(device.render_body(53)) == 23
|
||
for ch in "yul":
|
||
device.handle("text", ch)
|
||
lines = device.render_body(53)
|
||
# The echo overlays the canvas's *last* row — directly above the F-key lane —
|
||
# so the ground never shifts: same body height, same viewport, top anchor held.
|
||
assert _plain(lines[-1]).strip() == "/yul"
|
||
assert "/yul" not in _plain(lines[0])
|
||
assert len(lines) == 23
|
||
# Clearing the find hands the row back to the canvas, still without a reflow.
|
||
for _ in "yul":
|
||
device.handle("backspace")
|
||
assert len(device.render_body(53)) == 23
|
||
assert "/" not in _plain(device.render_body(53))
|
||
finally:
|
||
set_platform(REGULAR)
|
||
|
||
|
||
def test_map_shifted_vertical_arrows_survive_the_console_keymap(monkeypatch) -> None:
|
||
"""Shift+↑ arrives as PgUp on the PicoCalc console; with Shift physically down it
|
||
must fine-pan, not zoom — and plain PgUp (the F-lane's zoom) stays a zoom."""
|
||
from meshterm.services import modifier_watch
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
screen = MapScreen(
|
||
_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14
|
||
)
|
||
screen.render_body(80)
|
||
zoom = screen._viewport.zoom
|
||
lat = screen._viewport.center_lat
|
||
|
||
monkeypatch.setattr(modifier_watch, "_shift_down", True)
|
||
screen.handle("pageup") # the keymap's Shift+↑, rescued by the watcher
|
||
assert screen._viewport.zoom == zoom # no zoom happened
|
||
assert screen._viewport.center_lat > lat # …the view nudged north instead
|
||
|
||
monkeypatch.setattr(modifier_watch, "_shift_down", False)
|
||
screen.handle("pageup") # a real zoom press (the F5 chip's action)
|
||
assert screen._viewport.zoom == zoom + 1
|
||
|
||
|
||
def test_map_frame_chip_lights_only_with_matches_to_frame() -> None:
|
||
"""``Frame`` is the find's Enter under another name — dim, and inert, without a query."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [
|
||
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
||
MapMarker("Alice", 45.40, -73.50),
|
||
]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
||
screen.render_body(80)
|
||
before = screen._viewport
|
||
screen.handle("frame") # no query typed: nothing to frame, so nothing moves
|
||
assert screen.fkey_lane[2].enabled is False and screen._viewport is before
|
||
|
||
for ch in "ali":
|
||
screen.handle("text", ch)
|
||
assert screen.fkey_lane[2].enabled is True
|
||
screen.handle("frame")
|
||
assert screen._viewport.center_lat == pytest.approx(45.40)
|
||
|
||
# A query nothing matches has no extent either — the chip goes back to dim.
|
||
for ch in "zzz":
|
||
screen.handle("text", ch)
|
||
assert screen._matches() == [] and screen.fkey_lane[2].enabled is False
|
||
|
||
|
||
def test_map_screen_find_filters_frames_and_clears() -> None:
|
||
"""Typing builds the query; ^Enter frames matches; Enter/Esc drop it; Esc then leaves."""
|
||
import asyncio
|
||
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [
|
||
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
||
MapMarker("Alice", 45.40, -73.50),
|
||
MapMarker("Yagi-North", 45.60, -73.65),
|
||
]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
||
screen.render_body(80)
|
||
|
||
for ch in "yu":
|
||
screen.handle("text", ch)
|
||
assert screen._filter == "yu"
|
||
assert [m.label for m in screen._matches()] == ["YUL-Cartierville"]
|
||
screen.render_body(80)
|
||
assert "1 of 3 match" in screen.title
|
||
assert "find: yu" in screen.footer_hint
|
||
|
||
# Only the match keeps a label; the others dim to bare context glyphs.
|
||
body = _plain(screen.render_body(80))
|
||
assert "YUL-Cartierville" in body
|
||
assert "Alice" not in body and "Yagi-North" not in body
|
||
|
||
# ^Enter frames the matches: the view centres on YUL, and the query survives it.
|
||
screen.handle("ctrl_enter")
|
||
assert screen._viewport.center_lat == pytest.approx(45.53, abs=0.05)
|
||
assert screen._viewport.center_lon == pytest.approx(-73.71, abs=0.05)
|
||
assert screen._filter == "yu"
|
||
|
||
# Plain Enter is the other way out: the query goes, the view stays exactly where the
|
||
# frame left it.
|
||
centre = (screen._viewport.center_lat, screen._viewport.center_lon)
|
||
screen.handle("enter")
|
||
assert screen._filter == ""
|
||
assert (screen._viewport.center_lat, screen._viewport.center_lon) == centre
|
||
|
||
# Backspace edits; Esc clears the filter first and only then dismisses.
|
||
for ch in "yu":
|
||
screen.handle("text", ch)
|
||
screen.handle("backspace")
|
||
assert screen._filter == "y"
|
||
|
||
async def drive() -> object:
|
||
screen.future = asyncio.get_running_loop().create_future()
|
||
screen.handle("escape")
|
||
assert screen._filter == "" and not screen.future.done()
|
||
screen.handle("escape")
|
||
return await screen.future
|
||
|
||
assert asyncio.run(drive()) is None
|
||
|
||
|
||
def test_map_screen_frame_zooms_in_on_a_single_match() -> None:
|
||
"""^Enter on a lone filtered node homes in close on it, not the fit's neutral default."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import _FIND_ZOOM, MapScreen
|
||
|
||
markers = [
|
||
MapMarker("YUL-Cartierville", 45.53, -73.71, is_repeater=True),
|
||
MapMarker("Alice", 45.40, -73.50),
|
||
]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 19) # source max 19
|
||
screen.render_body(80)
|
||
for ch in "yul":
|
||
screen.handle("text", ch)
|
||
assert [m.label for m in screen._matches()] == ["YUL-Cartierville"]
|
||
screen.handle("ctrl_enter")
|
||
assert screen._viewport.center_lat == pytest.approx(45.53, abs=0.01)
|
||
assert screen._viewport.center_lon == pytest.approx(-73.71, abs=0.01)
|
||
assert screen._viewport.zoom == _FIND_ZOOM # zoomed in, not the fit's default 14
|
||
|
||
|
||
def test_map_screen_restores_and_persists_view() -> None:
|
||
"""The screen reopens on a saved view and reports every centre/zoom change."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
saved: list[tuple[float, float, int]] = []
|
||
markers = [MapMarker("A", 45.50, -73.60), MapMarker("B", 45.40, -73.50)]
|
||
screen = MapScreen(
|
||
_StubSession(80, 24), markers, _StubSource(), 14,
|
||
saved_view=(46.80, -71.20, 12),
|
||
on_view_change=lambda vp: saved.append(
|
||
(vp.center_lat, vp.center_lon, vp.zoom)
|
||
),
|
||
)
|
||
|
||
screen.render_body(80)
|
||
# Restored to the saved centre/zoom, not a fit of the markers.
|
||
assert screen._viewport.zoom == 12
|
||
assert screen._viewport.center_lat == pytest.approx(46.80)
|
||
assert screen._viewport.center_lon == pytest.approx(-71.20)
|
||
assert saved == [] # reopening unchanged doesn't rewrite the stored view
|
||
|
||
screen.handle("right") # pan east persists the moved view
|
||
assert saved and saved[-1][2] == 12 # zoom unchanged
|
||
assert saved[-1][1] > -71.20 # centre moved east
|
||
|
||
screen.handle("pageup") # zooming in persists too
|
||
assert saved[-1][2] == 13
|
||
|
||
|
||
async def test_open_map_focus_centres_on_the_node_without_clobbering_the_saved_view(
|
||
ctx, monkeypatch
|
||
) -> None:
|
||
"""Opening the full map focused on a node (from its detail page) centres there, at the
|
||
inline preview's zoom — and, being a transient peek, never overwrites the persisted
|
||
'where you left the map' global view, even as the peek is panned and zoomed."""
|
||
from meshterm.core.geo import clamp_lat
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import open_map
|
||
from meshterm.ui.surface import TuiUi
|
||
|
||
monkeypatch.setattr("meshterm.ui.map_screen.basemap_source", lambda _c: _StubSource())
|
||
ctx.repo.set_map_view(10.0, 20.0, 8) # where the user last left the global map
|
||
|
||
session = _CapturingSession(80, 24, keys=("right", "pageup")) # pan + zoom the peek
|
||
ctx.ui = TuiUi(session)
|
||
await open_map(
|
||
ctx,
|
||
[MapMarker("Hub", 45.5, -73.6, is_repeater=True)],
|
||
focus=(45.5, -73.6),
|
||
find="Hub",
|
||
)
|
||
|
||
# Opened on the node, at the detail preview's zoom (min(13, max_zoom)).
|
||
assert session.screen._saved_view == (clamp_lat(45.5), -73.6, 13)
|
||
assert session.screen._on_view_change is None # a focused peek wires no persistence
|
||
assert session.screen._filter == "Hub" # the find opens seeded to the focused node
|
||
# Panning/zooming the peek left the global 'where you left the map' view untouched.
|
||
assert ctx.repo.get_map_view() == pytest.approx((10.0, 20.0, 8))
|
||
assert session.repainted # cleaned the terminal on the way out, like the plain open
|
||
|
||
|
||
async def test_open_map_without_focus_restores_and_persists_the_global_view(
|
||
ctx, monkeypatch
|
||
) -> None:
|
||
"""With no focus the full map is unchanged: it reopens where the user left it and pans
|
||
persist straight back to that global view."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import open_map
|
||
from meshterm.ui.surface import TuiUi
|
||
|
||
monkeypatch.setattr("meshterm.ui.map_screen.basemap_source", lambda _c: _StubSource())
|
||
ctx.repo.set_map_view(46.80, -71.20, 12) # the saved view to reopen on
|
||
|
||
session = _CapturingSession(80, 24, keys=("right",)) # pan east
|
||
ctx.ui = TuiUi(session)
|
||
await open_map(ctx, [MapMarker("Hub", 46.8, -71.2)])
|
||
|
||
assert session.screen._viewport.zoom == 12 # reopened on the saved view, not a node fit
|
||
lat, lon, zoom = ctx.repo.get_map_view()
|
||
assert zoom == 12 and lon > -71.20 # the eastward pan persisted back to the global view
|
||
|
||
|
||
def test_map_screen_seeded_find_behaves_like_a_typed_one() -> None:
|
||
"""A seeded find matches, titles, and clears exactly as if the user had typed it."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [MapMarker("Hub", 45.5, -73.6), MapMarker("Far", 45.6, -73.5)]
|
||
screen = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14, find="Hub")
|
||
assert [m.label for m in screen._matches()] == ["Hub"] # only the seed's node lights
|
||
assert "find: Hub" in screen.footer_hint # the query shows, editable, as ever
|
||
screen.handle("escape") # first Esc clears the find (not the map)
|
||
assert screen._filter == "" and len(screen._matches()) == 2
|
||
|
||
|
||
def test_map_screen_scrubs_right_edge_after_move() -> None:
|
||
"""A move asks the session to scrub the right edge once, to clear any braille smear."""
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.ui.map_screen import MapScreen
|
||
|
||
markers = [MapMarker("A", 45.5, -73.6)]
|
||
braille = MapScreen(_StubSession(80, 24), markers, _StubSource(), 14)
|
||
braille.render_body(80)
|
||
braille.handle("right") # a move flags the edge for a scrub
|
||
assert braille.consume_edge_scrub() == 2 # right padding + border columns
|
||
assert braille.consume_edge_scrub() == 0 # consumed — not repeated without another move
|
||
|
||
|
||
def test_map_screen_escape_dismisses() -> None:
|
||
"""Esc resolves the screen's future with None (backs out to the menu)."""
|
||
import asyncio
|
||
|
||
from meshterm.ui.map_render import MapMarker
|
||
from meshterm.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
|
||
|
||
|
||
def test_render_map_labels_take_the_name_hue_ours_white() -> None:
|
||
"""Node labels carry their key-derived hue; our own label is white while the
|
||
★ glyph stays yellow; a keyless label lands on the muted grey."""
|
||
from meshterm.ui.map_render import _SELF, MapMarker, render_map
|
||
from meshterm.ui.widgets import name_rgb
|
||
|
||
vp = Viewport.fit([(45.5, -73.6), (45.4, -73.5)], 120, 80, max_zoom=14)
|
||
markers = [
|
||
MapMarker("US", 45.5, -73.6, is_self=True, key="cc" * 32),
|
||
MapMarker("KEYED", 45.4, -73.5, key="d4" * 32),
|
||
MapMarker("BARE", 45.45, -73.55),
|
||
]
|
||
lines = render_map(vp, {}, markers)
|
||
assert _glyph_color(lines, "★") == parse_hex(_SELF[1]) # the glyph keeps its yellow
|
||
assert _glyph_color(lines, "US") == (255, 255, 255) # ...the label goes you-white
|
||
assert _glyph_color(lines, "KEYED") == name_rgb("KEYED", "d4" * 32)
|
||
assert _glyph_color(lines, "BARE") == (148, 163, 184) # no key, the muted grey
|
||
|
||
|
||
def test_render_map_find_matches_still_label_white() -> None:
|
||
"""An active find filter keeps forcing matching labels to full white."""
|
||
from meshterm.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("TARGET", 45.5, -73.6, key="d4" * 32)]
|
||
lines = render_map(vp, {}, markers, find="targ")
|
||
assert _glyph_color(lines, "TARGET") == (255, 255, 255)
|
||
|
||
|
||
# --- the ground raster runs off the paint path ---------------------------------------
|
||
|
||
|
||
def _async_map(monkeypatch, drawn: list):
|
||
"""A map screen whose rasters are captured instead of run, with a live event loop."""
|
||
from meshterm.ui import map_screen as ms
|
||
from meshterm.ui.map_render import MapMarker
|
||
|
||
monkeypatch.setattr(ms, "_loop_running", lambda: True)
|
||
started: list[tuple] = []
|
||
monkeypatch.setattr(
|
||
ms.asyncio, "ensure_future", lambda coro: (coro.close(), started.append(coro))[0]
|
||
)
|
||
screen = ms.MapScreen(
|
||
_StubSession(80, 24), [MapMarker("A", 45.5, -73.6)], _StubSource(), 14
|
||
)
|
||
drawn.append(started)
|
||
return screen
|
||
|
||
|
||
def test_map_paints_without_waiting_for_the_ground(monkeypatch) -> None: # noqa: ANN001
|
||
"""A pan must answer the keystroke now — rasterizing takes ~1 s on the PicoCalc."""
|
||
from meshterm.ui import map_screen as ms
|
||
|
||
calls: list[tuple] = []
|
||
real = ms.render_map
|
||
monkeypatch.setattr(ms, "render_map", lambda vp, tiles, m, **k: calls.append(tiles) or real(vp, tiles, m, **k))
|
||
started: list = []
|
||
screen = _async_map(monkeypatch, started)
|
||
|
||
screen.render_body(80)
|
||
# Whatever ran on the paint path drew no tiles: the ground is somebody else's job.
|
||
assert calls, "the paint drew nothing at all"
|
||
assert all(tiles == {} for tiles in calls), calls
|
||
assert screen._drawing is not None, "no background raster was scheduled"
|
||
|
||
|
||
def test_map_keeps_one_raster_in_flight_while_panning(monkeypatch) -> None: # noqa: ANN001
|
||
"""A held arrow key must not queue a second of thread work per keypress."""
|
||
started: list = []
|
||
screen = _async_map(monkeypatch, started)
|
||
screen.render_body(80)
|
||
scheduled = len(started[0])
|
||
|
||
for _ in range(8):
|
||
screen.handle("right")
|
||
screen.render_body(80)
|
||
|
||
assert len(started[0]) == scheduled, "a second raster started before the first finished"
|
||
assert screen._wanted is not None, "the newest view was not remembered for next"
|
||
|
||
|
||
def test_map_holds_an_aligned_frame_while_a_tile_lands(monkeypatch) -> None: # noqa: ANN001
|
||
"""A tile arriving must not blank streets that are still in the right place.
|
||
|
||
The view has not moved, so the finished raster is still correctly aligned — it is only
|
||
missing the newcomer's detail. Redrawing from nothing would flash the ground away for
|
||
the second it takes to draw the same picture again.
|
||
"""
|
||
started: list = []
|
||
screen = _async_map(monkeypatch, started)
|
||
screen.render_body(80)
|
||
# Pretend the scheduled raster finished.
|
||
screen._frame = ["ground"] * 20
|
||
screen._frame_key = screen._ground_key(screen._viewport)
|
||
screen._drawing = None
|
||
assert screen.render_body(80) == ["ground"] * 20
|
||
|
||
in_view = screen._viewport.tiles(14)[0]
|
||
screen._tiles[in_view] = _loaded_tile() # a tile lands, view unmoved
|
||
assert screen.render_body(80) == ["ground"] * 20, "dropped an aligned frame"
|
||
assert screen._drawing is not None, "did not redraw for the new tile"
|
||
|
||
|
||
def test_map_drops_a_stale_frame_once_the_view_moves(monkeypatch) -> None: # noqa: ANN001
|
||
"""Streets drawn for somewhere else are a lie about where you are looking."""
|
||
started: list = []
|
||
screen = _async_map(monkeypatch, started)
|
||
screen.render_body(80)
|
||
screen._frame = ["ground"] * 20
|
||
screen._frame_key = screen._ground_key(screen._viewport)
|
||
screen._drawing = None
|
||
|
||
screen.handle("right")
|
||
assert screen.render_body(80) != ["ground"] * 20, "kept a misaligned frame"
|
||
|
||
|
||
async def test_map_pans_on_the_ground_its_last_raster_left(monkeypatch) -> None: # noqa: ANN001
|
||
"""A misaligned frame is dropped, but the ground it drew is kept and reprojected.
|
||
|
||
It is the only ground anyone has until the next raster lands, and a pan that answers
|
||
with an empty canvas blanks the map to black between every keypress.
|
||
"""
|
||
from meshterm.ui import map_screen as ms
|
||
|
||
started: list = []
|
||
screen = _async_map(monkeypatch, started)
|
||
screen.render_body(80)
|
||
|
||
await screen._draw_ground(screen._ground_key(screen._viewport), screen._viewport, {})
|
||
assert screen._ghost is not None, "the finished raster left no ground behind"
|
||
assert screen._ghost.viewport == screen._viewport
|
||
|
||
handed: list = []
|
||
real = ms.render_map
|
||
monkeypatch.setattr(
|
||
ms, "render_map",
|
||
lambda vp, tiles, m, **k: handed.append(k.get("ghost")) or real(vp, tiles, m, **k),
|
||
)
|
||
screen.handle("right")
|
||
screen.render_body(80)
|
||
assert handed and handed[-1] is screen._ghost, "the pan painted on an empty canvas"
|
||
|
||
|
||
def test_each_road_class_keeps_its_own_colour_through_the_per_tile_style_memo() -> None:
|
||
"""The tile's palette is resolved once per class, not once per feature.
|
||
|
||
``mark_rgb`` was reached ~30 000 times a frame and ``Feature.get`` ~32 000 — both
|
||
answering the same handful of questions. Hoisting them behind a per-tile memo is only
|
||
sound while every class still resolves to exactly the colour it did: a motorway must
|
||
not inherit the footway's grey because it drew second.
|
||
"""
|
||
from meshterm.core.mvt import GEOM_LINE, Feature, Layer
|
||
from meshterm.ui.map_render import (
|
||
_RAIL,
|
||
_ROAD_DEFAULT,
|
||
_ROAD_STYLE,
|
||
_draw_tile,
|
||
_Frame,
|
||
)
|
||
from meshterm.ui.mapcanvas import MapCanvas
|
||
from meshterm.ui.theme import mark_rgb
|
||
|
||
classes = ["motorway", "footway", "residential", "rail", "no_such_class", None]
|
||
features = [
|
||
Feature(
|
||
geom_type=GEOM_LINE,
|
||
# One short horizontal run per class, each on its own row of the tile.
|
||
rings=[[(200, 500 + i * 500), (3900, 500 + i * 500)]],
|
||
tags={} if cls is None else {"class": cls},
|
||
)
|
||
for i, cls in enumerate(classes)
|
||
]
|
||
# Wide enough that the whole tile lands on the canvas, so every class gets drawn.
|
||
vp = Viewport(45.5019, -73.5674, 14, 512, 512)
|
||
canvas = MapCanvas(vp.dot_w // 2, vp.dot_h // 4)
|
||
frame = _Frame(canvas=canvas, viewport=vp)
|
||
_draw_tile(
|
||
frame, [Layer(name="transportation", extent=4096, features=features)],
|
||
14, 4843, 5861,
|
||
)
|
||
|
||
painted = {colour for row in canvas._color for colour in row if colour}
|
||
for cls in classes:
|
||
if cls == "rail":
|
||
expected = _RAIL
|
||
else:
|
||
expected = _ROAD_STYLE.get(cls or "", _ROAD_DEFAULT)
|
||
assert mark_rgb(expected[0]) in painted, f"{cls} lost its own colour"
|