mirror of
https://github.com/MarekWo/mc-webui.git
synced 2026-07-26 19:42:43 +02:00
feat(observer): MQTT publishing module (meshcore-packet-capture compatible)
New app/observer.py: ObserverManager owns per-broker paho-mqtt clients (own network threads, QoS 0, LWT + retained status) and the packet fan-out. Packet decode/hash/format functions are ports of upstream meshcore-packet-capture including its legacy path fallbacks and unknown-version quirk; validated against 10 live packets from the reference observer network (hash/route/type/len all byte-identical). Adds paho-mqtt==2.1.0. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
MarekWo
+571
@@ -0,0 +1,571 @@
|
||||
"""Observer — publish captured mesh packets to MQTT brokers.
|
||||
|
||||
mc-webui equivalent of the standalone meshcore-packet-capture project
|
||||
(github.com/agessaman/meshcore-packet-capture): every packet the device
|
||||
overhears (RX_LOG_DATA) is formatted in the exact same JSON shape and
|
||||
published to all user-configured MQTT brokers, so letsmesh-style
|
||||
analyzers and other existing consumers work unchanged.
|
||||
|
||||
Threading model:
|
||||
- DeviceManager's asyncio thread calls handle_raw_packet(): pure CPU
|
||||
(header parse + SHA-256 + json.dumps) plus paho's non-blocking
|
||||
publish() enqueue. No await, no network, no DB, no locks on this path.
|
||||
- Each broker gets its own paho client with its own network thread
|
||||
(loop_start), auto-reconnecting with backoff. QoS 0 everywhere, so a
|
||||
dead broker never accumulates queued messages.
|
||||
- Lifecycle (start/stop/reload) runs on Flask request threads or a
|
||||
dedicated daemon thread, serialized by _lifecycle_lock. The client
|
||||
list is swapped as a whole (atomic reference assignment), so the hot
|
||||
path can read it lock-free.
|
||||
|
||||
The packet parsing/formatting functions below are ports of the upstream
|
||||
implementation (src/meshcore_packet_capture/packet_capture.py) including
|
||||
its quirks — e.g. packets with an unknown payload version still get
|
||||
published, with default route "U" / packet_type "0" — because downstream
|
||||
consumers may rely on them.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.version import VERSION_STRING
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Wire route_type (header & 0x03): TRANSPORT_FLOOD, FLOOD, DIRECT, TRANSPORT_DIRECT
|
||||
_ROUTE_LETTERS = {0x00: "F", 0x01: "F", 0x02: "D", 0x03: "T"}
|
||||
_TRANSPORT_ROUTES = (0x00, 0x03) # carry a 4-byte transport code after the header
|
||||
_EMPTY_HASH = "0000000000000000"
|
||||
|
||||
|
||||
def _decode_packed_path_length(path_len_byte: int, max_path_size: int = 64) -> tuple:
|
||||
"""Decode the packed path_len byte per MeshCore firmware.
|
||||
|
||||
Low 6 bits = hop count, high 2 bits = bytes-per-hop minus 1.
|
||||
Returns (path_byte_len, bytes_per_hop). Keeps upstream's two legacy
|
||||
fallbacks (reserved 4-bytes/hop mode, oversized path) so decoded
|
||||
output matches meshcore-packet-capture exactly.
|
||||
"""
|
||||
hop_count = path_len_byte & 0x3F
|
||||
bytes_per_hop = (path_len_byte >> 6) + 1
|
||||
if bytes_per_hop == 4: # reserved mode -> legacy one-byte interpretation
|
||||
return path_len_byte, 1
|
||||
path_byte_len = hop_count * bytes_per_hop
|
||||
if path_byte_len > max_path_size: # invalid packed value -> legacy
|
||||
return path_len_byte, 1
|
||||
return path_byte_len, bytes_per_hop
|
||||
|
||||
|
||||
def _split_path_hops(path_bytes: bytes, bytes_per_hop: int) -> list:
|
||||
"""Split path bytes into per-hop hex tokens (1-3 bytes per hop)."""
|
||||
path_hex = path_bytes.hex()
|
||||
hop_hex_chars = max(bytes_per_hop, 1) * 2
|
||||
nodes = [path_hex[i:i + hop_hex_chars] for i in range(0, len(path_hex), hop_hex_chars)]
|
||||
if (len(path_hex) % hop_hex_chars) != 0:
|
||||
nodes = [path_hex[i:i + 2] for i in range(0, len(path_hex), 2)]
|
||||
return nodes
|
||||
|
||||
|
||||
def decode_packet(raw_hex: str):
|
||||
"""Decode packet header/path — port of upstream decode_and_publish_message().
|
||||
|
||||
Returns None for truncated packets and unknown payload versions
|
||||
(those still get published, with default route/type). ADVERT payload
|
||||
contents are not parsed: upstream never publishes them in the
|
||||
packets payload.
|
||||
"""
|
||||
try:
|
||||
byte_data = bytes.fromhex(raw_hex)
|
||||
except ValueError:
|
||||
return None
|
||||
if len(byte_data) < 2:
|
||||
return None
|
||||
header = byte_data[0]
|
||||
route_type = header & 0x03
|
||||
has_transport = route_type in _TRANSPORT_ROUTES
|
||||
offset = 1
|
||||
if has_transport:
|
||||
offset += 4
|
||||
if len(byte_data) <= offset:
|
||||
return None
|
||||
path_len_byte = byte_data[offset]
|
||||
offset += 1
|
||||
path_byte_len, bytes_per_hop = _decode_packed_path_length(path_len_byte)
|
||||
if len(byte_data) < offset + path_byte_len:
|
||||
return None
|
||||
path_bytes = byte_data[offset:offset + path_byte_len]
|
||||
if (header >> 6) & 0x03 != 0: # only payload version VER_1 (0) is understood
|
||||
return None
|
||||
return {
|
||||
'route_type': route_type,
|
||||
'payload_type': (header >> 2) & 0x0F,
|
||||
'path': _split_path_hops(path_bytes, bytes_per_hop),
|
||||
'path_byte_len': path_byte_len,
|
||||
'has_transport': has_transport,
|
||||
}
|
||||
|
||||
|
||||
def calculate_packet_hash(raw_hex: str) -> str:
|
||||
"""Packet identity hash — port of firmware Packet::calculatePacketHash().
|
||||
|
||||
SHA256(payload_type byte || [packed path_len as uint16 LE, TRACE only]
|
||||
|| payload bytes), first 16 hex chars uppercase.
|
||||
"""
|
||||
try:
|
||||
byte_data = bytes.fromhex(raw_hex)
|
||||
header = byte_data[0]
|
||||
payload_type = (header >> 2) & 0x0F
|
||||
offset = 1
|
||||
if (header & 0x03) in _TRANSPORT_ROUTES:
|
||||
offset += 4
|
||||
if len(byte_data) <= offset:
|
||||
return _EMPTY_HASH
|
||||
path_len_byte = byte_data[offset]
|
||||
offset += 1
|
||||
path_byte_len, _ = _decode_packed_path_length(path_len_byte)
|
||||
payload_start = offset + path_byte_len
|
||||
if payload_start > len(byte_data):
|
||||
return _EMPTY_HASH
|
||||
hash_obj = hashlib.sha256()
|
||||
hash_obj.update(bytes([payload_type]))
|
||||
if payload_type == 9: # TRACE: firmware hashes the path_len as uint16 too
|
||||
hash_obj.update(path_len_byte.to_bytes(2, byteorder='little'))
|
||||
hash_obj.update(byte_data[payload_start:])
|
||||
return hash_obj.hexdigest()[:16].upper()
|
||||
except Exception:
|
||||
return _EMPTY_HASH
|
||||
|
||||
|
||||
def format_packet_data(raw_hex: str, origin: str, origin_id: str,
|
||||
snr=None, rssi=None, payload_length=None) -> dict:
|
||||
"""Build the letsmesh-compatible packet JSON — port of upstream (all values strings)."""
|
||||
now = datetime.now(timezone.utc)
|
||||
decoded = decode_packet(raw_hex)
|
||||
packet_len = len(raw_hex) // 2
|
||||
route = "U"
|
||||
packet_type = "0"
|
||||
payload_len = "0"
|
||||
if decoded:
|
||||
route = _ROUTE_LETTERS.get(decoded['route_type'], "U")
|
||||
packet_type = str(decoded['payload_type'])
|
||||
if payload_length is not None:
|
||||
payload_len = str(payload_length)
|
||||
else:
|
||||
transport_bytes = 4 if decoded['has_transport'] else 0
|
||||
payload_len = str(max(0, packet_len - 1 - transport_bytes - 1 - decoded['path_byte_len']))
|
||||
packet = {
|
||||
"origin": origin,
|
||||
"origin_id": (origin_id or '').upper(),
|
||||
"timestamp": now.isoformat(),
|
||||
"type": "PACKET",
|
||||
"direction": "rx",
|
||||
"time": now.strftime("%H:%M:%S"),
|
||||
"date": now.strftime("%d/%m/%Y"),
|
||||
"len": str(packet_len),
|
||||
"packet_type": packet_type,
|
||||
"route": route,
|
||||
"payload_len": payload_len,
|
||||
"raw": raw_hex.upper(),
|
||||
"SNR": str(snr) if snr is not None else "Unknown",
|
||||
"RSSI": str(rssi) if rssi is not None else "Unknown",
|
||||
"hash": calculate_packet_hash(raw_hex),
|
||||
}
|
||||
if route == "D" and decoded:
|
||||
packet["path"] = ",".join(decoded['path'])
|
||||
return packet
|
||||
|
||||
|
||||
class ObserverManager:
|
||||
"""Owns MQTT clients and the packet fan-out for the Observer feature."""
|
||||
|
||||
SETTINGS_KEY = 'observer_settings'
|
||||
LAST_ADVERT_KEY = 'observer_last_advert_at'
|
||||
DEFAULT_SETTINGS = {'enabled': False, 'iata': '', 'advert_interval_hours': 0}
|
||||
STATUS_EMIT_MIN_INTERVAL = 2.0 # seconds between observer_status socket emits
|
||||
|
||||
def __init__(self, db, socketio=None):
|
||||
self.db = db
|
||||
self.socketio = socketio
|
||||
self.device_manager = None # set by main.py once DeviceManager exists
|
||||
self._lifecycle_lock = threading.Lock()
|
||||
self._clients = [] # per-broker dicts; list ref swapped atomically
|
||||
self._started = False
|
||||
self._enabled = False
|
||||
self._iata = ''
|
||||
self._origin = None
|
||||
self._origin_id = None # uppercase device public key
|
||||
self._status_extras = {'model': 'unknown', 'firmware_version': 'unknown',
|
||||
'radio': 'unknown'}
|
||||
self._packets_seen = 0
|
||||
self._packets_published = 0
|
||||
self._last_status_emit = 0.0
|
||||
|
||||
# ---------------- settings ----------------
|
||||
|
||||
def get_settings(self) -> dict:
|
||||
settings = dict(self.DEFAULT_SETTINGS)
|
||||
stored = self.db.get_setting_json(self.SETTINGS_KEY, {}) or {}
|
||||
for key in settings:
|
||||
if key in stored:
|
||||
settings[key] = stored[key]
|
||||
return settings
|
||||
|
||||
# ---------------- identity (from DeviceManager) ----------------
|
||||
|
||||
def set_identity(self, name, public_key, self_info=None, device_info=None):
|
||||
"""Record device identity; called from DeviceManager after connect."""
|
||||
new_origin_id = (public_key or '').upper()
|
||||
identity_changed = self._started and new_origin_id != self._origin_id
|
||||
self._origin = name or 'MeshCore Device'
|
||||
self._origin_id = new_origin_id
|
||||
if self_info:
|
||||
self._status_extras['radio'] = ",".join(
|
||||
str(self_info.get(key, 0))
|
||||
for key in ('radio_freq', 'radio_bw', 'radio_sf', 'radio_cr'))
|
||||
if device_info:
|
||||
# Mirrors upstream get_firmware_info(): fw ver >= 3 carries strings
|
||||
fw_ver = device_info.get('fw ver', 0) or 0
|
||||
if fw_ver >= 3:
|
||||
version = str(device_info.get('ver', 'Unknown'))
|
||||
if version.startswith('v'):
|
||||
version = version[1:]
|
||||
self._status_extras['model'] = device_info.get('model', 'Unknown')
|
||||
self._status_extras['firmware_version'] = (
|
||||
f"v{version} (Build: {device_info.get('fw_build', 'Unknown')})")
|
||||
else:
|
||||
self._status_extras['firmware_version'] = f"v{fw_ver}"
|
||||
if identity_changed:
|
||||
logger.info("Observer: device identity changed, reloading MQTT clients")
|
||||
self.reload()
|
||||
|
||||
# ---------------- lifecycle ----------------
|
||||
|
||||
def ensure_started(self):
|
||||
"""Start clients if enabled and not yet running (device just connected)."""
|
||||
if self._started:
|
||||
return
|
||||
self.reload()
|
||||
|
||||
def reload(self):
|
||||
"""Re-read config and rebuild MQTT clients on a daemon thread.
|
||||
|
||||
loop_stop() joins paho's network thread and can take seconds when a
|
||||
client is mid-reconnect, so this must never run on a Flask worker
|
||||
or the DeviceManager loop.
|
||||
"""
|
||||
threading.Thread(target=self._reload_sync, name='observer-reload',
|
||||
daemon=True).start()
|
||||
|
||||
def stop(self, publish_offline=True):
|
||||
with self._lifecycle_lock:
|
||||
clients = self._clients
|
||||
self._clients = []
|
||||
self._started = False
|
||||
self._stop_clients(clients, publish_offline=publish_offline)
|
||||
|
||||
def _reload_sync(self):
|
||||
with self._lifecycle_lock:
|
||||
try:
|
||||
settings = self.get_settings()
|
||||
self._enabled = bool(settings.get('enabled'))
|
||||
self._iata = str(settings.get('iata') or '').strip()
|
||||
old_clients = self._clients
|
||||
self._clients = []
|
||||
self._started = False
|
||||
self._stop_clients(old_clients, publish_offline=True)
|
||||
if not self._enabled:
|
||||
return
|
||||
if not self._origin_id:
|
||||
logger.info("Observer enabled, waiting for device connection "
|
||||
"(identity unknown yet)")
|
||||
return
|
||||
new_clients = []
|
||||
for row in self.db.list_observer_brokers():
|
||||
if row.get('is_disabled'):
|
||||
continue
|
||||
try:
|
||||
new_clients.append(self._build_client(row))
|
||||
except Exception as e:
|
||||
logger.error(f"Observer: failed to set up MQTT client for "
|
||||
f"broker '{row.get('name')}': {e}")
|
||||
self._clients = new_clients
|
||||
self._started = bool(new_clients)
|
||||
if new_clients:
|
||||
logger.info(f"Observer running with {len(new_clients)} MQTT client(s), "
|
||||
f"packets topic: {self._topic('packets')}")
|
||||
else:
|
||||
logger.info("Observer enabled but no active brokers configured")
|
||||
except Exception as e:
|
||||
logger.error(f"Observer reload failed: {e}", exc_info=True)
|
||||
finally:
|
||||
self._emit_status(force=True)
|
||||
|
||||
def _stop_clients(self, clients, publish_offline=True):
|
||||
for ci in clients:
|
||||
client = ci['client']
|
||||
try:
|
||||
if publish_offline and client.is_connected():
|
||||
msg = client.publish(ci['status_topic'],
|
||||
json.dumps(self._status_payload('offline')),
|
||||
qos=0, retain=True)
|
||||
try:
|
||||
msg.wait_for_publish(timeout=2)
|
||||
except Exception:
|
||||
pass
|
||||
client.disconnect()
|
||||
client.loop_stop()
|
||||
except Exception as e:
|
||||
logger.debug(f"Observer: error stopping MQTT client "
|
||||
f"'{ci.get('name')}': {e}")
|
||||
|
||||
def _build_client(self, row) -> dict:
|
||||
# Lazy import: a missing paho package must not break app startup —
|
||||
# the observer just stays inert and reports the error.
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
status_topic = self._topic('status')
|
||||
base = re.sub(r"[^a-zA-Z0-9_-]", "",
|
||||
f"meshcore_client_{(self._origin_id or 'unknown')[:8]}")[:23]
|
||||
ci = {
|
||||
'broker_id': row['id'],
|
||||
'name': row['name'],
|
||||
'host': row['host'],
|
||||
'port': int(row['port'] or 1883),
|
||||
'status_topic': status_topic,
|
||||
'packets_topic': self._topic('packets'),
|
||||
'connected': False,
|
||||
'last_error': None,
|
||||
}
|
||||
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
|
||||
client_id=f"{base}_{row['id']}",
|
||||
clean_session=True)
|
||||
ci['client'] = client
|
||||
client.user_data_set(ci)
|
||||
client.reconnect_delay_set(min_delay=1, max_delay=120)
|
||||
if row.get('username'):
|
||||
client.username_pw_set(row['username'], row.get('password') or None)
|
||||
# LWT must be set before connect: retained offline status is the
|
||||
# authoritative "observer gone" signal (the app has no graceful
|
||||
# shutdown path — Docker SIGTERM just kills the process).
|
||||
client.will_set(status_topic, json.dumps(self._lwt_payload()),
|
||||
qos=0, retain=True)
|
||||
if row.get('use_tls'):
|
||||
import ssl
|
||||
if row.get('tls_verify', 1):
|
||||
client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
|
||||
else:
|
||||
client.tls_set(cert_reqs=ssl.CERT_NONE)
|
||||
client.tls_insecure_set(True)
|
||||
client.on_connect = self._on_connect_cb
|
||||
client.on_disconnect = self._on_disconnect_cb
|
||||
client.on_connect_fail = self._on_connect_fail_cb
|
||||
client.connect_async(ci['host'], ci['port'], keepalive=60)
|
||||
client.loop_start()
|
||||
return ci
|
||||
|
||||
# ---------------- topics & payloads ----------------
|
||||
|
||||
def _topic(self, kind: str) -> str:
|
||||
if self._iata:
|
||||
return f"meshcore/{self._iata.upper()}/{self._origin_id}/{kind}"
|
||||
return f"meshcore/{kind}" # upstream classic fallback when IATA unset
|
||||
|
||||
def _status_payload(self, status: str) -> dict:
|
||||
return {
|
||||
'status': status,
|
||||
'timestamp': datetime.now(timezone.utc).isoformat(),
|
||||
'origin': self._origin or 'MeshCore Device',
|
||||
'origin_id': self._origin_id or 'DEVICE',
|
||||
'model': self._status_extras.get('model', 'unknown'),
|
||||
'firmware_version': self._status_extras.get('firmware_version', 'unknown'),
|
||||
'radio': self._status_extras.get('radio', 'unknown'),
|
||||
'client_version': f"mc-webui/{VERSION_STRING}",
|
||||
}
|
||||
|
||||
def _lwt_payload(self) -> dict:
|
||||
# Upstream's LWT is the minimal 4-field variant
|
||||
return {
|
||||
'status': 'offline',
|
||||
'timestamp': datetime.now(timezone.utc).isoformat(),
|
||||
'origin': self._origin or 'MeshCore Device',
|
||||
'origin_id': self._origin_id or 'DEVICE',
|
||||
}
|
||||
|
||||
# ---------------- hot path (DeviceManager asyncio thread) ----------------
|
||||
|
||||
def handle_raw_packet(self, raw_hex, snr=None, rssi=None, payload_length=None):
|
||||
"""Format one overheard packet and publish it to all connected brokers."""
|
||||
if not self._started:
|
||||
return
|
||||
clients = self._clients # atomic ref read; never mutated in place
|
||||
if not clients:
|
||||
return
|
||||
self._packets_seen += 1
|
||||
try:
|
||||
payload = json.dumps(format_packet_data(
|
||||
raw_hex, self._origin, self._origin_id,
|
||||
snr=snr, rssi=rssi, payload_length=payload_length))
|
||||
except Exception as e:
|
||||
logger.debug(f"Observer: packet formatting failed: {e}")
|
||||
return
|
||||
published = False
|
||||
for ci in clients:
|
||||
try:
|
||||
if ci['client'].is_connected():
|
||||
result = ci['client'].publish(ci['packets_topic'], payload,
|
||||
qos=0, retain=False)
|
||||
if result.rc == 0: # mqtt.MQTT_ERR_SUCCESS
|
||||
published = True
|
||||
except Exception as e:
|
||||
logger.debug(f"Observer: publish to '{ci['name']}' failed: {e}")
|
||||
if published:
|
||||
self._packets_published += 1
|
||||
self._emit_status()
|
||||
|
||||
# ---------------- MQTT callbacks (paho network threads) ----------------
|
||||
|
||||
def _on_connect_cb(self, client, userdata, flags, reason_code, properties=None):
|
||||
failed = getattr(reason_code, 'is_failure', None)
|
||||
if failed is None: # MQTTv3 int fallback
|
||||
failed = reason_code != 0
|
||||
if failed:
|
||||
userdata['connected'] = False
|
||||
userdata['last_error'] = str(reason_code)
|
||||
logger.warning(f"Observer: MQTT broker '{userdata['name']}' refused "
|
||||
f"connection: {reason_code}")
|
||||
else:
|
||||
userdata['connected'] = True
|
||||
userdata['last_error'] = None
|
||||
logger.info(f"Observer: connected to MQTT broker '{userdata['name']}' "
|
||||
f"({userdata['host']}:{userdata['port']})")
|
||||
try:
|
||||
# Retained online status; re-published on every auto-reconnect
|
||||
client.publish(userdata['status_topic'],
|
||||
json.dumps(self._status_payload('online')),
|
||||
qos=0, retain=True)
|
||||
except Exception as e:
|
||||
logger.debug(f"Observer: online status publish failed: {e}")
|
||||
self._emit_status(force=True)
|
||||
|
||||
def _on_disconnect_cb(self, client, userdata, disconnect_flags, reason_code,
|
||||
properties=None):
|
||||
userdata['connected'] = False
|
||||
reason = str(reason_code)
|
||||
if reason not in ('Success', '0'):
|
||||
userdata['last_error'] = reason
|
||||
logger.info(f"Observer: lost connection to MQTT broker "
|
||||
f"'{userdata['name']}' ({reason}), auto-reconnecting")
|
||||
self._emit_status(force=True)
|
||||
|
||||
def _on_connect_fail_cb(self, client, userdata):
|
||||
userdata['connected'] = False
|
||||
if not userdata.get('last_error'):
|
||||
userdata['last_error'] = 'connection failed'
|
||||
self._emit_status() # throttled: fires on every paho retry
|
||||
|
||||
# ---------------- status for UI ----------------
|
||||
|
||||
def _status_reason(self):
|
||||
if not self._enabled:
|
||||
return 'disabled'
|
||||
if not self._origin_id:
|
||||
return 'waiting for device connection'
|
||||
if not self._clients:
|
||||
return 'no active brokers configured'
|
||||
return None
|
||||
|
||||
def _live_status(self) -> dict:
|
||||
"""Status snapshot without DB access — safe for socket emits."""
|
||||
return {
|
||||
'enabled': self._enabled,
|
||||
'running': self._started,
|
||||
'reason': self._status_reason(),
|
||||
'origin': self._origin,
|
||||
'origin_id': self._origin_id,
|
||||
'iata': self._iata,
|
||||
'packets_seen': self._packets_seen,
|
||||
'packets_published': self._packets_published,
|
||||
'brokers': [{'id': ci['broker_id'], 'name': ci['name'],
|
||||
'connected': ci['connected'],
|
||||
'last_error': ci['last_error']} for ci in self._clients],
|
||||
}
|
||||
|
||||
def get_status(self) -> dict:
|
||||
"""Full status for the API: settings + DB broker rows merged with live state."""
|
||||
status = self._live_status()
|
||||
status['settings'] = self.get_settings()
|
||||
live_by_id = {b['id']: b for b in status.pop('brokers')}
|
||||
brokers = []
|
||||
for row in self.db.list_observer_brokers():
|
||||
live = live_by_id.get(row['id'], {})
|
||||
brokers.append({
|
||||
'id': row['id'],
|
||||
'name': row['name'],
|
||||
'host': row['host'],
|
||||
'port': row['port'],
|
||||
'username': row.get('username', ''),
|
||||
'has_password': bool(row.get('password')),
|
||||
'use_tls': bool(row.get('use_tls')),
|
||||
'tls_verify': bool(row.get('tls_verify', 1)),
|
||||
'is_disabled': bool(row.get('is_disabled')),
|
||||
'connected': live.get('connected', False),
|
||||
'last_error': live.get('last_error'),
|
||||
})
|
||||
status['brokers'] = brokers
|
||||
try:
|
||||
status['last_advert_at'] = int(float(
|
||||
self.db.get_setting(self.LAST_ADVERT_KEY) or 0)) or None
|
||||
except (TypeError, ValueError):
|
||||
status['last_advert_at'] = None
|
||||
return status
|
||||
|
||||
def _emit_status(self, force=False):
|
||||
if not self.socketio:
|
||||
return
|
||||
now = time.time()
|
||||
if not force and (now - self._last_status_emit) < self.STATUS_EMIT_MIN_INTERVAL:
|
||||
return
|
||||
self._last_status_emit = now
|
||||
try:
|
||||
self.socketio.emit('observer_status', self._live_status(),
|
||||
namespace='/chat')
|
||||
except Exception as e:
|
||||
logger.debug(f"Observer: status emit failed: {e}")
|
||||
|
||||
# ---------------- advert scheduler (APScheduler job) ----------------
|
||||
|
||||
def maybe_send_advert(self):
|
||||
"""Send a flood advert when the configured interval has elapsed.
|
||||
|
||||
Called every few minutes by APScheduler; the last-advert timestamp
|
||||
persists in app_settings so restarts don't re-advertise early.
|
||||
"""
|
||||
try:
|
||||
settings = self.get_settings()
|
||||
if not settings.get('enabled'):
|
||||
return
|
||||
interval_hours = int(settings.get('advert_interval_hours') or 0)
|
||||
if interval_hours <= 0:
|
||||
return
|
||||
dm = self.device_manager
|
||||
if not dm or not getattr(dm, 'is_connected', False):
|
||||
return
|
||||
last = float(self.db.get_setting(self.LAST_ADVERT_KEY) or 0)
|
||||
now = time.time()
|
||||
if now - last < interval_hours * 3600:
|
||||
return
|
||||
result = dm.send_advert(flood=True)
|
||||
if result and result.get('success'):
|
||||
self.db.set_setting(self.LAST_ADVERT_KEY, str(int(now)))
|
||||
logger.info(f"Observer: sent scheduled flood advert "
|
||||
f"(interval {interval_hours}h)")
|
||||
self._emit_status(force=True)
|
||||
else:
|
||||
logger.warning(f"Observer: scheduled flood advert failed: {result}")
|
||||
except Exception as e:
|
||||
logger.error(f"Observer: advert scheduler error: {e}")
|
||||
@@ -28,6 +28,9 @@ pycryptodome==3.21.0
|
||||
|
||||
# WebSocket support for console (threading mode - no gevent needed)
|
||||
flask-socketio==5.3.6
|
||||
|
||||
# Observer: MQTT packet publishing (meshcore-packet-capture compatible)
|
||||
paho-mqtt==2.1.0
|
||||
python-socketio==5.10.0
|
||||
python-engineio==4.8.1
|
||||
|
||||
|
||||
Reference in New Issue
Block a user