Files
meshview/meshview/web_api/api.py
T
2026-10-15 22:49:55 -07:00

1333 lines
45 KiB
Python

"""API endpoints for MeshView."""
import asyncio
import datetime
import json
import logging
import math
import os
import re
from aiohttp import web
from sqlalchemy import func, select
from meshtastic.protobuf.portnums_pb2 import PortNum
from meshview import database, decode_payload, store, traceroute
from meshview.__version__ import __version__, _git_revision_short, get_version_info
from meshview.config import CONFIG
from meshview.models import DailySnapshot, Node
from meshview.models import Packet as PacketModel
from meshview.models import PacketSeen as PacketSeenModel
from meshview.radio.coverage import (
DEFAULT_MAX_DBM,
DEFAULT_MIN_DBM,
DEFAULT_RELIABILITY,
DEFAULT_THRESHOLD_DBM,
ITM_AVAILABLE,
compute_coverage,
compute_perimeter,
)
logger = logging.getLogger(__name__)
# Will be set by web.py during initialization
Packet = None
SEQ_REGEX = None
LANG_DIR = None
_LANG_CACHE = {}
# Create dedicated route table for API endpoints
routes = web.RouteTableDef()
# Language codes are short identifiers such as "en", "es", "pt-BR".
LANG_CODE_RE = re.compile(r"^[A-Za-z]{2,3}(?:[-_][A-Za-z0-9]{2,8})?$")
def _config_bool(section: str, key: str, default: bool = False) -> bool:
return str(CONFIG.get(section, {}).get(key, default)).lower() in ("1", "true", "yes", "on")
def _config_int(section: str, key: str, default: int = 0) -> int:
try:
return int(CONFIG.get(section, {}).get(key, default))
except (TypeError, ValueError):
return default
def _config_str(section: str, key: str, default: str = "") -> str:
value = CONFIG.get(section, {}).get(key, default)
return str(value) if value is not None else default
def _cleanup_status_file() -> str:
cleanup_logfile = CONFIG.get("logging", {}).get("db_cleanup_logfile", "dbcleanup.log")
path_without_extension, _ = os.path.splitext(cleanup_logfile)
return f"{path_without_extension}.status.json"
def _backup_status_file() -> str:
cleanup_logfile = CONFIG.get("logging", {}).get("db_cleanup_logfile", "dbcleanup.log")
return os.path.join(os.path.dirname(cleanup_logfile), "dbbackup.status.json")
def _get_cleanup_health() -> dict:
cleanup_health = {
"enabled": _config_bool("cleanup", "enabled", False),
"days_to_keep": _config_int("cleanup", "days_to_keep", 14),
"scheduled_time": (
f"{_config_int('cleanup', 'hour', 2):02d}:{_config_int('cleanup', 'minute', 0):02d}"
),
"vacuum": _config_bool("cleanup", "vacuum", False),
"status": "disabled",
}
if cleanup_health["enabled"]:
cleanup_health["status"] = "unknown"
status_file = _cleanup_status_file()
cleanup_health["status_file"] = status_file
if not os.path.exists(status_file):
return cleanup_health
try:
with open(status_file, encoding="utf-8") as f:
last_run = json.load(f)
except Exception as e:
cleanup_health["status"] = "error"
cleanup_health["error"] = f"Unable to read cleanup status: {e}"
return cleanup_health
cleanup_health["status"] = last_run.get("status", cleanup_health["status"])
cleanup_health["last_run"] = last_run
return cleanup_health
def _get_backup_health() -> dict:
backup_hour = _config_int("cleanup", "backup_hour", _config_int("cleanup", "hour", 2))
backup_minute = _config_int("cleanup", "backup_minute", _config_int("cleanup", "minute", 0))
backup_health = {
"enabled": _config_bool("cleanup", "backup_enabled", False),
"backup_dir": _config_str("cleanup", "backup_dir", "./backups"),
"scheduled_time": f"{backup_hour:02d}:{backup_minute:02d}",
"status": "disabled",
}
if backup_health["enabled"]:
backup_health["status"] = "unknown"
status_file = _backup_status_file()
backup_health["status_file"] = status_file
if not os.path.exists(status_file):
return backup_health
try:
with open(status_file, encoding="utf-8") as f:
last_run = json.load(f)
except Exception as e:
backup_health["status"] = "error"
backup_health["error"] = f"Unable to read backup status: {e}"
return backup_health
backup_health["status"] = last_run.get("status", backup_health["status"])
backup_health["last_run"] = last_run
return backup_health
def _haversine_km(lat1, lon1, lat2, lon2):
r = 6371.0
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlambda = math.radians(lon2 - lon1)
a = math.sin(dphi / 2.0) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2.0) ** 2
return 2 * r * math.asin(math.sqrt(a))
def _bearing_deg(lat1, lon1, lat2, lon2):
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
dlambda = math.radians(lon2 - lon1)
y = math.sin(dlambda) * math.cos(phi2)
x = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dlambda)
bearing = math.degrees(math.atan2(y, x))
return (bearing + 360.0) % 360.0
OBSERVED_MAX_DISTANCE_KM = 50.0
# Bounds for /api/coverage query parameters. The ITM loop runs once per (distance step x
# bearing), so the request size must be capped or a single call can hold the CPU for minutes.
COVERAGE_LIMITS = {
"freq_mhz": (1.0, 10_000.0),
"tx_dbm": (-50.0, 60.0),
"tx_height_m": (0.0, 1_000.0),
"rx_height_m": (0.0, 1_000.0),
"radius_km": (1.0, 200.0),
"step_km": (0.05, 50.0),
"reliability": (0.01, 0.99),
"threshold_dbm": (-200.0, 0.0),
}
COVERAGE_MAX_STEPS = 4_000 # radius_km / step_km
def init_api_module(packet_class, seq_regex, lang_dir):
"""Initialize API module with dependencies from main web module."""
global Packet, SEQ_REGEX, LANG_DIR
Packet = packet_class
SEQ_REGEX = seq_regex
LANG_DIR = lang_dir
@routes.get("/api/channels")
async def api_channels(request: web.Request):
period_type = request.query.get("period_type", "hour")
length = int(request.query.get("length", 24))
try:
channels = await store.get_channels_in_period(period_type, length)
return web.json_response({"channels": channels})
except Exception as e:
return web.json_response({"channels": [], "error": str(e)})
@routes.get("/api/nodes")
async def api_nodes(request):
try:
# Optional query parameters
node_id = request.query.get("node_id")
role = request.query.get("role")
channel = request.query.get("channel")
hw_model = request.query.get("hw_model")
days_active = request.query.get("days_active")
if days_active:
try:
days_active = int(days_active)
except ValueError:
days_active = None
# Fetch nodes from database
nodes = await store.get_nodes(
node_id=node_id, role=role, channel=channel, hw_model=hw_model, days_active=days_active
)
# Prepare the JSON response
nodes_data = []
for n in nodes:
nodes_data.append(
{
"id": getattr(n, "id", None),
"node_id": n.node_id,
"long_name": n.long_name,
"short_name": n.short_name,
"hw_model": n.hw_model,
"firmware": n.firmware,
"role": n.role,
"last_lat": getattr(n, "last_lat", None),
"last_long": getattr(n, "last_long", None),
"channel": n.channel,
"is_mqtt_gateway": getattr(n, "is_mqtt_gateway", None),
# "last_update": n.last_update.isoformat(),
"first_seen_us": n.first_seen_us,
"last_seen_us": n.last_seen_us,
}
)
return web.json_response({"nodes": nodes_data})
except Exception as e:
logger.error(f"Error in /api/nodes: {e}")
return web.json_response({"error": "Failed to fetch nodes"}, status=500)
@routes.get("/api/packets")
async def api_packets(request):
try:
# --- Parse query parameters ---
packet_id_str = request.query.get("packet_id")
limit_str = request.query.get("limit", "50")
since_str = request.query.get("since")
portnum_str = request.query.get("portnum")
contains = request.query.get("contains")
# NEW — explicit filters
from_node_id_str = request.query.get("from_node_id")
to_node_id_str = request.query.get("to_node_id")
node_id_str = request.query.get("node_id") # legacy: match either from/to
# --- If a packet_id is provided, return only that packet ---
if packet_id_str:
try:
packet_id = int(packet_id_str)
except ValueError:
return web.json_response({"error": "Invalid packet_id format"}, status=400)
packet = await store.get_packet(packet_id)
if not packet:
return web.json_response({"packets": []})
p = Packet.from_model(packet)
data = {
"id": p.id,
"storage_id": p.id,
"packet_id": p.packet_id,
"from_node_id": p.from_node_id,
"to_node_id": p.to_node_id,
"portnum": int(p.portnum) if p.portnum is not None else None,
"payload": (p.payload or "").strip(),
"import_time_us": p.import_time_us,
"channel": p.channel,
"long_name": getattr(p.from_node, "long_name", ""),
}
return web.json_response({"packets": [data]})
# --- Parse limit ---
try:
limit = min(max(int(limit_str), 1), 1000)
except ValueError:
limit = 50
# --- Parse since timestamp ---
since = None
if since_str:
try:
since = int(since_str)
except ValueError:
logger.warning(f"Invalid 'since' value (expected microseconds): {since_str}")
# --- Parse portnum ---
portnum = None
if portnum_str:
try:
portnum = int(portnum_str)
except ValueError:
logger.warning(f"Invalid portnum: {portnum_str}")
# --- Parse node filters ---
from_node_id = None
to_node_id = None
node_id = None # legacy: match either from/to
if from_node_id_str:
try:
from_node_id = int(from_node_id_str, 0)
except ValueError:
logger.warning(f"Invalid from_node_id: {from_node_id_str}")
if to_node_id_str:
try:
to_node_id = int(to_node_id_str, 0)
except ValueError:
logger.warning(f"Invalid to_node_id: {to_node_id_str}")
if node_id_str:
try:
node_id = int(node_id_str, 0)
except ValueError:
logger.warning(f"Invalid node_id: {node_id_str}")
# --- Fetch packets using explicit filters ---
contains_for_query = contains
if portnum == PortNum.TEXT_MESSAGE_APP and contains:
contains_for_query = None
packets = await store.get_packets(
from_node_id=from_node_id,
to_node_id=to_node_id,
node_id=node_id,
portnum=portnum,
after=since,
contains=contains_for_query,
limit=limit,
)
ui_packets = [Packet.from_model(p) for p in packets]
# --- Text message filtering ---
if portnum == PortNum.TEXT_MESSAGE_APP:
ui_packets = [p for p in ui_packets if p.payload and not SEQ_REGEX.fullmatch(p.payload)]
if contains:
ui_packets = [p for p in ui_packets if contains.lower() in p.payload.lower()]
# --- Sort descending by import_time_us ---
ui_packets.sort(
key=lambda p: (p.import_time_us is not None, p.import_time_us or 0), reverse=True
)
ui_packets = ui_packets[:limit]
# --- Build JSON output ---
packets_data = []
for p in ui_packets:
packet_dict = {
"id": p.id,
"storage_id": p.id,
"packet_id": p.packet_id,
"import_time_us": p.import_time_us,
"channel": p.channel,
"from_node_id": p.from_node_id,
"to_node_id": p.to_node_id,
"portnum": int(p.portnum),
"long_name": getattr(p.from_node, "long_name", ""),
"payload": (p.payload or "").strip(),
"to_long_name": getattr(p.to_node, "long_name", ""),
}
reply_id = getattr(
getattr(getattr(p, "raw_mesh_packet", None), "decoded", None),
"reply_id",
None,
)
if reply_id:
packet_dict["reply_id"] = reply_id
packets_data.append(packet_dict)
# --- Latest import_time_us for incremental fetch ---
latest_import_time = None
if packets_data:
for p in packets_data:
if p.get("import_time_us") and p["import_time_us"] > 0:
latest_import_time = max(latest_import_time or 0, p["import_time_us"])
response = {"packets": packets_data}
if latest_import_time is not None:
response["latest_import_time"] = latest_import_time
return web.json_response(response)
except Exception as e:
logger.error(f"Error in /api/packets: {e}")
return web.json_response({"error": "Failed to fetch packets"}, status=500)
@routes.get("/api/stats")
async def api_stats(request):
"""
Enhanced stats endpoint:
- Supports global stats (existing behavior)
- Supports per-node stats using ?node=<node_id>
returning both sent AND seen counts in the specified period
"""
allowed_periods = {"hour", "day"}
period_type = request.query.get("period_type", "hour").lower()
if period_type not in allowed_periods:
return web.json_response(
{"error": f"Invalid period_type. Must be one of {allowed_periods}"},
status=400,
)
try:
length = int(request.query.get("length", 24))
except ValueError:
return web.json_response({"error": "length must be an integer"}, status=400)
# NEW: optional combined node stats
node_str = request.query.get("node")
if node_str:
try:
node_id = int(node_str)
except ValueError:
return web.json_response({"error": "node must be an integer"}, status=400)
# Fetch sent packets
sent = await store.get_packet_stats(
period_type=period_type,
length=length,
from_node=node_id,
)
# Fetch seen packets
seen = await store.get_packet_stats(
period_type=period_type,
length=length,
to_node=node_id,
)
return web.json_response(
{
"node_id": node_id,
"period_type": period_type,
"length": length,
"sent": sent.get("total", 0),
"seen": seen.get("total", 0),
}
)
# ---- Existing full stats mode (unchanged) ----
channel = request.query.get("channel")
def parse_int_param(name):
value = request.query.get(name)
if value is not None:
try:
return int(value)
except ValueError:
raise web.HTTPBadRequest(
text=json.dumps({"error": f"{name} must be an integer"}),
content_type="application/json",
) from None
return None
portnum = parse_int_param("portnum")
to_node = parse_int_param("to_node")
from_node = parse_int_param("from_node")
stats = await store.get_packet_stats(
period_type=period_type,
length=length,
channel=channel,
portnum=portnum,
to_node=to_node,
from_node=from_node,
)
return web.json_response(stats)
@routes.get("/api/stats/count")
async def api_stats_count(request):
"""
Returns packet and packet_seen totals.
Behavior:
• If no filters → total packets ever + total seen ever
• If filters → apply window/channel/from/to + packet_id
"""
# -------- Parse request parameters --------
packet_id_str = request.query.get("packet_id")
packet_id = None
if packet_id_str:
try:
packet_id = int(packet_id_str)
except ValueError:
return web.json_response({"error": "packet_id must be integer"}, status=400)
period_type = request.query.get("period_type")
length_str = request.query.get("length")
length = None
if length_str:
try:
length = int(length_str)
except ValueError:
return web.json_response({"error": "length must be integer"}, status=400)
channel = request.query.get("channel")
def parse_int(name):
value = request.query.get(name)
if value is None:
return None
try:
return int(value)
except ValueError:
raise web.HTTPBadRequest(
text=json.dumps({"error": f"{name} must be integer"}),
content_type="application/json",
) from None
from_node = parse_int("from_node")
to_node = parse_int("to_node")
# -------- Case 1: NO FILTERS → return global totals --------
no_filters = (
period_type is None
and length is None
and channel is None
and from_node is None
and to_node is None
and packet_id is None
)
if no_filters:
total_packets = await store.get_total_packet_count()
total_seen = await store.get_total_packet_seen_count()
return web.json_response({"total_packets": total_packets, "total_seen": total_seen})
# -------- Case 2: Apply filters → compute totals --------
total_packets = await store.get_total_packet_count(
period_type=period_type,
length=length,
channel=channel,
from_node=from_node,
to_node=to_node,
)
total_seen = await store.get_total_packet_seen_count(
packet_id=packet_id,
period_type=period_type,
length=length,
channel=channel,
from_node=from_node,
to_node=to_node,
)
return web.json_response({"total_packets": total_packets, "total_seen": total_seen})
@routes.get("/api/snapshots/daily")
async def api_daily_snapshots(request):
length_raw = request.query.get("length")
length = None
if length_raw is not None:
try:
length = int(length_raw)
except ValueError:
return web.json_response({"error": "length must be an integer"}, status=400)
length = max(1, min(length, 3650))
try:
async with database.async_session() as session:
if length is not None:
rows = (
(
await session.execute(
select(DailySnapshot)
.order_by(DailySnapshot.snapshot_date.desc())
.limit(length)
)
)
.scalars()
.all()
)
rows = list(reversed(rows))
else:
rows = (
(
await session.execute(
select(DailySnapshot).order_by(DailySnapshot.snapshot_date.asc())
)
)
.scalars()
.all()
)
data = [
{
"date": row.snapshot_date.isoformat(),
"node_count": row.node_count,
"packet_count": row.packet_count,
"gateway_count": row.gateway_count,
"captured_at_us": row.captured_at_us,
}
for row in rows
]
return web.json_response({"data": data})
except Exception as e:
logger.error(f"Error in /api/snapshots/daily: {e}")
return web.json_response({"error": "Failed to fetch daily snapshots"}, status=500)
@routes.get("/api/edges")
async def api_edges(request):
since = datetime.datetime.now() - datetime.timedelta(hours=12)
filter_type = request.query.get("type")
# NEW → optional single-node filter
node_filter_str = request.query.get("node_id")
node_filter = None
if node_filter_str:
try:
node_filter = int(node_filter_str)
except ValueError:
return web.json_response({"error": "node_id must be integer"}, status=400)
edges = {}
traceroute_count = 0
edges_added_tr = 0
edges_added_neighbor = 0
# --- Traceroute edges ---
if filter_type in (None, "traceroute"):
async for tr in store.get_traceroutes(since):
traceroute_count += 1
try:
route = decode_payload.decode_payload(PortNum.TRACEROUTE_APP, tr.route)
except Exception:
continue
if route is None or tr.packet is None:
continue
path = traceroute.forward_path(
tr.packet.from_node_id,
tr.packet.to_node_id,
route.route,
tr.done,
tr.gateway_node_id,
)
for a, b in traceroute.edges(path):
if (a, b) not in edges:
edges[(a, b)] = "traceroute"
edges_added_tr += 1
# --- Neighbor edges ---
if filter_type in (None, "neighbor"):
packets = await store.get_packets(portnum=71)
for packet in packets:
try:
_, neighbor_info = decode_payload.decode(packet)
except Exception:
continue
for node in neighbor_info.neighbors:
edge = (node.node_id, packet.from_node_id)
if edge not in edges:
edges[edge] = "neighbor"
edges_added_neighbor += 1
# Convert to list
edges_list = [
{"from": frm, "to": to, "type": edge_type} for (frm, to), edge_type in edges.items()
]
# NEW → apply node_id filtering
if node_filter is not None:
edges_list = [e for e in edges_list if e["from"] == node_filter or e["to"] == node_filter]
return web.json_response({"edges": edges_list})
@routes.get("/api/config")
async def api_config(request):
try:
# ------------------ Helpers ------------------
def get(section, key, default=None):
"""Safe getter for both dict and ConfigParser."""
if isinstance(section, dict):
return section.get(key, default)
return section.get(key, fallback=default)
def get_bool(section, key, default=False):
val = get(section, key, default)
if isinstance(val, bool):
return "true" if val else "false"
if isinstance(val, str):
return "true" if val.lower() in ("1", "true", "yes", "on") else "false"
return "true" if bool(val) else "false"
def get_float(section, key, default=0.0):
try:
return float(get(section, key, default))
except Exception:
return float(default)
def get_int(section, key, default=0):
try:
return int(get(section, key, default))
except Exception:
return default
def get_str(section, key, default=""):
val = get(section, key, default)
return str(val) if val is not None else str(default)
# ------------------ SITE ------------------
site = CONFIG.get("site", {})
safe_site = {
"domain": get_str(site, "domain", ""),
"language": get_str(site, "language", "en"),
"title": get_str(site, "title", ""),
"message": get_str(site, "message", ""),
"starting": get_str(site, "starting", "/chat"),
"nodes": get_bool(site, "nodes", True),
"chat": get_bool(site, "chat", True),
"everything": get_bool(site, "everything", True),
"graphs": get_bool(site, "graphs", True),
"stats": get_bool(site, "stats", True),
"net": get_bool(site, "net", True),
"map": get_bool(site, "map", True),
"top": get_bool(site, "top", True),
"map_top_left_lat": get_float(site, "map_top_left_lat", 39.0),
"map_top_left_lon": get_float(site, "map_top_left_lon", -123.0),
"map_bottom_right_lat": get_float(site, "map_bottom_right_lat", 36.0),
"map_bottom_right_lon": get_float(site, "map_bottom_right_lon", -121.0),
"map_interval": get_int(site, "map_interval", 3),
"map_channel_colors": {
channel: color.strip()
for channel, color in CONFIG.get("map_channel_colors", {}).items()
if re.fullmatch(r"#[0-9a-fA-F]{6}", color.strip())
},
"firehose_interval": get_int(site, "firehose_interval", 3),
"weekly_net_message": get_str(
site, "weekly_net_message", "Weekly Mesh check-in message."
),
"net_tag": get_str(site, "net_tag", "#BayMeshNet"),
"version": str(__version__),
}
# ------------------ MQTT ------------------
mqtt = CONFIG.get("mqtt", {})
topics_raw = get(mqtt, "topics", [])
if isinstance(topics_raw, str):
try:
topics = json.loads(topics_raw)
except Exception:
topics = [topics_raw]
elif isinstance(topics_raw, list):
topics = topics_raw
else:
topics = []
safe_mqtt = {
"server": get_str(mqtt, "server", ""),
"topics": topics,
}
# ------------------ CLEANUP ------------------
cleanup = CONFIG.get("cleanup", {})
safe_cleanup = {
"enabled": get_bool(cleanup, "enabled", False),
"days_to_keep": get_str(cleanup, "days_to_keep", "14"),
"hour": get_str(cleanup, "hour", "2"),
"minute": get_str(cleanup, "minute", "0"),
"vacuum": get_bool(cleanup, "vacuum", False),
}
safe_config = {
"site": safe_site,
"mqtt": safe_mqtt,
"cleanup": safe_cleanup,
}
return web.json_response(safe_config)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@routes.get("/api/lang")
async def api_lang(request):
# Language from ?lang=xx, fallback to config, then to "en"
lang_code = request.query.get("lang") or CONFIG.get("site", {}).get("language", "en")
section = request.query.get("section")
# The code names a file, so it must stay a plain language code. Anything else (a path,
# an absolute filename) falls back to English just like an unknown language does.
if not LANG_CODE_RE.match(lang_code):
lang_code = "en"
lang_file = os.path.join(LANG_DIR, f"{lang_code}.json")
if not os.path.exists(lang_file):
lang_file = os.path.join(LANG_DIR, "en.json")
# Cache by file + mtime to avoid re-reading on every request
try:
mtime = os.path.getmtime(lang_file)
except OSError:
mtime = None
cache_key = lang_file
cached = _LANG_CACHE.get(cache_key)
if cached and cached.get("mtime") == mtime:
translations = cached["translations"]
else:
with open(lang_file, encoding="utf-8") as f:
translations = json.load(f)
_LANG_CACHE[cache_key] = {"mtime": mtime, "translations": translations}
if section:
section = section.lower()
if section in translations:
return web.json_response(translations[section])
else:
return web.json_response(
{"error": f"Section '{section}' not found in {lang_code}"}, status=404
)
# if no section requested → return full translation file
return web.json_response(translations)
@routes.get("/health")
async def health_check(request):
"""Health check endpoint for monitoring and load balancers."""
health_status = {
"status": "healthy",
"timestamp": datetime.datetime.now(datetime.UTC).isoformat(),
"version": __version__,
"git_revision": _git_revision_short,
"cleanup": _get_cleanup_health(),
"backup": _get_backup_health(),
}
# Check database connectivity
try:
async with database.async_session() as session:
result = await session.execute(select(func.max(PacketModel.import_time_us)))
last_import_time_us = result.scalar()
health_status["database"] = "connected"
if last_import_time_us is not None:
now_us = int(datetime.datetime.now(datetime.UTC).timestamp() * 1_000_000)
health_status["seconds_since_last_message"] = round(
(now_us - last_import_time_us) / 1_000_000, 1
)
except Exception as e:
logger.error(f"Database health check failed: {e}")
health_status["database"] = "disconnected"
health_status["status"] = "unhealthy"
return web.json_response(health_status, status=503)
# Get database file size
try:
db_url = CONFIG.get("database", {}).get("connection_string", "")
# Extract file path from SQLite connection string (e.g., "sqlite+aiosqlite:///packets.db")
if "sqlite" in db_url.lower():
db_path = db_url.split("///")[-1].split("?")[0]
if os.path.exists(db_path):
db_size_bytes = os.path.getsize(db_path)
# Convert to human-readable format
if db_size_bytes < 1024:
health_status["database_size"] = f"{db_size_bytes} B"
elif db_size_bytes < 1024 * 1024:
health_status["database_size"] = f"{db_size_bytes / 1024:.2f} KB"
elif db_size_bytes < 1024 * 1024 * 1024:
health_status["database_size"] = f"{db_size_bytes / (1024 * 1024):.2f} MB"
else:
health_status["database_size"] = (
f"{db_size_bytes / (1024 * 1024 * 1024):.2f} GB"
)
health_status["database_size_bytes"] = db_size_bytes
except Exception as e:
logger.warning(f"Failed to get database size: {e}")
# Don't fail health check if we can't get size
return web.json_response(health_status)
@routes.get("/version")
async def version_endpoint(request):
"""Return version information including semver and git revision."""
try:
version_info = get_version_info()
return web.json_response(version_info)
except Exception as e:
logger.error(f"Error in /version: {e}")
return web.json_response({"error": "Failed to fetch version info"}, status=500)
@routes.get("/api/packets_seen/{packet_id}")
async def api_packets_seen(request):
try:
# --- Validate packet_id ---
try:
packet_id = int(request.match_info["packet_id"])
except (KeyError, ValueError):
return web.json_response(
{"error": "Invalid or missing packet_id"},
status=400,
)
# --- Fetch list using your helper ---
rows = await store.get_packets_seen(packet_id)
items = []
for row in rows: # <-- FIX: normal for-loop
items.append(
{
"packet_id": row.packet_id,
"node_id": row.node_id,
"rx_time": row.rx_time,
"hop_limit": row.hop_limit,
"hop_start": row.hop_start,
"channel": row.channel,
"rx_snr": row.rx_snr,
"rx_rssi": row.rx_rssi,
"topic": row.topic,
"import_time_us": row.import_time_us,
}
)
return web.json_response({"seen": items})
except Exception:
logger.exception("Error in /api/packets_seen")
return web.json_response(
{"error": "Internal server error"},
status=500,
)
@routes.get("/api/traceroute/{packet_id}")
async def api_traceroute(request):
packet_id = int(request.match_info['packet_id'])
traceroutes = list(await store.get_traceroute(packet_id))
packet = await store.get_packet(packet_id)
if not packet:
return web.json_response({"error": "Packet not found"}, status=404)
tr_groups = []
# --------------------------------------------
# Decode each traceroute entry
# --------------------------------------------
for idx, tr in enumerate(traceroutes):
route = decode_payload.decode_payload(PortNum.TRACEROUTE_APP, tr.route)
if route is None:
continue
forward_list = list(route.route)
reverse_list = list(route.route_back)
tr_groups.append(
{
"index": idx,
"gateway_node_id": tr.gateway_node_id,
"done": tr.done,
"forward_hops": forward_list,
"reverse_hops": reverse_list,
"reverse_complete": traceroute.return_is_complete(route),
}
)
# --------------------------------------------
# Compute UNIQUE paths + counts + winning path
# --------------------------------------------
from collections import Counter
forward_paths = []
reverse_paths = []
forward_complete = set()
winning_forward_paths = []
winning_reverse_paths = []
for tr in tr_groups:
f = tuple(tr["forward_hops"])
r = tuple(tr["reverse_hops"])
if tr["forward_hops"]:
forward_paths.append(f)
if tr["reverse_hops"]:
reverse_paths.append(r)
# A response proves the request reached the target, whatever its return trip did.
if tr["done"]:
forward_complete.add(f)
winning_forward_paths.append(f)
winning_reverse_paths.append((r, tr["reverse_complete"]))
# Deduplicate
unique_forward_paths = sorted(set(forward_paths))
unique_reverse_paths = sorted(set(reverse_paths))
# Count occurrences
forward_counts = Counter(forward_paths)
# Convert for JSON output
unique_forward_paths_json = [
{"path": list(p), "count": forward_counts[p], "complete": p in forward_complete}
for p in unique_forward_paths
]
unique_reverse_paths_json = [list(p) for p in unique_reverse_paths]
from_node_id = packet.from_node_id
to_node_id = packet.to_node_id
# Winning paths are only built from responses, which carry the endpoints reversed.
winning_forward_with_endpoints = []
for hops in dict.fromkeys(winning_forward_paths):
path = traceroute.forward_path(from_node_id, to_node_id, hops, done=True)
if len(path) > 1:
winning_forward_with_endpoints.append(path)
winning_reverse_with_endpoints = []
for hops, complete in winning_reverse_paths:
path = traceroute.return_path(from_node_id, to_node_id, hops, complete)
if len(path) > 1 and path not in winning_reverse_with_endpoints:
winning_reverse_with_endpoints.append(path)
winning_paths_json = {
"forward": winning_forward_with_endpoints,
"reverse": winning_reverse_with_endpoints,
}
# --------------------------------------------
# Final API output
# --------------------------------------------
initiator, target = traceroute.endpoints(
from_node_id, to_node_id, any(tr["done"] for tr in tr_groups)
)
return web.json_response(
{
"packet": {
"id": packet.id,
"from": packet.from_node_id,
"to": packet.to_node_id,
"initiator": initiator,
"target": target,
"channel": packet.channel,
},
"traceroute_packets": tr_groups,
"unique_forward_paths": unique_forward_paths_json,
"unique_reverse_paths": unique_reverse_paths_json,
"winning_paths": winning_paths_json,
}
)
@routes.get("/api/stats/top")
async def api_stats_top(request):
"""
Returns nodes sorted by SEEN (high → low) with pagination.
"""
period_type = request.query.get("period_type", "day")
length = int(request.query.get("length", 1))
channel = request.query.get("channel")
limit = min(int(request.query.get("limit", 20)), 100)
offset = int(request.query.get("offset", 0))
multiplier = 3600 if period_type == "hour" else 86400
window_us = length * multiplier * 1_000_000
max_packet_import = select(func.max(PacketModel.import_time_us)).scalar_subquery()
max_seen_import = select(func.max(PacketSeenModel.import_time_us)).scalar_subquery()
sent_cte = (
select(PacketModel.from_node_id.label("node_id"), func.count().label("sent"))
.where(PacketModel.import_time_us >= max_packet_import - window_us)
.group_by(PacketModel.from_node_id)
.cte("sent")
)
seen_cte = (
select(PacketModel.from_node_id.label("node_id"), func.count().label("seen"))
.select_from(PacketSeenModel)
.join(PacketModel, PacketModel.id == PacketSeenModel.packet_id)
.where(PacketSeenModel.import_time_us >= max_seen_import - window_us)
.group_by(PacketModel.from_node_id)
.cte("seen")
)
query = (
select(
Node.node_id,
Node.long_name,
Node.short_name,
Node.channel,
func.coalesce(sent_cte.c.sent, 0).label("sent"),
func.coalesce(seen_cte.c.seen, 0).label("seen"),
)
.select_from(Node)
.outerjoin(sent_cte, sent_cte.c.node_id == Node.node_id)
.outerjoin(seen_cte, seen_cte.c.node_id == Node.node_id)
.order_by(func.coalesce(seen_cte.c.seen, 0).desc())
.limit(limit)
.offset(offset)
)
count_query = select(func.count()).select_from(Node)
if channel:
query = query.where(Node.channel == channel)
count_query = count_query.where(Node.channel == channel)
async with database.async_session() as session:
rows = (await session.execute(query)).all()
total = (await session.execute(count_query)).scalar() or 0
nodes = []
for r in rows:
avg = r.seen / max(r.sent, 1)
nodes.append(
{
"node_id": r.node_id,
"long_name": r.long_name,
"short_name": r.short_name,
"channel": r.channel,
"sent": r.sent,
"seen": r.seen,
"avg": round(avg, 2),
}
)
return web.json_response(
{
"total": total,
"limit": limit,
"offset": offset,
"nodes": nodes,
}
)
@routes.get("/api/node/{node_id}/qr")
async def api_node_qr(request):
"""
Generate a Meshtastic URL for importing the node as a contact.
Returns the URL that can be used to generate a QR code.
"""
try:
node_id_str = request.match_info["node_id"]
node_id = int(node_id_str, 0)
except (KeyError, ValueError):
return web.json_response({"error": "Invalid node_id"}, status=400)
node = await store.get_node(node_id)
if not node:
return web.json_response({"error": "Node not found"}, status=404)
try:
from meshtastic.protobuf.admin_pb2 import SharedContact
from meshtastic.protobuf.config_pb2 import Config
from meshtastic.protobuf.mesh_pb2 import User
user = User()
user.id = f"!{node_id:08x}"
if node.long_name:
user.long_name = node.long_name
if node.short_name:
user.short_name = node.short_name
if node.hw_model:
try:
from meshtastic.protobuf.mesh_pb2 import HardwareModel
hw_model_value = getattr(HardwareModel, node.hw_model.upper(), None)
if hw_model_value is not None:
user.hw_model = hw_model_value
except (AttributeError, TypeError):
pass
if node.role:
try:
role_value = getattr(Config.DeviceConfig.Role, node.role.upper(), None)
if role_value is not None:
user.role = role_value
except (AttributeError, TypeError):
pass
contact = SharedContact()
contact.node_num = node_id
contact.user.CopyFrom(user)
contact.manually_verified = False
contact_bytes = contact.SerializeToString()
import base64
contact_b64 = base64.b64encode(contact_bytes).decode("ascii")
contact_b64url = contact_b64.replace("+", "-").replace("/", "_").rstrip("=")
meshtastic_url = f"https://meshtastic.org/v/#{contact_b64url}"
return web.json_response(
{
"node_id": node_id,
"long_name": node.long_name,
"short_name": node.short_name,
"role": node.role,
"meshtastic_url": meshtastic_url,
}
)
except Exception as e:
import traceback
logger.error(f"Error generating QR URL for node {node_id}: {e}")
logger.error(traceback.format_exc())
return web.json_response({"error": f"Failed to generate URL: {str(e)}"}, status=500)
@routes.get("/api/coverage/{node_id}")
async def api_coverage(request):
try:
node_id = int(request.match_info["node_id"], 0)
except (KeyError, ValueError):
return web.json_response({"error": "Invalid node_id"}, status=400)
if not ITM_AVAILABLE:
return web.json_response(
{"error": "Coverage requires pyitm. Run: pip install -r requirements.txt"},
status=503,
)
def parse_float(name, default):
value = request.query.get(name)
if value is None:
return default
try:
parsed = float(value)
except ValueError as exc:
raise web.HTTPBadRequest(
text=json.dumps({"error": f"{name} must be a number"}),
content_type="application/json",
) from exc
low, high = COVERAGE_LIMITS[name]
if not math.isfinite(parsed) or not (low <= parsed <= high):
raise web.HTTPBadRequest(
text=json.dumps({"error": f"{name} must be between {low} and {high}"}),
content_type="application/json",
)
return parsed
try:
freq_mhz = parse_float("freq_mhz", 907.0)
tx_dbm = parse_float("tx_dbm", 20.0)
tx_height_m = parse_float("tx_height_m", 5.0)
rx_height_m = parse_float("rx_height_m", 1.5)
radius_km = parse_float("radius_km", 40.0)
step_km = parse_float("step_km", 0.25)
reliability = parse_float("reliability", DEFAULT_RELIABILITY)
threshold_dbm = parse_float("threshold_dbm", DEFAULT_THRESHOLD_DBM)
except web.HTTPBadRequest as exc:
raise exc
if radius_km / step_km > COVERAGE_MAX_STEPS:
return web.json_response(
{"error": f"radius_km / step_km must not exceed {COVERAGE_MAX_STEPS}"}, status=400
)
node = await store.get_node(node_id)
if not node or not node.last_lat or not node.last_long:
return web.json_response({"error": "Node not found or missing location"}, status=404)
lat = node.last_lat * 1e-7
lon = node.last_long * 1e-7
mode = request.query.get("mode", "perimeter")
# The ITM model is pure CPU work; keep it off the event loop so other requests
# are still served while it runs.
if mode == "perimeter":
perimeter = await asyncio.to_thread(
compute_perimeter,
lat=round(lat, 7),
lon=round(lon, 7),
freq_mhz=round(freq_mhz, 3),
tx_dbm=round(tx_dbm, 2),
tx_height_m=round(tx_height_m, 2),
rx_height_m=round(rx_height_m, 2),
radius_km=round(radius_km, 2),
step_km=round(step_km, 3),
reliability=round(reliability, 3),
threshold_dbm=round(threshold_dbm, 1),
)
return web.json_response(
{"mode": "perimeter", "threshold_dbm": threshold_dbm, "perimeter": perimeter}
)
points = await asyncio.to_thread(
compute_coverage,
lat=round(lat, 7),
lon=round(lon, 7),
freq_mhz=round(freq_mhz, 3),
tx_dbm=round(tx_dbm, 2),
tx_height_m=round(tx_height_m, 2),
rx_height_m=round(rx_height_m, 2),
radius_km=round(radius_km, 2),
step_km=round(step_km, 3),
reliability=round(reliability, 3),
)
min_dbm = DEFAULT_MIN_DBM
max_dbm = DEFAULT_MAX_DBM
if points:
vals = [p[2] for p in points]
min_dbm = min(min_dbm, min(vals))
max_dbm = max(max_dbm, max(vals))
return web.json_response(
{"mode": "heatmap", "min_dbm": min_dbm, "max_dbm": max_dbm, "points": points}
)