diff --git a/app/live_layers.py b/app/live_layers.py index 3e214c6..fc24256 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -58,7 +58,9 @@ _COMPASS = { } _cache: dict[str, tuple[float, Any]] = {} -_cache_lock = asyncio.Lock() +_key_locks: dict[str, asyncio.Lock] = {} +_key_locks_guard = asyncio.Lock() +_QUANT = 0.25 # degrees — pan jitter inside a cell reuses the TTL entry # Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker. vessel_last_known: dict[str, dict] = {} @@ -116,6 +118,33 @@ def parse_bbox(bbox: str) -> tuple[float, float, float, float]: return minlon, minlat, maxlon, maxlat +def quantize_bbox( + minlon: float, minlat: float, maxlon: float, maxlat: float, + step: float = _QUANT, +) -> tuple[float, float, float, float]: + """Snap a viewport to a coarse cell so nearby pans share a cache key. + + The returned envelope is expanded to cover the original box. + """ + def q_down(v: float, lo: float, hi: float) -> float: + v = max(lo, min(hi, v)) + return math.floor(v / step) * step + + return ( + round(q_down(minlon, -180.0, 180.0), 4), + round(q_down(minlat, -90.0, 90.0), 4), + round(q_down(maxlon, -180.0, 180.0) + step, 4), + round(q_down(maxlat, -90.0, 90.0) + step, 4), + ) + + +def bbox_cell_key(bbox: str | None) -> str: + """Stable cache-key fragment for a viewport (or 'all').""" + if not bbox: + return "all" + return ",".join(f"{v:.4f}" for v in quantize_bbox(*parse_bbox(bbox))) + + def bbox_center_radius_nm( minlon: float, minlat: float, maxlon: float, maxlat: float, ) -> tuple[float, float, int]: @@ -404,12 +433,22 @@ def _headers() -> dict[str, str]: return {"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"} +async def _lock_for(key: str) -> asyncio.Lock: + async with _key_locks_guard: + lock = _key_locks.get(key) + if lock is None: + lock = asyncio.Lock() + _key_locks[key] = lock + return lock + + async def _ttl_get(key: str, ttl: float, factory: Callable[[], Awaitable[Any]]) -> Any: now = time.monotonic() hit = _cache.get(key) if hit and now - hit[0] < ttl: return hit[1] - async with _cache_lock: + lock = await _lock_for(key) + async with lock: hit = _cache.get(key) if hit and time.monotonic() - hit[0] < ttl: return hit[1] @@ -428,7 +467,8 @@ async def _get_json(url: str, params: dict | None = None) -> Any: async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) - lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat) + qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(minlon, minlat, maxlon, maxlat) + lat, lon, radius = bbox_center_radius_nm(qminlon, qminlat, qmaxlon, qmaxlat) cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}" async def _load(): @@ -493,7 +533,7 @@ def _wfigs_params(bbox: str | None) -> dict: "resultRecordCount": 2000, } if bbox: - minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + minlon, minlat, maxlon, maxlat = quantize_bbox(*parse_bbox(bbox)) params["geometry"] = f"{minlon},{minlat},{maxlon},{maxlat}" params["geometryType"] = "esriGeometryEnvelope" params["inSR"] = "4326" @@ -511,7 +551,7 @@ async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) -> async def _load(): return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params)) - rows = await _ttl_get(f"wfigs:inc:{bbox or 'all'}", 600.0, _load) + rows = await _ttl_get(f"wfigs:inc:{bbox_cell_key(bbox)}", 600.0, _load) return rows[:limit] @@ -525,7 +565,7 @@ async def fetch_fire_perimeters(bbox: str | None) -> dict: async def _load(): return await _get_json(WFIGS_PERIMETERS, params) - fc = await _ttl_get(f"wfigs:per:{bbox or 'all'}", 600.0, _load) + fc = await _ttl_get(f"wfigs:per:{bbox_cell_key(bbox)}", 600.0, _load) if not isinstance(fc, dict): return {"type": "FeatureCollection", "features": []} return fc @@ -539,7 +579,7 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict: if area: nws_params["area"] = area.upper() elif bbox: - minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + minlon, minlat, maxlon, maxlat = quantize_bbox(*parse_bbox(bbox)) nws_params["bbox"] = f"{minlon},{minlat},{maxlon},{maxlat}" nws_fc: dict = {"features": []} sbw_fc: dict = {"features": []} @@ -571,7 +611,7 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict: features.append(feat) return {"type": "FeatureCollection", "features": features} - key = f"alerts:{area or ''}:{bbox or ''}" + key = f"alerts:{area or ''}:{bbox_cell_key(bbox) if bbox else ''}" return await _ttl_get(key, 30.0, _load) diff --git a/tests/test_live_layers.py b/tests/test_live_layers.py index 2214a5c..ba999d1 100644 --- a/tests/test_live_layers.py +++ b/tests/test_live_layers.py @@ -5,6 +5,7 @@ from live_layers import ( bbox_center_radius_nm, filter_points_bbox, parse_bbox, + quantize_bbox, rainviewer_tile_url, to_marker, transform_adsb_lol, @@ -12,6 +13,8 @@ from live_layers import ( transform_amtraker, transform_nhc_storms, transform_wfigs_incidents, + _cache, + _ttl_get, ) from camera_scraper import parse_caltrans_json @@ -244,3 +247,43 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls(): assert "I-80" in cam["location_name"] assert "rtsp://" not in cam["source_url"].lower() assert "rtsp://" not in cam["snapshot_url"].lower() + + +def test_quantize_bbox_stable_under_jitter(): + a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102")) + b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090")) + assert a == b + minlon, minlat, maxlon, maxlat = a + assert minlon <= -78.7912 + assert minlat <= 35.7700 + assert maxlon >= -78.6101 + assert maxlat >= 35.9102 + + +def test_ttl_get_does_not_block_other_keys(): + import asyncio + + _cache.clear() + order = [] + + async def slow(): + order.append("slow-start") + await asyncio.sleep(0.2) + order.append("slow-end") + return "S" + + async def fast(): + order.append("fast") + return "F" + + async def run(): + t1 = asyncio.create_task(_ttl_get("slow", 5, slow)) + await asyncio.sleep(0.01) + t2 = asyncio.create_task(_ttl_get("fast", 5, fast)) + await asyncio.gather(t1, t2) + + asyncio.run(run()) + assert order.index("fast") < order.index("slow-end") + assert _cache["slow"][1] == "S" + assert _cache["fast"][1] == "F" + _cache.clear()