Grigo/LoraMapTester
1
0
Fork 0
generated from Grigo/AndroidTemplate
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

added opentopo

Browse Source
This commit is contained in:
grigo
2026-06-17 13:03:11 +03:00
parent f4ef87705c
commit 4891933879
8 changed files with 369 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/core/elevation.py
+235 -41
View File
@@ -1,4 +1,4 @@
"""Terrain elevation via self-hosted Open-Meteo-compatible API."""
"""Terrain elevation: OpenTopoData primary, Open-Meteo-compatible fallback."""
from __future__ import annotations
@@ -10,8 +10,9 @@ from typing import Any, Optional
import httpx
from .config import (
ELEVATION_API_URL,
ELEVATION_CONNECT_TIMEOUT,
ELEVATION_FALLBACK_URL,
ELEVATION_OPENTOPO_URL,
ELEVATION_PROBE_TTL_SEC,
)
@@ -21,8 +22,11 @@ _BATCH_SIZE = 100
_MAX_PROFILE_POINTS = 500
_CACHE: dict[tuple[float, float], Optional[float]] = {}
_probe_checked_at = 0.0
_probe_ok = False
_probe_error: Optional[str] = None
_probe_opentopo_ok = False
_probe_opentopo_error: Optional[str] = None
_probe_fallback_ok = False
_probe_fallback_error: Optional[str] = None
_last_fetch_source: Optional[str] = None
def _cache_key(lat: float, lon: float) -> tuple[float, float]:
@@ -42,9 +46,8 @@ def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
return r * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
def probe_elevation_api(force: bool = False) -> dict[str, Any]:
"""Ping elevation service before batch requests (cached for TTL)."""
global _probe_checked_at, _probe_ok, _probe_error
def _probe_opentopodata(force: bool = False) -> dict[str, Any]:
global _probe_checked_at, _probe_opentopo_ok, _probe_opentopo_error
now = time.monotonic()
if (
@@ -53,35 +56,129 @@ def probe_elevation_api(force: bool = False) -> dict[str, Any]:
and now - _probe_checked_at < ELEVATION_PROBE_TTL_SEC
):
return {
"ok": _probe_ok,
"url": ELEVATION_API_URL,
"error": _probe_error,
"ok": _probe_opentopo_ok,
"url": ELEVATION_OPENTOPO_URL,
"error": _probe_opentopo_error,
}
try:
with httpx.Client(timeout=ELEVATION_CONNECT_TIMEOUT) as client:
r = client.get(
ELEVATION_API_URL,
ELEVATION_OPENTOPO_URL,
params={"locations": "0.000000,0.000000"},
)
r.raise_for_status()
data = r.json()
if data.get("status") != "OK":
raise ValueError(f"status={data.get('status')}")
results = data.get("results") or []
if not results or results[0].get("elevation") is None:
raise ValueError("response has no elevation values")
_probe_opentopo_ok = True
_probe_opentopo_error = None
logger.info("OpenTopoData ok: %s", ELEVATION_OPENTOPO_URL)
except Exception as e:
_probe_opentopo_ok = False
_probe_opentopo_error = str(e)
logger.warning(
"OpenTopoData unreachable %s: %s", ELEVATION_OPENTOPO_URL, e
)
return {
"ok": _probe_opentopo_ok,
"url": ELEVATION_OPENTOPO_URL,
"error": _probe_opentopo_error,
}
def _probe_fallback(force: bool = False) -> dict[str, Any]:
global _probe_checked_at, _probe_fallback_ok, _probe_fallback_error
now = time.monotonic()
if (
not force
and _probe_checked_at > 0
and now - _probe_checked_at < ELEVATION_PROBE_TTL_SEC
):
return {
"ok": _probe_fallback_ok,
"url": ELEVATION_FALLBACK_URL,
"error": _probe_fallback_error,
}
try:
with httpx.Client(timeout=ELEVATION_CONNECT_TIMEOUT) as client:
r = client.get(
ELEVATION_FALLBACK_URL,
params={"latitude": "0.000000", "longitude": "0.000000"},
)
r.raise_for_status()
data = r.json()
if "elevation" not in data:
raise ValueError("response has no elevation field")
_probe_checked_at = now
_probe_ok = True
_probe_error = None
logger.info("elevation API ok: %s", ELEVATION_API_URL)
_probe_fallback_ok = True
_probe_fallback_error = None
logger.info("elevation fallback ok: %s", ELEVATION_FALLBACK_URL)
except Exception as e:
_probe_checked_at = now
_probe_ok = False
_probe_error = str(e)
logger.warning("elevation API unreachable %s: %s", ELEVATION_API_URL, e)
_probe_fallback_ok = False
_probe_fallback_error = str(e)
logger.warning(
"elevation fallback unreachable %s: %s", ELEVATION_FALLBACK_URL, e
)
return {
"ok": _probe_ok,
"url": ELEVATION_API_URL,
"error": _probe_error,
"ok": _probe_fallback_ok,
"url": ELEVATION_FALLBACK_URL,
"error": _probe_fallback_error,
}
def probe_elevation_api(force: bool = False) -> dict[str, Any]:
"""Ping elevation providers before batch requests (cached for TTL)."""
global _probe_checked_at
now = time.monotonic()
if (
not force
and _probe_checked_at > 0
and now - _probe_checked_at < ELEVATION_PROBE_TTL_SEC
):
op = {
"ok": _probe_opentopo_ok,
"url": ELEVATION_OPENTOPO_URL,
"error": _probe_opentopo_error,
}
fb = {
"ok": _probe_fallback_ok,
"url": ELEVATION_FALLBACK_URL,
"error": _probe_fallback_error,
}
else:
op = _probe_opentopodata(force=True)
fb = _probe_fallback(force=True)
_probe_checked_at = now
ok = op["ok"] or fb["ok"]
if op["ok"]:
url = op["url"]
error = None
elif fb["ok"]:
url = fb["url"]
error = None
else:
url = ELEVATION_OPENTOPO_URL
error = f"opentopodata: {op['error']}; fallback: {fb['error']}"
return {
"ok": ok,
"url": url,
"error": error,
"opentopodata_ok": op["ok"],
"opentopodata_url": op["url"],
"opentopodata_error": op["error"],
"fallback_ok": fb["ok"],
"fallback_url": fb["url"],
"fallback_error": fb["error"],
}
@@ -91,10 +188,40 @@ def elevation_status(force: bool = False) -> dict[str, Any]:
"elevation_ok": probe["ok"],
"elevation_url": probe["url"],
"elevation_error": probe["error"],
"elevation_opentopodata_ok": probe.get("opentopodata_ok"),
"elevation_opentopodata_url": probe.get("opentopodata_url"),
"elevation_fallback_ok": probe.get("fallback_ok"),
"elevation_fallback_url": probe.get("fallback_url"),
}
def _fetch_elevation_batch(
def _fetch_opentopodata_batch(
batch_lat: list[float], batch_lon: list[float]
) -> list[Optional[float]]:
if not batch_lat:
return []
locations = "|".join(
f"{lat:.6f},{lon:.6f}" for lat, lon in zip(batch_lat, batch_lon)
)
with httpx.Client(timeout=ELEVATION_CONNECT_TIMEOUT) as client:
r = client.get(ELEVATION_OPENTOPO_URL, params={"locations": locations})
r.raise_for_status()
data = r.json()
if data.get("status") != "OK":
raise ValueError(f"OpenTopoData status={data.get('status')}")
results = data.get("results") or []
out: list[Optional[float]] = []
for j, item in enumerate(results):
if j >= len(batch_lat):
break
elev = item.get("elevation")
out.append(None if elev is None else float(elev))
while len(out) < len(batch_lat):
out.append(None)
return out
def _fetch_fallback_batch(
batch_lat: list[float], batch_lon: list[float]
) -> list[Optional[float]]:
if not batch_lat:
@@ -104,7 +231,7 @@ def _fetch_elevation_batch(
"longitude": ",".join(f"{lon:.6f}" for lon in batch_lon),
}
with httpx.Client(timeout=ELEVATION_CONNECT_TIMEOUT) as client:
r = client.get(ELEVATION_API_URL, params=params)
r = client.get(ELEVATION_FALLBACK_URL, params=params)
r.raise_for_status()
data = r.json()
elevations = data.get("elevation") or []
@@ -112,15 +239,81 @@ def _fetch_elevation_batch(
for j, elev in enumerate(elevations):
if j >= len(batch_lat):
break
if elev is None:
out.append(None)
else:
out.append(float(elev))
out.append(None if elev is None else float(elev))
while len(out) < len(batch_lat):
out.append(None)
return out
def _fetch_batch_with_fallback(
batch_lat: list[float], batch_lon: list[float]
) -> list[Optional[float]]:
global _last_fetch_source
probe = probe_elevation_api()
op_ok = probe.get("opentopodata_ok", False)
fb_ok = probe.get("fallback_ok", False)
if not op_ok and not fb_ok:
return [None] * len(batch_lat)
out: list[Optional[float]] = [None] * len(batch_lat)
used_opentopo = False
used_fallback = False
if op_ok:
try:
out = _fetch_opentopodata_batch(batch_lat, batch_lon)
used_opentopo = any(v is not None for v in out)
except Exception as e:
logger.warning(
"OpenTopoData batch failed (%s points): %s", len(batch_lat), e
)
out = [None] * len(batch_lat)
missing_idx = [i for i, v in enumerate(out) if v is None]
if missing_idx and fb_ok:
miss_lat = [batch_lat[i] for i in missing_idx]
miss_lon = [batch_lon[i] for i in missing_idx]
try:
fb_vals = _fetch_fallback_batch(miss_lat, miss_lon)
for j, idx in enumerate(missing_idx):
out[idx] = fb_vals[j] if j < len(fb_vals) else None
if any(v is not None for v in fb_vals):
used_fallback = True
except Exception as e:
logger.warning(
"elevation fallback batch failed (%s points): %s",
len(miss_lat),
e,
)
if used_opentopo and used_fallback:
_last_fetch_source = "opentopodata+openmeteo"
elif used_opentopo:
_last_fetch_source = "opentopodata"
elif used_fallback:
_last_fetch_source = "openmeteo"
else:
_last_fetch_source = None
return out
def _active_elevation_url() -> str:
probe = probe_elevation_api()
if probe.get("opentopodata_ok"):
return ELEVATION_OPENTOPO_URL
if probe.get("fallback_ok"):
return ELEVATION_FALLBACK_URL
return ELEVATION_OPENTOPO_URL
def _active_api_source() -> str:
return _last_fetch_source or (
"opentopodata" if probe_elevation_api().get("opentopodata_ok") else "openmeteo"
)
def fetch_elevation_m(lat: float, lon: float) -> Optional[float]:
vals = fetch_elevations_batch([lat], [lon])
return vals[0] if vals else None
@@ -135,7 +328,7 @@ def fetch_elevations_batch(
probe = probe_elevation_api()
if not probe["ok"]:
logger.warning(
"skip elevation fetch: API unreachable (%s)",
"skip elevation fetch: all providers unreachable (%s)",
probe.get("error"),
)
return [None] * len(lats)
@@ -159,14 +352,15 @@ def fetch_elevations_batch(
batch_lat = pending_lat[start : start + _BATCH_SIZE]
batch_lon = pending_lon[start : start + _BATCH_SIZE]
try:
batch_vals = _fetch_elevation_batch(batch_lat, batch_lon)
batch_vals = _fetch_batch_with_fallback(batch_lat, batch_lon)
for j, val in enumerate(batch_vals):
lat = batch_lat[j]
lon = batch_lon[j]
_CACHE[_cache_key(lat, lon)] = val
out[batch_i[j]] = val
logger.info(
"elevation ok: %s points, sample=%s",
"elevation ok (%s): %s points, sample=%s",
_last_fetch_source,
len(batch_lat),
batch_vals[0] if batch_vals else None,
)
@@ -178,7 +372,7 @@ def fetch_elevations_batch(
)
for j in range(len(batch_lat)):
try:
single = _fetch_elevation_batch(
single = _fetch_batch_with_fallback(
[batch_lat[j]], [batch_lon[j]]
)
val = single[0] if single else None
@@ -329,7 +523,7 @@ def build_elevation_profile(
"total_m": 0.0,
"api_source": "elevation",
"api_error": f"elevation API unreachable: {probe['error']}",
"elevation_url": ELEVATION_API_URL,
"elevation_url": _active_elevation_url(),
}
lats = [s["lat"] for s in samples]
@@ -356,8 +550,8 @@ def build_elevation_profile(
"min_elevation_m": min(elev_vals) if elev_vals else None,
"max_elevation_m": max(elev_vals) if elev_vals else None,
"points": profile,
"api_source": "elevation",
"elevation_url": ELEVATION_API_URL,
"api_source": _active_api_source(),
"elevation_url": _active_elevation_url(),
}
if not elev_vals:
result["api_error"] = "elevation API returned no values"
@@ -429,7 +623,7 @@ def build_elevation_grid(
return {
"ok": False,
"error": f"elevation API unreachable: {probe['error']}",
"elevation_url": ELEVATION_API_URL,
"elevation_url": _active_elevation_url(),
}
radius_m = max(50.0, min(float(radius_m), 500.0))
@@ -481,8 +675,8 @@ def build_elevation_grid(
"points": points,
"min_delta_m": round(min(deltas), 1),
"max_delta_m": round(max(deltas), 1),
"api_source": "elevation",
"elevation_url": ELEVATION_API_URL,
"api_source": _active_api_source(),
"elevation_url": _active_elevation_url(),
}
@@ -499,7 +693,7 @@ def find_nearest_hill(
return {
"ok": False,
"error": f"elevation API unreachable: {probe['error']}",
"elevation_url": ELEVATION_API_URL,
"elevation_url": _active_elevation_url(),
}
radius_m = max(500.0, min(float(radius_m), 15_000.0))
@@ -590,6 +784,6 @@ def find_nearest_hill(
"candidates": len(candidates),
"radius_m": radius_m,
"step_m": step_m,
"api_source": "elevation",
"elevation_url": ELEVATION_API_URL,
"api_source": _active_api_source(),
"elevation_url": _active_elevation_url(),
}