fix(coverage): round derived coords to 6dp; skip roadless traffic cells (#66)

USGS rejects bBox coords with >7 decimals (raw Leaflet clicks have 14) —
round all coverage-derived coordinates to 6dp so USGS/others accept them.
TomTom flow 400 ("Point too far from nearest existing segment") on rural
grid cells is expected no-data, not an error — log debug and skip instead
of warning. Fix the fires log to not claim "in US-ID" under coverage mode.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-06 11:15:56 -06:00 committed by GitHub
commit 8125ba0978
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 136 additions and 7 deletions

View file

@ -73,9 +73,14 @@ def bbox_intersects(a, b) -> bool:
def centroid(bbox) -> tuple[float, float]:
"""Return the (lat, lon) center of *bbox* = [W, S, E, N]."""
"""Return the (lat, lon) center of *bbox* = [W, S, E, N].
Coordinates are rounded to 6 decimal places (0.11 m precision) so that
values derived from high-precision Leaflet clicks never exceed external API
decimal-place limits.
"""
west, south, east, north = bbox
return ((south + north) / 2.0, (west + east) / 2.0)
return (round((south + north) / 2.0, 6), round((west + east) / 2.0, 6))
def grid_points(bbox, cols: int = 3, rows: int = 3) -> list[tuple[float, float]]:
@ -95,7 +100,7 @@ def grid_points(bbox, cols: int = 3, rows: int = 3) -> list[tuple[float, float]]
lat = south + (row + 0.5) * lat_step
for col in range(cols):
lon = west + (col + 0.5) * lon_step
points.append((lat, lon))
points.append((round(lat, 6), round(lon, 6)))
return points
@ -323,7 +328,9 @@ def resolve_adapter_coverage(
if adapter in _GLOBAL_ADAPTERS:
return None
bbox = list(coverage_bbox) # defensive copy
# Defensive copy + precision cap: raw Leaflet clicks produce 14+ decimal
# places which USGS (and others) reject. 6dp ≈ 0.11 m — more than enough.
bbox = [round(float(c), 6) for c in coverage_bbox]
if adapter in ("fires", "wfigs", "nicf"):
return {

View file

@ -195,7 +195,8 @@ class NICFFiresAdapter:
self._is_loaded = True
if changed:
logger.info(f"NIFC fires updated: {len(new_events)} active in {self._state}")
loc = "the coverage area" if self._coverage is not None else self._state
logger.info(f"NIFC fires updated: {len(new_events)} active in {loc}")
return changed

View file

@ -162,6 +162,15 @@ class TomTomTrafficAdapter:
if e.code == 401 or e.code == 403:
logger.error(f"TomTom auth error: {e.code} - check API key")
self._last_error = f"Auth error {e.code}"
self._consecutive_errors += 1
elif e.code == 400:
# "Point too far from nearest existing segment" — expected for
# wilderness/rural grid cells with no TomTom road data.
# This is not an error; the cell simply has no data.
logger.debug(
"TomTom: no road segment near %s (%.5f,%.5f) — skipping",
name, lat, lon,
)
else:
logger.warning(f"TomTom HTTP error for {name}: {e.code}")
self._last_error = f"HTTP {e.code}"

View file

@ -1,7 +1,8 @@
"""Tests for TomTom traffic adapter Phase 2.7 — to_event() method."""
import time
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from urllib.error import HTTPError
import pytest
@ -203,3 +204,48 @@ def test_to_event_missing_properties_returns_none(adapter):
def test_to_event_does_not_raise_on_corrupted_dict(adapter):
"""Corrupted dict returns None without raising."""
assert adapter.to_event({"garbage": True}) is None
# ============================================================
# _fetch_point 400 — roadless cell is no-data, not an error
# ============================================================
def test_fetch_point_400_does_not_set_last_error(mock_config):
"""A 400 HTTPError (roadless cell) must not set _last_error or increment
consecutive_errors it is expected no-data, not a failure."""
adapter = TomTomTrafficAdapter(mock_config)
assert adapter._last_error is None
assert adapter._consecutive_errors == 0
err = HTTPError(
url="https://api.tomtom.com/...",
code=400,
msg="Bad Request",
hdrs=None,
fp=None,
)
with patch("meshai.env.traffic.urlopen", side_effect=err):
result = adapter._fetch_point("wilderness_cell", 43.5, -115.0, 0.0)
assert result is None, "400 must return None (no data)"
assert adapter._last_error is None, "_last_error must not be set on 400"
assert adapter._consecutive_errors == 0, "consecutive_errors must not increment on 400"
def test_fetch_point_non400_http_error_sets_last_error(mock_config):
"""Non-400/401/403 HTTP errors (e.g. 503) must still set _last_error."""
adapter = TomTomTrafficAdapter(mock_config)
err = HTTPError(
url="https://api.tomtom.com/...",
code=503,
msg="Service Unavailable",
hdrs=None,
fp=None,
)
with patch("meshai.env.traffic.urlopen", side_effect=err):
result = adapter._fetch_point("some_corridor", 43.5, -116.0, 0.0)
assert result is None
assert adapter._last_error == "HTTP 503"
assert adapter._consecutive_errors == 1

View file

@ -473,3 +473,69 @@ def test_resolve_does_not_mutate_input_bbox():
result = resolve_adapter_coverage("usgs_quake", original)
result["bbox"].append(999) # mutate return value
assert len(original) == 4 # original untouched
# ===========================================================================
# Coordinate precision — all outputs must be ≤6 decimal places (Fix 1)
# ===========================================================================
# Raw Leaflet map-click bbox: ~14 decimal places, triggers USGS HTTP 400.
HIGH_PREC_BOX = [
-116.99340820312501,
41.95949009892467,
-110.98388671875001,
44.09547572946637,
]
def _decimal_places(v) -> int:
"""Return significant decimal places in *v* (trailing zeros not counted)."""
s = f"{float(v):.10f}"
decimal_part = s.split(".")[1]
stripped = decimal_part.rstrip("0")
return len(stripped) if stripped else 0
def _assert_max_6dp(v, label: str = "") -> None:
"""Assert float *v* has at most 6 decimal places."""
n = _decimal_places(v)
assert n <= 6, f"{label}: {v!r} has {n} decimal places (max 6)"
def test_high_precision_bbox_coords_rounded():
"""bbox key in every adapter that returns one must have ≤6dp per coordinate."""
bbox_adapters = ("usgs_quake", "firms", "roads511", "usgs", "nws", "wzdx",
"fires", "wfigs", "nicf")
for adapter in bbox_adapters:
result = resolve_adapter_coverage(adapter, HIGH_PREC_BOX)
assert result is not None, f"{adapter} unexpectedly returned None"
if "bbox" in result:
for i, coord in enumerate(result["bbox"]):
_assert_max_6dp(coord, f"{adapter}.bbox[{i}]")
def test_high_precision_envelope_geometry_rounded():
"""ArcGIS envelope geometry string coords must have ≤6dp (fires adapter)."""
result = resolve_adapter_coverage("fires", HIGH_PREC_BOX)
assert result is not None
geom = result["envelope"]["geometry"]
for part in geom.split(","):
_assert_max_6dp(part.strip(), f"fires.envelope.geometry part {part!r}")
def test_high_precision_centroid_rounded():
"""centroid() computed from a high-precision bbox must have ≤6dp per coord."""
result = resolve_adapter_coverage("satpass", HIGH_PREC_BOX)
assert result is not None
lat, lon = result["centroid"]
_assert_max_6dp(lat, "satpass.centroid.lat")
_assert_max_6dp(lon, "satpass.centroid.lon")
def test_high_precision_grid_points_rounded():
"""All grid_points() coords from a high-precision bbox must have ≤6dp."""
result = resolve_adapter_coverage("traffic", HIGH_PREC_BOX)
assert result is not None
for i, (plat, plon) in enumerate(result["points"]):
_assert_max_6dp(plat, f"traffic.points[{i}].lat")
_assert_max_6dp(plon, f"traffic.points[{i}].lon")