mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-10-30 17:42:50 +00:00
agessaman
987df9b124
A no-coverage reply (400/404) from the NWS alerts API set a single flag on wx and on the weather service that skipped every later alert request for the rest of the run, so one wx request for a city abroad silently removed alerts from every US reply after it. NWSNoCoverageCache remembers no-coverage per point rounded to two decimals, for 24 hours, at most 256 points (least recently used dropped first); a successful response clears its point, and the warning is logged once per point. The weather service polls one position, so for it the change means a retry after 24 hours instead of never.
37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""Bounded, temporary suppression of NWS requests for points without coverage."""
|
|
|
|
import time
|
|
from collections import OrderedDict
|
|
|
|
|
|
class NWSNoCoverageCache:
|
|
MAX_POINTS = 256
|
|
TTL_SECONDS = 24 * 60 * 60
|
|
|
|
def __init__(self) -> None:
|
|
self._points: OrderedDict[tuple[float, float], float] = OrderedDict()
|
|
|
|
def is_unavailable(self, lat: float, lon: float) -> bool:
|
|
point = (round(lat, 2), round(lon, 2))
|
|
expires = self._points.get(point)
|
|
if expires is None:
|
|
return False
|
|
if time.monotonic() >= expires:
|
|
del self._points[point]
|
|
return False
|
|
self._points.move_to_end(point)
|
|
return True
|
|
|
|
def mark_unavailable(self, lat: float, lon: float) -> bool:
|
|
"""Remember the point; return whether it needs a new warning."""
|
|
if self.is_unavailable(lat, lon):
|
|
return False
|
|
point = (round(lat, 2), round(lon, 2))
|
|
self._points[point] = time.monotonic() + self.TTL_SECONDS
|
|
if len(self._points) > self.MAX_POINTS:
|
|
self._points.popitem(last=False)
|
|
return True
|
|
|
|
def mark_available(self, lat: float, lon: float) -> None:
|
|
self._points.pop((round(lat, 2), round(lon, 2)), None)
|