Merge branch '3.0.8'

This commit is contained in:
pablorevilla-meshtastic committed 2026-09-09 12:27:04 -07:00
commit 85753d8704
19 files changed
+761 -98

No files matched your search

+1
View File
@@ -22,6 +22,7 @@ meshview-web.pid
/table_details.py
config.ini
*.log
*.status.json
# Screenshots
screenshots/*
+1 -1
View File
@@ -52,7 +52,7 @@ Unacceptable behavior includes harassment, insults, hate speech, personal attack
---
## I Want to Contribute
## I Want to Contribute!
### Legal Notice
By contributing to Meshview, you agree that:
+1 -1
View File
@@ -9,7 +9,7 @@ The project serves as a real-time monitoring and diagnostic tool for the Meshtas
- Node list/map UX: added node list filtering for wider active-node ranges, included all database nodes where appropriate, and made map node colors persistent across reboots.
- QR/API fixes: corrected QR code generation when a node role is missing or unavailable.
- Database cleanup: removed stored node public key data and added the Alembic migration to drop `node.public_key`.
- Protobufs: refreshed vendored Meshtastic protobuf Python modules to upstream commit `9ab4a1d08cb833d897aee2367013318718391f2f`, including new hardware models such as `RAK6421`.
- Protobufs: refreshed Meshtastic protobuf Python modules to upstream commit
- Tooling: improved the protobuf updater so it supports raw commit SHAs, the newer upstream `meshtastic/*.proto` layout, generated `.pyi` stubs, and the vendored `nanopb`/`serial_hal` outputs.
### Version 3.0.6 — March 2026
+77 -9
View File
@@ -203,6 +203,12 @@ Returns network edges (connections between nodes) based on traceroutes and neigh
Traceroute edges are collected over the last 12 hours. Neighbor edges are based on
port 71 packets.
A completed traceroute is observed as a response packet, whose `from`/`to` are reversed
with respect to the traced path, so edges are built as `to -> route[] -> from`. `route[]`
may contain `0xFFFFFFFF` placeholders for hops the mesh could not identify; the edges on
either side of such a placeholder are omitted, because the nodes around an unknown hop
are not neighbours.
Query Parameters
- `type` (optional, string): `traceroute` or `neighbor`. If omitted, returns both.
- `node_id` (optional, int): Filter edges to only those touching a node.
@@ -324,13 +330,30 @@ Returns traceroute details and derived paths for a packet.
Path Parameters
- `packet_id` (required, int): Packet ID.
A completed traceroute is observed as a response packet, whose `from`/`to` are reversed
with respect to the traced path. `initiator` and `target` resolve this: the forward path
runs `initiator -> forward_hops -> target` and the return path runs
`target -> reverse_hops -> initiator`, in both directions of travel. `forward_hops` and
`reverse_hops` contain only the intermediate hops, never the endpoints.
A gateway usually reports a response that is still travelling home, so `reverse_hops` is
often incomplete. `reverse_complete` says whether that observation had reached the
initiator; only then does the corresponding `winning_paths.reverse` entry end at the
initiator.
`winning_paths` is built from response observations, since a response proves the request
reached the target. `complete` on a `unique_forward_paths` entry says whether some
response reported that hop list, and so whether it is known to end at the target.
Response Example
```json
{
"packet": {
"id": 123,
"from": 111,
"to": 222,
"from": 222,
"to": 111,
"initiator": 111,
"target": 222,
"channel": "main"
},
"traceroute_packets": [
@@ -338,19 +361,21 @@ Response Example
"index": 0,
"gateway_node_id": 333,
"done": true,
"forward_hops": [111, 444, 222],
"reverse_hops": [222, 444, 111]
"forward_hops": [444, 555],
"reverse_hops": [555, 444],
"reverse_complete": true
}
],
"unique_forward_paths": [
{ "path": [111, 444, 222], "count": 2 }
{ "path": [444, 555], "count": 2, "complete": true }
],
"unique_reverse_paths": [
[222, 444, 111]
[555, 444]
],
"winning_paths": [
[111, 444, 222]
]
"winning_paths": {
"forward": [[111, 444, 555, 222]],
"reverse": [[222, 555, 444, 111]]
}
}
```
@@ -368,6 +393,49 @@ Response Example
"timestamp": "2025-07-22T12:45:00+00:00",
"version": "3.0.3",
"git_revision": "abc1234",
"cleanup": {
"enabled": true,
"days_to_keep": 14,
"scheduled_time": "02:00",
"vacuum": false,
"status": "ok",
"status_file": "dbcleanup.status.json",
"last_run": {
"status": "ok",
"started_at": "2026-06-04T09:00:00+00:00",
"completed_at": "2026-06-04T09:00:03+00:00",
"cutoff_at": "2026-05-21T09:00:00+00:00",
"days_to_keep": 14,
"vacuum_requested": false,
"vacuum_completed": false,
"rows_deleted": {
"packet": 1200,
"packet_seen": 3400,
"traceroute": 42,
"node": 3
},
"error": null
}
},
"backup": {
"enabled": true,
"backup_dir": "./backups",
"scheduled_time": "02:00",
"status": "ok",
"status_file": "dbbackup.status.json",
"last_run": {
"status": "ok",
"started_at": "2026-06-04T09:00:00+00:00",
"completed_at": "2026-06-04T09:00:02+00:00",
"backup_dir": "./backups",
"database_path": "packets.db",
"backup_file": "backups/packets_backup_20260604_090000.db.gz",
"original_size_bytes": 12939444,
"compressed_size_bytes": 4211560,
"compression_percent": 67.5,
"error": null
}
},
"database": "connected",
"database_size": "12.34 MB",
"database_size_bytes": 12939444
+1 -1
View File
@@ -3,7 +3,7 @@
import subprocess
from pathlib import Path
__version__ = "3.0.7"
__version__ = "3.0.8"
__release_date__ = "2026-5-12"
+1
View File
@@ -159,6 +159,7 @@
"times_seen": "Seen (24h)",
"avg_gateways": "Avg Gateways",
"showing_nodes": "Showing",
"all_channels": "All Channels",
"nodes_suffix": "nodes"
},
+1
View File
@@ -155,6 +155,7 @@
"times_seen": "Visto (24h)",
"avg_gateways": "Promedio de Gateways",
"showing_nodes": "Mostrando",
"all_channels": "Todos los Canales",
"nodes_suffix": "nodos"
},
+1
View File
@@ -145,6 +145,7 @@
"times_seen": "Принято (24ч)",
"avg_gateways": "Среднее количество шлюзов",
"showing_nodes": "Показать",
"all_channels": "Все каналы",
"nodes_suffix": "ноды"
},
"nodegraph": {
+32 -4
View File
@@ -12,7 +12,10 @@ from meshtastic.protobuf.mesh_pb2 import Data
from meshtastic.protobuf.mqtt_pb2 import ServiceEnvelope
from meshview.config import CONFIG
PRIMARY_KEY = base64.b64decode("1PG7OiApB1nwvP+rz05pAQ==")
# Meshtastic's well-known default PSK, i.e. the expansion of the single byte PSK index 1.
DEFAULT_PSK = base64.b64decode("1PG7OiApB1nwvP+rz05pAQ==")
PRIMARY_KEY = DEFAULT_PSK
VALID_KEY_LENGTHS = (16, 32)
logging.basicConfig(
level=logging.INFO,
@@ -52,6 +55,10 @@ def _strip_quotes(value):
return value
def _expand_short_psk(index):
return DEFAULT_PSK[:-1] + bytes([(DEFAULT_PSK[-1] + index - 1) % 256])
def _parse_secondary_keys():
mqtt_config = CONFIG.get("mqtt", {})
raw_value = mqtt_config.get("secondary_keys", "")
@@ -68,12 +75,33 @@ def _parse_secondary_keys():
keys = []
for value in values:
cleaned = _strip_quotes(str(value).strip())
if not cleaned:
continue
try:
cleaned = _strip_quotes(str(value).strip())
if cleaned:
keys.append(base64.b64decode(cleaned))
key = base64.b64decode(cleaned)
except (TypeError, ValueError):
logger.warning("Invalid base64 key in mqtt.secondary_keys: %s", value)
continue
if len(key) == 1:
if key[0] == 0:
logger.warning(
"Skipping PSK %s in mqtt.secondary_keys: index 0 means no encryption", value
)
continue
key = _expand_short_psk(key[0])
if len(key) not in VALID_KEY_LENGTHS:
logger.error(
"Skipping key %s in mqtt.secondary_keys: decoded to %d bytes, expected 16 or 32",
value,
len(key),
)
continue
keys.append(key)
return keys
+40 -9
View File
@@ -17,6 +17,34 @@ from meshview.models import DailySnapshot, Node, Packet, PacketSeen, Traceroute
logger = logging.getLogger(__name__)
MQTT_GATEWAY_CACHE: set[int] = set()
UNKNOWN_NODE_NAME = "unknown name"
NODE_NAME_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
MQTT_DIRECT_NODE_ID = 1
def normalize_node_name(name: str) -> str:
name = name.strip()
if not name or NODE_NAME_CONTROL_CHARS.search(name):
return UNKNOWN_NODE_NAME
return name
def storage_packet_id(packet) -> int | None:
if packet.id:
return packet.id
if packet.decoded.portnum != PortNum.MAP_REPORT_APP:
return None
from_node_id = getattr(packet, "from", 0) or 0
now_us = int(time.time() * 1_000_000)
return -(now_us * 2048 + (from_node_id & 0x7FF))
def storage_to_node_id(packet) -> int:
if packet.decoded.portnum == PortNum.MAP_REPORT_APP:
return MQTT_DIRECT_NODE_ID
return packet.to
async def capture_daily_snapshot() -> None:
@@ -138,20 +166,21 @@ async def process_envelope(topic, env):
await session.commit()
if not env.packet.id:
packet_id = storage_packet_id(env.packet)
if packet_id is None:
return
async with mqtt_database.async_session() as session:
# --- Packet insert with ON CONFLICT DO NOTHING
result = await session.execute(select(Packet).where(Packet.id == env.packet.id))
result = await session.execute(select(Packet).where(Packet.id == packet_id))
packet = result.scalar_one_or_none()
if not packet:
now_us = int(time.time() * 1_000_000)
packet_values = {
"id": env.packet.id,
"id": packet_id,
"portnum": env.packet.decoded.portnum,
"from_node_id": getattr(env.packet, "from"),
"to_node_id": env.packet.to,
"to_node_id": storage_to_node_id(env.packet),
"payload": env.packet.SerializeToString(),
"import_time_us": now_us,
"channel": env.channel_id,
@@ -199,7 +228,7 @@ async def process_envelope(topic, env):
now_us = int(time.time() * 1_000_000)
seen_values = {
"packet_id": env.packet.id,
"packet_id": packet_id,
"node_id": int(env.gateway_id[1:], 16),
"channel": env.channel_id,
"rx_time": env.packet.rx_time,
@@ -268,11 +297,13 @@ async def process_envelope(topic, env):
).scalar_one_or_none()
now_us = int(time.time() * 1_000_000)
long_name = normalize_node_name(user.long_name)
short_name = normalize_node_name(user.short_name)
if node:
node.node_id = node_id
node.long_name = user.long_name
node.short_name = user.short_name
node.long_name = long_name
node.short_name = short_name
node.hw_model = hw_model
node.role = role
node.channel = env.channel_id
@@ -283,8 +314,8 @@ async def process_envelope(topic, env):
node = Node(
id=user.id,
node_id=node_id,
long_name=user.long_name,
short_name=user.short_name,
long_name=long_name,
short_name=short_name,
hw_model=hw_model,
role=role,
channel=env.channel_id,
+12 -7
View File
@@ -241,7 +241,7 @@ function logPacketTimes(packet) {
const times = formatTimes(packet.import_time_us);
console.log(
"[firehose] packet time",
"id=" + packet.id,
"id=" + (packet.packet_id ?? packet.id),
"epoch_us=" + times.epoch,
"local=" + times.local,
"utc=" + times.utc
@@ -365,13 +365,21 @@ async function fetchUpdates() {
if (match) traceId = match[1];
inlineLinks += ` <a class="inline-link"
href="/graph/traceroute/${traceId}"
target="_blank">⮕</a>`;
href="/traceroute/${traceId}"
>⮕</a>`;
}
const safePayload = (pkt.payload || "")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;");
const packetIdCell = pkt.packet_id
? `<a href="/packet/${pkt.storage_id || pkt.id}"
style="text-decoration:underline; color:inherit;">
${pkt.packet_id}
</a>`
: pkt.portnum === 73
? `<span class="text-secondary" title="MQTT map report">Not at Packet</span>`
: `<span class="text-secondary" title="No Meshtastic packet ID">—</span>`;
const html = `
<tr class="packet-row">
@@ -382,10 +390,7 @@ async function fetchUpdates() {
<td>
<span class="toggle-btn">▶</span>
<a href="/packet/${pkt.id}"
style="text-decoration:underline; color:inherit;">
${pkt.id}
</a>
${packetIdCell}
</td>
<td>${from}</td>
+134 -14
View File
@@ -148,6 +148,7 @@
<script src="https://unpkg.com/leaflet-polylinedecorator@1.6.0/dist/leaflet.polylinedecorator.js"
integrity="sha384-FhPn/2P/fJGhQLeNWDn9B/2Gml2bPOrKJwFqJXgR3xOPYxWg5mYQ5XZdhUSugZT0"
crossorigin></script>
<script src="https://unpkg.com/leaflet.heat/dist/leaflet-heat.js"></script>
<script src="/static/portmaps.js"></script>
<script>
@@ -188,20 +189,50 @@ function applyTranslationsMap(root = document) {
====================================================== */
var map = L.map('map');
L.tileLayer('https://tile.openstreetmap.org/{z}/{x}/{y}.png',
{ maxZoom:19, attribution:'&copy; OpenStreetMap' }).addTo(map);
var baseLayers = {
"OpenStreetMap": L.tileLayer('https://tile.openstreetmap.org/{z}/{x}/{y}.png', {
maxZoom: 19,
attribution: '&copy; OpenStreetMap'
}),
"OpenTopoMap": L.tileLayer('https://{s}.tile.opentopomap.org/{z}/{x}/{y}.png', {
maxZoom: 17,
attribution: 'Map data: &copy; OpenStreetMap, SRTM | Map style: &copy; OpenTopoMap'
})
};
baseLayers["OpenStreetMap"].addTo(map);
L.control.scale({
position: 'bottomleft',
metric: true,
imperial: true
}).addTo(map);
var nodeLayer = L.layerGroup().addTo(map);
var edgeLayer = L.layerGroup().addTo(map);
var trafficHeatLayer = L.heatLayer([], {
radius: 34,
blur: 18,
maxZoom: 12,
minOpacity: 0.3,
gradient: {
0.2: "#0b1f4d",
0.45: "#12436b",
0.7: "#7f2704",
1.0: "#4d0000"
}
});
var overlayLayers = {
"Nodes": nodeLayer,
"Traffic HeatMap": trafficHeatLayer
};
L.control.layers(baseLayers, overlayLayers, { collapsed: false }).addTo(map);
// Data structures
var nodes = [], markers = {}, markerById = {}, nodeMap = new Map();
var edgeLayer = L.layerGroup().addTo(map), selectedNodeId = null;
var selectedNodeId = null;
var activeBlinks = new Map(), lastImportTime = null;
var mapInterval = 0;
var unmappedPackets = [];
var trafficHeatLoaded = false;
var trafficHeatLoading = false;
const UNMAPPED_LIMIT = 50;
const UNMAPPED_TTL_MS = 5000;
@@ -392,6 +423,15 @@ document.addEventListener("visibilitychange",()=>{
document.hidden?stopPacketFetcher():startPacketFetcher();
});
map.on("overlayadd", e => {
if(e.layer === nodeLayer) updateNodeVisibility();
if(e.layer === trafficHeatLayer) loadTrafficHeat();
});
map.on("overlayremove", e => {
if(e.layer === nodeLayer) clearActiveBlinks();
});
async function waitForConfig() {
while (typeof window._siteConfigPromise === "undefined") {
await new Promise(r => setTimeout(r, 100));
@@ -473,6 +513,9 @@ fetch('/api/nodes?days_active=3')
renderNodesOnMap();
createChannelFilters();
if(map.hasLayer(trafficHeatLayer)){
loadTrafficHeat();
}
})
.catch(console.error);
@@ -493,7 +536,7 @@ function renderNodesOnMap(){
fillColor: color,
fillOpacity: 1,
weight: 0.7
}).addTo(map);
}).addTo(nodeLayer);
marker.nodeId = node.key;
marker.originalColor = color;
@@ -516,6 +559,56 @@ function renderNodesOnMap(){
setTimeout(() => applyTranslationsMap(), 50);
}
/* ======================================================
TRAFFIC HEAT LAYER
====================================================== */
async function loadTrafficHeat(){
if(trafficHeatLoaded || trafficHeatLoading) return;
if(nodeMap.size === 0) return;
trafficHeatLoading = true;
try {
const url = new URL("/api/packets", window.location.origin);
url.searchParams.set("limit", 1000);
const res = await fetch(url);
if(!res.ok) return;
const data = await res.json();
const counts = new Map();
(data.packets || []).forEach(pkt => {
const nodeId = pkt.from_node_id;
if(nodeId == null) return;
counts.set(nodeId, (counts.get(nodeId) || 0) + 1);
});
const maxCount = Math.max(...counts.values(), 0);
const heatPoints = [];
counts.forEach((count, nodeId) => {
const node = nodeMap.get(nodeId);
if(!node || isInvalidCoord(node)) return;
const markerLatLng = markerById[nodeId]?.getLatLng();
const lat = markerLatLng?.lat ?? node.lat;
const lng = markerLatLng?.lng ?? node.long;
const intensity = maxCount > 0 ? count / maxCount : 0;
heatPoints.push([lat, lng, Math.max(0.15, intensity)]);
});
trafficHeatLayer.setLatLngs(heatPoints);
trafficHeatLoaded = true;
} catch(err) {
console.error("Failed to load traffic heat layer:", err);
} finally {
trafficHeatLoading = false;
}
}
/* ======================================================
UNMAPPED PACKETS LIST
====================================================== */
@@ -654,13 +747,33 @@ function isTextPort(portnum){
return portnum === 1 || portnum === 7;
}
function isNodeVisibleForPackets(marker){
return map.hasLayer(nodeLayer) && nodeLayer.hasLayer(marker);
}
function clearBlinkForMarker(marker){
if(activeBlinks.has(marker)){
const interval = activeBlinks.get(marker);
clearInterval(interval);
activeBlinks.delete(marker);
}
marker.setStyle({ fillColor: marker.originalColor });
if(marker.tooltip){
map.removeLayer(marker.tooltip);
marker.tooltip = null;
}
}
function clearActiveBlinks(){
activeBlinks.forEach((interval, marker) => clearBlinkForMarker(marker));
activeBlinks.clear();
}
function blinkNode(marker,longName,portnum,payload){
if(!map.hasLayer(marker)) return;
if(!isNodeVisibleForPackets(marker)) return;
if(activeBlinks.has(marker)){
clearInterval(activeBlinks.get(marker));
marker.setStyle({ fillColor: marker.originalColor });
if(marker.tooltip) map.removeLayer(marker.tooltip);
clearBlinkForMarker(marker);
}
let blinkCount = 0;
@@ -688,19 +801,19 @@ function blinkNode(marker,longName,portnum,payload){
marker.tooltip = tooltip;
const interval = setInterval(()=>{
if(map.hasLayer(marker)){
if(isNodeVisibleForPackets(marker)){
marker.setStyle({
fillColor: blinkCount%2===0 ? 'yellow' : marker.originalColor
});
marker.bringToFront();
} else {
clearBlinkForMarker(marker);
return;
}
blinkCount++;
if(Date.now() - blinkStart > blinkDurationMs){
clearInterval(interval);
marker.setStyle({ fillColor: marker.originalColor });
map.removeLayer(tooltip);
activeBlinks.delete(marker);
clearBlinkForMarker(marker);
}
},500);
@@ -760,6 +873,8 @@ function saveFiltersToLocalStorage(){
}
function updateNodeVisibility(){
if(!map.hasLayer(nodeLayer)) return;
const routerOnly = document.getElementById("filter-routers-only").checked;
const mqttOnly = document.getElementById("filter-mqtt-only").checked;
const activeChannels = [...channelSet].filter(ch =>
@@ -774,7 +889,12 @@ function updateNodeVisibility(){
(!mqttOnly || n.is_mqtt_gateway) &&
activeChannels.includes(n.channel);
visible ? map.addLayer(marker) : map.removeLayer(marker);
if(visible){
nodeLayer.addLayer(marker);
} else {
clearBlinkForMarker(marker);
nodeLayer.removeLayer(marker);
}
}
});
}
+12 -5
View File
@@ -1071,18 +1071,24 @@ async function loadPackets(filters = {}) {
const match = pkt.payload?.match(/ID:\s*(\d+)/i);
if (match) traceId = match[1];
inlineLinks +=
` <a class="inline-link" href="/graph/traceroute/${traceId}" target="_blank">⮕</a>`;
` <a class="inline-link" href="/traceroute/${traceId}" target="_blank">⮕</a>`;
}
const sizeBytes = packetSizeBytes(pkt);
const packetIdCell = pkt.packet_id
? `<a href="/packet/${pkt.storage_id || pkt.id}"
style="text-decoration:underline; color:inherit;">
${pkt.packet_id}
</a>`
: pkt.portnum === 73
? `<span class="text-secondary" title="MQTT map report">Not a Packet</span>`
: `<span class="text-secondary" title="No Meshtastic packet ID">—</span>`;
list.insertAdjacentHTML("afterbegin", `
<tr class="packet-row">
<td>${localTime}</td>
<td><span class="toggle-btn">▶</span>
<a href="/packet/${pkt.id}" style="text-decoration:underline; color:inherit;">
${pkt.id}
</a>
${packetIdCell}
</td>
<td>${fromCell}</td>
<td>${toCell}</td>
@@ -1706,6 +1712,7 @@ async function loadNodeStats(nodeId) {
: "";
const portName = PORT_LABEL_MAP[pkt.portnum] || `Port ${pkt.portnum}`;
const displayPacketId = pkt.packet_id || (pkt.portnum === 73 ? "Not a Packet" : "");
// Escape quotes + line breaks for CSV safety
const payload = (pkt.payload || "")
@@ -1714,7 +1721,7 @@ async function loadNodeStats(nodeId) {
rows.push([
time,
pkt.id,
displayPacketId,
pkt.from_node_id,
pkt.to_node_id,
pkt.portnum,
+6 -1
View File
@@ -151,6 +151,11 @@ async function loadChannels() {
const sel = document.getElementById("channelFilter");
sel.innerHTML = "";
const allOpt = document.createElement("option");
allOpt.value = "";
allOpt.textContent = topTranslations.all_channels || "All channels";
sel.appendChild(allOpt);
for (const ch of data.channels || []) {
const opt = document.createElement("option");
opt.value = ch;
@@ -158,7 +163,7 @@ async function loadChannels() {
sel.appendChild(opt);
}
sel.value = "MediumFast";
sel.value = "";
}
/* ======================================================
+138 -13
View File
@@ -7,12 +7,53 @@
{% block css %}
#traceroute-graph {
width: 100%;
height: 85vh;
height: calc(100vh - 270px);
border: 1px solid #2a2f36;
background: linear-gradient(135deg, #0f1216 0%, #171b22 100%);
border-radius: 10px;
}
#traceroute-graph-wrap {
position: relative;
}
#traceroute-legend {
position: absolute;
right: 16px;
bottom: 16px;
z-index: 2;
display: grid;
gap: 8px;
padding: 10px 12px;
color: #d6dde6;
background: rgba(15, 18, 22, 0.88);
border: 1px solid #303743;
border-radius: 8px;
font-size: 13px;
}
.traceroute-legend-row {
display: grid;
grid-template-columns: 34px auto;
align-items: center;
gap: 8px;
white-space: nowrap;
}
.traceroute-legend-line {
height: 0;
border-top: 3px solid var(--line-color);
}
.traceroute-legend-line.muted {
border-top-width: 2px;
opacity: 0.55;
}
.traceroute-legend-line.dashed {
border-top-style: dashed;
}
#traceroute-meta {
padding: 12px 16px;
color: #c8d0da;
@@ -28,7 +69,23 @@
<div><b>Traceroute</b> <span id="traceroute-title"></span></div>
<div id="traceroute-error"></div>
</div>
<div id="traceroute-graph"></div>
<div id="traceroute-graph-wrap">
<div id="traceroute-graph"></div>
<div id="traceroute-legend" aria-label="Traceroute edge legend">
<div class="traceroute-legend-row">
<span class="traceroute-legend-line" style="--line-color: #34c759"></span>
<span>Winning forward</span>
</div>
<div class="traceroute-legend-row">
<span class="traceroute-legend-line dashed" style="--line-color: #ff3b30"></span>
<span>Reported reverse</span>
</div>
<div class="traceroute-legend-row">
<span class="traceroute-legend-line muted" style="--line-color: #7c858f"></span>
<span>Reported route</span>
</div>
</div>
</div>
<script>
const el = document.getElementById("traceroute-graph");
@@ -49,6 +106,35 @@ function addPathEdges(path, edges, style) {
}
}
function pathKey(path) {
return path.map(id => String(id)).join(">");
}
function addMissingEndpoints(path, startId, endId) {
const fullPath = path.map(id => String(id));
if (startId && (!fullPath.length || fullPath[0] !== startId)) {
fullPath.unshift(startId);
}
if (endId && (!fullPath.length || fullPath[fullPath.length - 1] !== endId)) {
fullPath.push(endId);
}
return fullPath;
}
// These lists are deduplicated across observations. A forward path is only known to
// reach the target if some response reported it; a reverse path carries no such marker.
function uniqueForwardPaths(data, originId, targetId) {
return (data?.unique_forward_paths || [])
.map(item => addMissingEndpoints(item.path || [], originId, item.complete ? targetId : null))
.filter(path => path.length > 1);
}
function uniqueReversePaths(data, originId, targetId) {
return (data?.unique_reverse_paths || [])
.map(path => addMissingEndpoints(path || [], targetId, null))
.filter(path => path.length > 1);
}
async function loadTraceroute() {
const packetId = packetIdFromPath();
document.getElementById("traceroute-title").textContent = `#${packetId}`;
@@ -73,26 +159,62 @@ async function loadTraceroute() {
const nodes = new Map();
const edges = [];
const forwardPaths = data?.winning_paths?.forward || [];
const reversePaths = data?.winning_paths?.reverse || [];
const originId = data?.packet?.from != null ? String(data.packet.from) : null;
const targetId = data?.packet?.to != null ? String(data.packet.to) : null;
// A response packet carries from/to reversed, so use the endpoints the API resolved.
const originId = data?.packet?.initiator != null ? String(data.packet.initiator) : null;
const targetId = data?.packet?.target != null ? String(data.packet.target) : null;
const forwardPaths = (data?.winning_paths?.forward || []).map(path => path.map(id => String(id)));
const reversePaths = (data?.winning_paths?.reverse || []).map(path => path.map(id => String(id)));
const winningPathKeys = new Set([
...forwardPaths.map(pathKey),
...reversePaths.map(pathKey)
]);
const winningForwardNodeIds = new Set(forwardPaths.flat());
const hasWinningReversePath = reversePaths.some(path => path.length > 1);
const winningForwardPath = forwardPaths.find(path => path.length > 1);
const colorStartId = winningForwardPath ? winningForwardPath[0] : originId;
const colorEndId = winningForwardPath ? winningForwardPath[winningForwardPath.length - 1] : targetId;
const startNodeColor = hasWinningReversePath ? "#ff3b30" : "#34c759";
const endNodeColor = hasWinningReversePath ? "#34c759" : "#ff3b30";
uniqueForwardPaths(data, originId, targetId).forEach(path => {
if (winningPathKeys.has(pathKey(path))) {
return;
}
path.forEach(id => nodes.set(String(id), { name: String(id) }));
addPathEdges(path, edges, { color: "#7c858f", width: 1.5, opacity: 0.45 });
});
uniqueReversePaths(data, originId, targetId).forEach(path => {
if (winningPathKeys.has(pathKey(path))) {
return;
}
path.forEach(id => nodes.set(String(id), { name: String(id) }));
addPathEdges(path, edges, { color: "#7c858f", width: 1.5, opacity: 0.45, type: "dashed" });
});
forwardPaths.forEach(path => {
path.forEach(id => nodes.set(String(id), { name: String(id) }));
addPathEdges(path, edges, { color: "#ff5733", width: 3 });
addPathEdges(path, edges, { color: "#34c759", width: 3 });
});
reversePaths.forEach(path => {
path.forEach(id => nodes.set(String(id), { name: String(id) }));
addPathEdges(path, edges, { color: "#00c3ff", width: 2, type: "dashed" });
addPathEdges(path, edges, { color: "#ff3b30", width: 2, type: "dashed" });
});
const graphNodes = Array.from(nodes.values()).map(n => {
const isOrigin = originId && n.name === originId;
const isTarget = targetId && n.name === targetId;
const color = isOrigin ? "#ff3b30" : isTarget ? "#34c759" : "#8aa4c8";
const size = isOrigin || isTarget ? 44 : 36;
const isColorStart = colorStartId && n.name === colorStartId;
const isColorEnd = colorEndId && n.name === colorEndId;
const isWinningForwardNode = winningForwardNodeIds.has(n.name);
let color = "#3f454d";
if (isColorStart) {
color = startNodeColor;
} else if (isColorEnd) {
color = endNodeColor;
} else if (isWinningForwardNode) {
color = "#b88a2a";
}
const size = isColorStart || isColorEnd ? 44 : 36;
return {
id: n.name,
name: nodeShortNameById.get(n.name) || n.name,
@@ -104,7 +226,10 @@ async function loadTraceroute() {
fontWeight: "bold"
},
tooltip: {
formatter: () => nodeLongNameById.get(n.name) || n.name
formatter: () => {
const longName = nodeLongNameById.get(n.name) || n.name;
return longName === n.name ? n.name : `${longName} (${n.name})`;
}
}
};
});
+55
View File
@@ -0,0 +1,55 @@
"""Assemble node paths from Meshtastic RouteDiscovery payloads.
A traceroute response carries the endpoints reversed with respect to the path
being traced: `to` is the node that started the traceroute and `from` is the node
that answered. Only the response direction is affected, so both orientations are
handled here rather than at each call site.
"""
# Firmware inserts this into route[] as a placeholder for a hop it could not identify.
NODENUM_BROADCAST = 0xFFFFFFFF
def return_is_complete(route):
"""True when the response was seen after it reached the node that started it.
Relays append both their ID and their SNR, but the final recipient appends only its
SNR, so one spare SNR entry means the return trip finished. Gateways usually report
a response that is still in flight, where the remaining hops are not known yet.
"""
return len(route.snr_back) == len(route.route_back) + 1
def forward_path(from_node_id, to_node_id, hops, done, gateway_node_id=None):
"""Path towards the traced node, starting at the node that began the traceroute."""
if done:
return [node for node in (to_node_id, *hops, from_node_id) if node is not None]
path = [node for node in (from_node_id, *hops) if node is not None]
# Some nodes add themselves to the route before uplinking.
if gateway_node_id is not None and (not path or path[-1] != gateway_node_id):
path.append(gateway_node_id)
return path
def return_path(from_node_id, to_node_id, hops, complete):
"""Path back from the traced node, starting at the node that answered."""
path = [node for node in (from_node_id, *hops) if node is not None]
if complete and to_node_id is not None:
path.append(to_node_id)
return path
def edges(path):
"""Consecutive pairs of a path, skipping links that run through an unknown hop."""
for a, b in zip(path, path[1:], strict=False):
if NODENUM_BROADCAST in (a, b):
continue
yield a, b
def endpoints(from_node_id, to_node_id, done):
"""The (initiator, target) of the traceroute, regardless of packet direction."""
if done:
return to_node_id, from_node_id
return from_node_id, to_node_id
+29 -8
View File
@@ -16,8 +16,9 @@ from google.protobuf.message import Message
from jinja2 import Environment, PackageLoader, Undefined, select_autoescape
from markupsafe import Markup
from meshtastic.protobuf.mesh_pb2 import Routing
from meshtastic.protobuf.portnums_pb2 import PortNum
from meshview import config, database, decode_payload, migrations, models, store
from meshview import config, database, decode_payload, migrations, models, store, traceroute
from meshview.__version__ import (
__version_string__,
)
@@ -53,6 +54,7 @@ class Packet:
"""UI-friendly packet wrapper for templates and API payloads."""
id: int
packet_id: int | None
from_node_id: int
from_node: models.Node
to_node_id: int
@@ -80,6 +82,17 @@ class Packet:
if payload is None:
text_payload = "Did not decode"
elif (
packet.portnum == PortNum.ROUTING_APP
and isinstance(payload, Routing)
and payload.WhichOneof("variant") == "error_reason"
):
# error_reason NONE is the implicit ACK, not a failure.
text_payload = (
"ACK"
if payload.error_reason == Routing.Error.NONE
else f"NAK: {Routing.Error.Name(payload.error_reason)}"
)
elif isinstance(payload, Message):
text_payload = text_format.MessageToString(payload)
elif packet.portnum == PortNum.TEXT_MESSAGE_APP and packet.to_node_id != 0xFFFFFFFF:
@@ -99,8 +112,13 @@ class Packet:
f'<a href="https://www.google.com/maps/search/?api=1&query={payload.latitude_i * 1e-7},{payload.longitude_i * 1e-7}" target="_blank">map</a>'
)
packet_id = None
if mesh_packet and mesh_packet.id:
packet_id = mesh_packet.id
return cls(
id=packet.id,
packet_id=packet_id,
from_node=packet.from_node,
from_node_id=packet.from_node_id,
to_node=packet.to_node,
@@ -389,14 +407,17 @@ async def graph_traceroute(request):
route = decode_payload.decode_payload(PortNum.TRACEROUTE_APP, tr.route)
if route is None:
continue
path = [packet.from_node_id]
path.extend(route.route)
path = traceroute.forward_path(
packet.from_node_id,
packet.to_node_id,
route.route,
tr.done,
tr.gateway_node_id,
)
if not path:
continue
if tr.done:
dest = packet.to_node_id
path.append(packet.to_node_id)
elif path[-1] != tr.gateway_node_id:
# It seems some nodes add them self to the list before uplinking
path.append(tr.gateway_node_id)
dest = path[-1]
if not tr.done and tr.gateway_node_id not in node_seen_time and tr.import_time_us:
node_seen_time[path[-1]] = tr.import_time_us
+127 -23
View File
@@ -10,7 +10,7 @@ from aiohttp import web
from sqlalchemy import func, select
from meshtastic.protobuf.portnums_pb2 import PortNum
from meshview import database, decode_payload, store
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
@@ -38,6 +38,96 @@ _LANG_CACHE = {}
routes = web.RouteTableDef()
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)
@@ -160,6 +250,8 @@ async def api_packets(request):
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,
@@ -249,6 +341,8 @@ async def api_packets(request):
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,
@@ -536,10 +630,15 @@ async def api_edges(request):
if route is None or tr.packet is None:
continue
path = [tr.packet.from_node_id] + list(route.route)
path.append(tr.packet.to_node_id if tr.done else tr.gateway_node_id)
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 zip(path, path[1:], strict=False):
for a, b in traceroute.edges(path):
if (a, b) not in edges:
edges[(a, b)] = "traceroute"
edges_added_tr += 1
@@ -720,6 +819,8 @@ async def health_check(request):
"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
@@ -849,6 +950,7 @@ async def api_traceroute(request):
"done": tr.done,
"forward_hops": forward_list,
"reverse_hops": reverse_list,
"reverse_complete": traceroute.return_is_complete(route),
}
)
@@ -859,6 +961,7 @@ async def api_traceroute(request):
forward_paths = []
reverse_paths = []
forward_complete = set()
winning_forward_paths = []
winning_reverse_paths = []
@@ -872,11 +975,11 @@ async def api_traceroute(request):
if tr["reverse_hops"]:
reverse_paths.append(r)
# A response proves the request reached the target, whatever its return trip did.
if tr["done"]:
if tr["forward_hops"]:
winning_forward_paths.append(f)
if tr["reverse_hops"]:
winning_reverse_paths.append(r)
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))
@@ -887,30 +990,26 @@ async def api_traceroute(request):
# Convert for JSON output
unique_forward_paths_json = [
{"path": list(p), "count": forward_counts[p]} for p in unique_forward_paths
{"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 path in set(winning_forward_paths):
full_path = list(path)
if from_node_id is not None and (not full_path or full_path[0] != from_node_id):
full_path = [from_node_id, *full_path]
if to_node_id is not None and (not full_path or full_path[-1] != to_node_id):
full_path = [*full_path, to_node_id]
winning_forward_with_endpoints.append(full_path)
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 path in set(winning_reverse_paths):
full_path = list(path)
if to_node_id is not None and (not full_path or full_path[0] != to_node_id):
full_path = [to_node_id, *full_path]
if from_node_id is not None and (not full_path or full_path[-1] != from_node_id):
full_path = [*full_path, from_node_id]
winning_reverse_with_endpoints.append(full_path)
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,
@@ -920,12 +1019,17 @@ async def api_traceroute(request):
# --------------------------------------------
# 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,
+92 -2
View File
@@ -33,6 +33,8 @@ file_handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')
file_handler.setFormatter(formatter)
cleanup_logger.addHandler(file_handler)
cleanup_status_file = str(Path(cleanup_logfile).with_suffix(".status.json"))
backup_status_file = str(Path(cleanup_logfile).with_name("dbbackup.status.json"))
# -------------------------
@@ -49,6 +51,32 @@ def get_int(config, section, key, default=0):
return default
def _rowcount(value):
return value if value is not None and value >= 0 else None
def write_cleanup_status(status: dict) -> None:
status_path = Path(cleanup_status_file)
tmp_path = status_path.with_suffix(f"{status_path.suffix}.tmp")
try:
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(status, f, indent=2)
tmp_path.replace(status_path)
except Exception as e:
cleanup_logger.warning(f"Failed to write cleanup status file: {e}")
def write_backup_status(status: dict) -> None:
status_path = Path(backup_status_file)
tmp_path = status_path.with_suffix(f"{status_path.suffix}.tmp")
try:
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(status, f, indent=2)
tmp_path.replace(status_path)
except Exception as e:
cleanup_logger.warning(f"Failed to write backup status file: {e}")
# -------------------------
# Shared DB lock
# -------------------------
@@ -66,19 +94,40 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None:
database_url: SQLAlchemy connection string
backup_dir: Directory to store backups (default: current directory)
"""
backup_status = {
"status": "running",
"started_at": datetime.datetime.now(datetime.UTC).isoformat(),
"completed_at": None,
"backup_dir": backup_dir,
"database_path": None,
"backup_file": None,
"original_size_bytes": None,
"compressed_size_bytes": None,
"compression_percent": None,
"error": None,
}
write_backup_status(backup_status)
try:
url = make_url(database_url)
if not url.drivername.startswith("sqlite"):
cleanup_logger.warning("Backup only supported for SQLite databases")
backup_status["status"] = "unsupported"
backup_status["error"] = "Backup only supported for SQLite databases"
return
if not url.database or url.database == ":memory:":
cleanup_logger.error("Could not extract database path from connection string")
backup_status["status"] = "error"
backup_status["error"] = "Could not extract database path from connection string"
return
db_file = Path(url.database)
backup_status["database_path"] = str(db_file)
if not db_file.exists():
cleanup_logger.error(f"Database file not found: {db_file}")
backup_status["status"] = "error"
backup_status["error"] = f"Database file not found: {db_file}"
return
# Create backup directory if it doesn't exist
@@ -89,6 +138,7 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None:
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"{db_file.stem}_backup_{timestamp}.db.gz"
backup_file = backup_path / backup_filename
backup_status["backup_file"] = str(backup_file)
cleanup_logger.info(f"Creating backup: {backup_file}")
@@ -98,9 +148,15 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None:
shutil.copyfileobj(f_in, f_out)
# Get file sizes for logging
original_size = db_file.stat().st_size / (1024 * 1024) # MB
compressed_size = backup_file.stat().st_size / (1024 * 1024) # MB
original_size_bytes = db_file.stat().st_size
compressed_size_bytes = backup_file.stat().st_size
original_size = original_size_bytes / (1024 * 1024) # MB
compressed_size = compressed_size_bytes / (1024 * 1024) # MB
compression_ratio = (1 - compressed_size / original_size) * 100 if original_size > 0 else 0
backup_status["original_size_bytes"] = original_size_bytes
backup_status["compressed_size_bytes"] = compressed_size_bytes
backup_status["compression_percent"] = round(compression_ratio, 1)
backup_status["status"] = "ok"
cleanup_logger.info(
f"Backup created successfully: {backup_file.name} "
@@ -110,6 +166,11 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None:
except Exception as e:
cleanup_logger.error(f"Error creating database backup: {e}")
backup_status["status"] = "error"
backup_status["error"] = str(e)
finally:
backup_status["completed_at"] = datetime.datetime.now(datetime.UTC).isoformat()
write_backup_status(backup_status)
# -------------------------
@@ -179,6 +240,24 @@ async def daily_cleanup_at(
).replace(tzinfo=None)
cutoff_us = int(cutoff_dt.timestamp() * 1_000_000)
cleanup_logger.info(f"Running cleanup for records older than {cutoff_dt.isoformat()}...")
rows_deleted = {
"packet": None,
"packet_seen": None,
"traceroute": None,
"node": None,
}
cleanup_status = {
"status": "running",
"started_at": datetime.datetime.now(datetime.UTC).isoformat(),
"completed_at": None,
"cutoff_at": cutoff_dt.replace(tzinfo=datetime.UTC).isoformat(),
"days_to_keep": days_to_keep,
"vacuum_requested": vacuum_db,
"vacuum_completed": False,
"rows_deleted": rows_deleted,
"error": None,
}
write_cleanup_status(cleanup_status)
try:
async with db_lock: # Pause ingestion
@@ -191,6 +270,7 @@ async def daily_cleanup_at(
result = await session.execute(
delete(models.Packet).where(models.Packet.import_time_us < cutoff_us)
)
rows_deleted["packet"] = _rowcount(result.rowcount)
cleanup_logger.info(f"Deleted {result.rowcount} rows from Packet")
# -------------------------
@@ -201,6 +281,7 @@ async def daily_cleanup_at(
models.PacketSeen.import_time_us < cutoff_us
)
)
rows_deleted["packet_seen"] = _rowcount(result.rowcount)
cleanup_logger.info(f"Deleted {result.rowcount} rows from PacketSeen")
# -------------------------
@@ -211,6 +292,7 @@ async def daily_cleanup_at(
models.Traceroute.import_time_us < cutoff_us
)
)
rows_deleted["traceroute"] = _rowcount(result.rowcount)
cleanup_logger.info(f"Deleted {result.rowcount} rows from Traceroute")
# -------------------------
@@ -219,6 +301,7 @@ async def daily_cleanup_at(
result = await session.execute(
delete(models.Node).where(models.Node.last_seen_us < cutoff_us)
)
rows_deleted["node"] = _rowcount(result.rowcount)
cleanup_logger.info(f"Deleted {result.rowcount} rows from Node")
await session.commit()
@@ -228,14 +311,21 @@ async def daily_cleanup_at(
async with mqtt_database.engine.begin() as conn:
await conn.exec_driver_sql("VACUUM;")
cleanup_logger.info("VACUUM completed.")
cleanup_status["vacuum_completed"] = True
elif vacuum_db:
cleanup_logger.info("VACUUM skipped (not supported for this database).")
cleanup_logger.info("Cleanup completed successfully.")
cleanup_logger.info("Ingestion resumed after cleanup.")
cleanup_status["status"] = "ok"
except Exception as e:
cleanup_logger.error(f"Error during cleanup: {e}")
cleanup_status["status"] = "error"
cleanup_status["error"] = str(e)
finally:
cleanup_status["completed_at"] = datetime.datetime.now(datetime.UTC).isoformat()
write_cleanup_status(cleanup_status)
# -------------------------