adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-06 17:03:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update OTA repeater stats to return correct uptime, airtime, packet counts, etc.

Browse Source 
- Introduced `total_rx_airtime_ms` in `AirtimeManager` to track received packet airtime.
- Added `record_rx` method to log received airtime in `AirtimeManager`.
- Updated `RepeaterHandler` to count received packets and log RX airtime using the new method.
- Enhanced statistics reporting in `get_stats` to include total received airtime.
- Updated `ProtocolRequestHelper` to include total RX airtime in the RepeaterStats structure for better monitoring.
This commit is contained in:
agessaman
2026-03-09 17:27:51 -07:00
parent ee97ff736a
commit 25c2a14a81
4 changed files with 126 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/airtime.py
+6
View File
@@ -24,6 +24,7 @@ class AirtimeManager:
self.tx_history = [] # [(timestamp, airtime_ms), ...]
self.window_size = 60 # seconds
self.total_airtime_ms = 0
self.total_rx_airtime_ms = 0
def calculate_airtime(
self,
@@ -110,6 +111,10 @@ class AirtimeManager:
self.total_airtime_ms += airtime_ms
logger.debug(f"TX recorded: {airtime_ms: .1f}ms (total: {self.total_airtime_ms: .0f}ms)")
def record_rx(self, airtime_ms: float):
"""Record received packet airtime (for total RX airtime stats)."""
self.total_rx_airtime_ms += airtime_ms
def get_stats(self) -> dict:
now = time.time()
self.tx_history = [(ts, at) for ts, at in self.tx_history if now - ts < self.window_size]
@@ -122,4 +127,5 @@ class AirtimeManager:
"max_airtime_ms": self.max_airtime_per_minute,
"utilization_percent": utilization,
"total_airtime_ms": self.total_airtime_ms,
"total_rx_airtime_ms": self.total_rx_airtime_ms,
}
repeater/engine.py
+43 -3
View File
@@ -100,6 +100,13 @@ class RepeaterHandler(BaseHandler):
self.recent_packets = []
self.max_recent_packets = 50
self.start_time = time.time()
# Flood/direct and duplicate counters (for GET_STATUS / firmware RepeaterStats)
self.recv_flood_count = 0
self.recv_direct_count = 0
self.sent_flood_count = 0
self.sent_direct_count = 0
self.flood_dup_count = 0
self.direct_dup_count = 0
# Storage collector for persistent packet logging
try:
@@ -131,7 +138,21 @@ class RepeaterHandler(BaseHandler):
if metadata is None:
metadata = {}
self.rx_count += 1
# Only count as receive when packet came from the radio (not locally injected)
if not local_transmission:
self.rx_count += 1
route_type = packet.header & PH_ROUTE_MASK
if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
self.recv_flood_count += 1
elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
self.recv_direct_count += 1
try:
rx_airtime_ms = self.airtime_mgr.calculate_airtime(packet.get_raw_length())
self.airtime_mgr.record_rx(rx_airtime_ms)
except Exception:
pass
route_type = packet.header & PH_ROUTE_MASK
# Check if we're in monitor mode (receive only, no forwarding)
mode = self.config.get("repeater", {}).get("mode", "forward")
@@ -291,9 +312,14 @@ class RepeaterHandler(BaseHandler):
pkt_hash = packet.calculate_packet_hash().hex().upper()
is_dupe = pkt_hash in self.seen_packets and not transmitted
# Set drop reason for duplicates
# Set drop reason for duplicates and count flood vs direct dups
if is_dupe and drop_reason is None:
drop_reason = "Duplicate"
if is_dupe:
if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
self.flood_dup_count += 1
elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
self.direct_dup_count += 1
display_hashes = (
original_path_hashes if original_path_hashes else packet.get_path_hashes_hex()
@@ -416,7 +442,6 @@ class RepeaterHandler(BaseHandler):
self.recent_packets.append(packet_record)
if len(self.recent_packets) > self.max_recent_packets:
self.recent_packets.pop(0)
self.rx_count += 1
def cleanup_cache(self):
@@ -910,6 +935,7 @@ class RepeaterHandler(BaseHandler):
for attempt in range(2 if local_transmission else 1):
try:
await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False)
self._record_packet_sent(fwd_pkt)
if airtime_ms > 0:
self.airtime_mgr.record_tx(airtime_ms)
packet_size = fwd_pkt.get_raw_length()
@@ -931,6 +957,14 @@ class RepeaterHandler(BaseHandler):
return asyncio.create_task(delayed_send())
def _record_packet_sent(self, packet: Packet) -> None:
"""Record a packet send for flood/direct stats (forwarded and originated)."""
route = getattr(packet, "header", 0) & PH_ROUTE_MASK
if route in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
self.sent_flood_count += 1
elif route in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
self.sent_direct_count += 1
def get_noise_floor(self) -> Optional[float]:
try:
radio = self.dispatcher.radio if self.dispatcher else None
@@ -976,6 +1010,12 @@ class RepeaterHandler(BaseHandler):
"rx_count": self.rx_count,
"forwarded_count": self.forwarded_count,
"dropped_count": self.dropped_count,
"recv_flood_count": self.recv_flood_count,
"recv_direct_count": self.recv_direct_count,
"sent_flood_count": self.sent_flood_count,
"sent_direct_count": self.sent_direct_count,
"flood_dup_count": self.flood_dup_count,
"direct_dup_count": self.direct_dup_count,
"rx_per_hour": rx_per_hour,
"forwarded_per_hour": forwarded_per_hour,
"recent_packets": self.recent_packets,
repeater/handler_helpers/protocol_request.py
+71 -59
View File
@@ -129,70 +129,73 @@ class ProtocolRequestHelper:
return False
def _handle_get_status(self, client, timestamp: int, req_data: bytes):
"""Build 56-byte RepeaterStats (firmware layout from MeshCore simple_repeater/MyMesh.h)."""
# RepeaterStats: uint16 batt, uint16 curr_tx_queue_len, int16 noise_floor, int16 last_rssi,
# uint32 n_packets_recv, n_packets_sent, total_air_time_secs, total_up_time_secs,
# n_sent_flood, n_sent_direct, n_recv_flood, n_recv_direct,
# uint16 err_events, int16 last_snr (×4), uint16 n_direct_dups, n_flood_dups,
# uint32 total_rx_air_time_secs, n_recv_errors → 56 bytes
# C++ struct RepeaterStats (44 bytes total):
# uint16_t batt_milli_volts;
# uint16_t curr_tx_queue_len;
# int16_t noise_floor;
# int16_t last_rssi;
# uint32_t n_packets_recv;
# uint32_t n_packets_sent;
# uint32_t total_air_time_secs;
# uint32_t total_up_time_secs;
# uint32_t n_sent_flood;
# uint32_t n_sent_direct;
# uint32_t n_recv_flood;
# uint32_t n_recv_direct;
# uint32_t err_events;
# int16_t last_snr;
# uint32_t n_direct_dups;
# uint32_t n_flood_dups;
# uint32_t total_rx_air_time_secs;
# Uptime: use engine start_time when available (fixes wrong "20521 days" from time.time())
if self.engine and hasattr(self.engine, "start_time"):
total_up_time_secs = int(time.time() - self.engine.start_time)
else:
total_up_time_secs = 0
# Get stats from radio/engine
noise_floor = int(self.radio.get_noise_floor() * 1.0) if self.radio else -120
last_rssi = (
int(self.radio.last_rssi) if self.radio and hasattr(self.radio, "last_rssi") else -120
)
last_snr = int(
(self.radio.last_snr * 4.0) if self.radio and hasattr(self.radio, "last_snr") else 0
)
# Radio: noise floor, last RSSI, last SNR (firmware stores SNR × 4)
if self.radio:
noise_floor = int(getattr(self.radio, "get_noise_floor", lambda: 0)() or 0)
if callable(getattr(self.radio, "get_last_rssi", None)):
last_rssi = int(self.radio.get_last_rssi() or -120)
else:
last_rssi = int(getattr(self.radio, "last_rssi", -120) or -120)
if callable(getattr(self.radio, "get_last_snr", None)):
last_snr = int((self.radio.get_last_snr() or 0) * 4)
else:
last_snr = int((getattr(self.radio, "last_snr", 0) or 0) * 4)
else:
noise_floor = 0
last_rssi = -120
last_snr = 0
# Get packet counts
n_packets_recv = (
self.radio.packets_received
if self.radio and hasattr(self.radio, "packets_received")
else 0
)
n_packets_sent = (
self.radio.packets_sent if self.radio and hasattr(self.radio, "packets_sent") else 0
)
# Packet counts: prefer engine (rx_count, forwarded_count); fall back to radio if present
if self.engine:
n_packets_recv = getattr(self.engine, "rx_count", 0)
n_packets_sent = getattr(self.engine, "forwarded_count", 0)
elif self.radio:
n_packets_recv = getattr(self.radio, "packets_received", 0) or 0
n_packets_sent = getattr(self.radio, "packets_sent", 0) or 0
else:
n_packets_recv = 0
n_packets_sent = 0
# Get airtime stats
# Airtime (AirtimeManager uses total_airtime_ms for TX; total_rx_airtime_ms if we track RX)
total_air_time_secs = 0
total_rx_air_time_secs = 0
if self.engine and hasattr(self.engine, "airtime_manager"):
total_air_time_secs = int(self.engine.airtime_manager.total_tx_airtime_ms / 1000)
# Get routing stats
n_sent_flood = 0
n_sent_direct = 0
n_recv_flood = 0
n_recv_direct = 0
n_direct_dups = 0
n_flood_dups = 0
if self.engine:
n_sent_flood = getattr(self.engine, "sent_flood_count", 0)
n_sent_direct = getattr(self.engine, "sent_direct_count", 0)
n_recv_flood = getattr(self.engine, "recv_flood_count", 0)
n_recv_direct = getattr(self.engine, "recv_direct_count", 0)
n_direct_dups = getattr(self.engine, "direct_dup_count", 0)
n_flood_dups = getattr(self.engine, "flood_dup_count", 0)
am = getattr(self.engine, "airtime_mgr", None) or getattr(
self.engine, "airtime_manager", None
)
if am is not None:
total_air_time_secs = int(getattr(am, "total_airtime_ms", 0) or 0) // 1000
total_rx_air_time_secs = int(getattr(am, "total_rx_airtime_ms", 0) or 0) // 1000
# Pack struct (little-endian)
# Routing stats (flood/direct and dups - from engine when available)
n_sent_flood = getattr(self.engine, "sent_flood_count", 0) if self.engine else 0
n_sent_direct = getattr(self.engine, "sent_direct_count", 0) if self.engine else 0
n_recv_flood = getattr(self.engine, "recv_flood_count", 0) if self.engine else 0
n_recv_direct = getattr(self.engine, "recv_direct_count", 0) if self.engine else 0
n_direct_dups = getattr(self.engine, "direct_dup_count", 0) if self.engine else 0
n_flood_dups = getattr(self.engine, "flood_dup_count", 0) if self.engine else 0
n_recv_errors = (
int(getattr(self.radio, "crc_error_count", 0) or 0)
if self.radio
else 0
)
# Pack 56-byte RepeaterStats (layout matches firmware)
stats = struct.pack(
"<HHhhIIIIIIIIIhIII",
"<HHhhIIIIIIIIHhHHII",
0, # batt_milli_volts (not available on Pi)
0, # curr_tx_queue_len (TODO)
noise_floor,
@@ -200,7 +203,7 @@ class ProtocolRequestHelper:
n_packets_recv,
n_packets_sent,
total_air_time_secs,
int(time.time()), # total_up_time_secs
total_up_time_secs,
n_sent_flood,
n_sent_direct,
n_recv_flood,
@@ -210,8 +213,17 @@ class ProtocolRequestHelper:
n_direct_dups,
n_flood_dups,
total_rx_air_time_secs,
n_recv_errors,
)
logger.debug(f"GET_STATUS: noise={noise_floor}dBm, rssi={last_rssi}dBm, snr={last_snr/4}dB")
logger.debug(
"GET_STATUS: uptime=%ds, noise=%ddBm, rssi=%ddBm, snr=%.1fdB, rx=%s, tx=%s",
total_up_time_secs,
noise_floor,
last_rssi,
last_snr / 4.0,
n_packets_recv,
n_packets_sent,
)
return stats
repeater/packet_router.py
+6
View File
@@ -112,6 +112,9 @@ class PacketRouter:
packet, metadata, local_transmission=True
)
# Mark so when this packet is dequeued we don't pass to engine again (avoid double-send / double-count)
packet._injected_for_tx = True
# Enqueue so router can deliver to companion(s): TXT_MSG -> dest bridge, ACK -> all bridges (sender sees ACK)
await self.enqueue(packet)
@@ -360,6 +363,9 @@ class PacketRouter:
logger.debug(f"Companion bridge GRP_TXT error: {e}")
# Only pass to repeater engine if not already processed by injection
# Skip engine for packets we injected for TX (already sent; avoid double-send/double-count)
if getattr(packet, "_injected_for_tx", False):
processed_by_injection = True
if self.daemon.repeater_handler and not processed_by_injection:
metadata = {
"rssi": getattr(packet, "rssi", 0),