mirror of
https://github.com/markqvist/Sideband.git
synced 2025-02-22 07:30:01 -05:00
Added periodic telemetry data cleaning
This commit is contained in:
Mark Qvist
parent
3b2e1adaf2
commit
b4a063a4e7
@ -1,129 +0,0 @@
|
||||
import os
|
||||
import math
|
||||
import RNS
|
||||
import RNS.vendor.umsgpack as mp
|
||||
|
||||
def simulate(link_speed=9600, audio_slot_ms=70, codec_rate=1200, method="msgpack"):
|
||||
# Simulated on-air link speed
|
||||
LINK_SPEED = link_speed
|
||||
|
||||
# Packing method, can be "msgpack" or "protobuf"
|
||||
PACKING_METHOD = method
|
||||
|
||||
# The target audio slot time
|
||||
TARGET_MS = audio_slot_ms
|
||||
|
||||
# Packets needed per second for half-duplex audio
|
||||
PACKETS_PER_SECOND = 1000/TARGET_MS
|
||||
|
||||
# Effective audio encoder bitrate
|
||||
CODEC_RATE = codec_rate
|
||||
|
||||
# Maximum number of supported audio modes
|
||||
MAX_ENUM = 127
|
||||
|
||||
# Per-packet overhead on a established link is 19
|
||||
# bytes, 3 for header and context, 16 for link ID
|
||||
RNS_OVERHEAD = 19
|
||||
|
||||
# Physical-layer overhead. For RNode, this is 1
|
||||
# byte per RNS packet.
|
||||
PHY_OVERHEAD = 1
|
||||
|
||||
# Total transport overhead
|
||||
TRANSPORT_OVERHEAD = PHY_OVERHEAD+RNS_OVERHEAD
|
||||
|
||||
# Calculate parameters
|
||||
AUDIO_LEN = int(math.ceil(CODEC_RATE/(1000/TARGET_MS)/8))
|
||||
PER_BYTE_LATENCY_MS = 1000/(LINK_SPEED/8)
|
||||
|
||||
# Pack the message with msgpack to get real-
|
||||
# world packed message size
|
||||
|
||||
if PACKING_METHOD == "msgpack":
|
||||
# Calculate msgpack overhead
|
||||
PL_LEN = len(mp.packb([MAX_ENUM, os.urandom(AUDIO_LEN)]))
|
||||
PACKING_OVERHEAD = PL_LEN-AUDIO_LEN
|
||||
elif PACKING_METHOD == "protobuf":
|
||||
# For protobuf, assume the 8 bytes of stated overhead
|
||||
PACKING_OVERHEAD = 8
|
||||
PL_LEN = AUDIO_LEN+PACKING_OVERHEAD
|
||||
else:
|
||||
print("Unsupported packing method")
|
||||
exit(1)
|
||||
|
||||
# Calculate required encrypted token blocks
|
||||
BLOCKSIZE = 16
|
||||
REQUIRED_BLOCKS = math.ceil((PL_LEN+1)/BLOCKSIZE)
|
||||
ENCRYPTED_PAYLOAD_LEN = REQUIRED_BLOCKS*BLOCKSIZE
|
||||
BLOCK_HEADROOM = (REQUIRED_BLOCKS*BLOCKSIZE) - PL_LEN - 1
|
||||
|
||||
# The complete on-air packet length
|
||||
PACKET_LEN = PHY_OVERHEAD+RNS_OVERHEAD+ENCRYPTED_PAYLOAD_LEN
|
||||
PACKET_LATENCY = round(PACKET_LEN*PER_BYTE_LATENCY_MS, 1)
|
||||
|
||||
# TODO: This should include any additional
|
||||
# airtime consumption such as preamble and TX-tail.
|
||||
PACKET_AIRTIME = PACKET_LEN*PER_BYTE_LATENCY_MS
|
||||
AIRTIME_PCT = (PACKET_AIRTIME/TARGET_MS) * 100
|
||||
|
||||
# Maximum amount of concurrent full-duplex
|
||||
# calls that can coexist on the same channel
|
||||
CONCURRENT_CALLS = math.floor(100/AIRTIME_PCT)
|
||||
|
||||
# Calculate latencies
|
||||
TRANSPORT_LATENCY = round((PHY_OVERHEAD+RNS_OVERHEAD)*PER_BYTE_LATENCY_MS, 1)
|
||||
|
||||
PAYLOAD_LATENCY = round(ENCRYPTED_PAYLOAD_LEN*PER_BYTE_LATENCY_MS, 1)
|
||||
RAW_DATA_LATENCY = round(AUDIO_LEN*PER_BYTE_LATENCY_MS, 1)
|
||||
PACKING_LATENCY = round(PACKING_OVERHEAD*PER_BYTE_LATENCY_MS, 1)
|
||||
|
||||
DATA_LATENCY = round(ENCRYPTED_PAYLOAD_LEN*PER_BYTE_LATENCY_MS, 1)
|
||||
ENCRYPTION_LATENCY = round((ENCRYPTED_PAYLOAD_LEN-PL_LEN)*PER_BYTE_LATENCY_MS, 1)
|
||||
if ENCRYPTED_PAYLOAD_LEN-PL_LEN == 1:
|
||||
E_OPT_STR = "(optimal)"
|
||||
else:
|
||||
E_OPT_STR = "(sub-optimal)"
|
||||
|
||||
TOTAL_LATENCY = round(TARGET_MS+PACKET_LATENCY, 1)
|
||||
|
||||
print( "\n===== Simulation Parameters ===\n")
|
||||
print(f" Packing method : {method}")
|
||||
print(f" Sampling delay : {TARGET_MS}ms")
|
||||
print(f" Codec bitrate : {CODEC_RATE} bps")
|
||||
print(f" Audio data : {AUDIO_LEN} bytes")
|
||||
print(f" Packing overhead : {PACKING_OVERHEAD} bytes")
|
||||
print(f" Payload length : {PL_LEN} bytes")
|
||||
print(f" AES blocks needed : {REQUIRED_BLOCKS}")
|
||||
print(f" Encrypted payload : {ENCRYPTED_PAYLOAD_LEN} bytes")
|
||||
print(f" Transport overhead : {TRANSPORT_OVERHEAD} bytes ({RNS_OVERHEAD} from RNS, {PHY_OVERHEAD} from PHY)")
|
||||
print(f" On-air length : {PACKET_LEN} bytes")
|
||||
print(f" Packet airtime : {round(PACKET_AIRTIME,2)}ms")
|
||||
|
||||
print( "\n===== Results for "+RNS.prettyspeed(LINK_SPEED)+" Link Speed ===\n")
|
||||
print(f" Final latency : {TOTAL_LATENCY}ms")
|
||||
print(f" Recording latency : contributes {TARGET_MS}ms")
|
||||
print(f" Packet transport : contributes {PACKET_LATENCY}ms")
|
||||
print(f" Payload : contributes {PAYLOAD_LATENCY}ms")
|
||||
print(f" Audio data : contributes {RAW_DATA_LATENCY}ms")
|
||||
print(f" Packing format : contributes {PACKING_LATENCY}ms")
|
||||
print(f" Encryption : contributes {ENCRYPTION_LATENCY}ms {E_OPT_STR}")
|
||||
print(f" RNS+PHY overhead : contributes {TRANSPORT_LATENCY}ms")
|
||||
print(f"")
|
||||
print(f" Half-duplex airtime : {round(AIRTIME_PCT, 2)}% of link capacity")
|
||||
print(f" Concurrent calls : {int(CONCURRENT_CALLS)}\n")
|
||||
print(f" Full-duplex airtime : {round(AIRTIME_PCT*2, 2)}% of link capacity")
|
||||
print(f" Concurrent calls : {int(CONCURRENT_CALLS/2)}")
|
||||
|
||||
if BLOCK_HEADROOM != 0:
|
||||
print("")
|
||||
print(f" !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
|
||||
print(f" Unaligned AES block! Each packet could fit")
|
||||
print(f" {BLOCK_HEADROOM} bytes of additional audio data")
|
||||
print(f" !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
|
||||
|
||||
print( "\n= With mspack =================")
|
||||
simulate(method="msgpack")
|
||||
|
||||
#print("\n\n= With protobuf ===============")
|
||||
#simulate(method="protobuf")
|
||||
@ -2809,7 +2809,7 @@ class SidebandApp(MDApp):
|
||||
|
||||
str_comps = " - [b]Reticulum[/b] (MIT License)\n - [b]LXMF[/b] (MIT License)\n - [b]KivyMD[/b] (MIT License)"
|
||||
str_comps += "\n - [b]Kivy[/b] (MIT License)\n - [b]Codec2[/b] (LGPL License)\n - [b]PyCodec2[/b] (BSD-3 License)"
|
||||
str_comps += "\n - [b]PyDub[/b] (MIT License)\n - [b]PyOgg[/b] (Public Domain)\n - [b]MD2bbcode[/b] (GPL3 License)"
|
||||
str_comps += "\n - [b]PyDub[/b] (MIT License)\n - [b]PyOgg[/b] (Public Domain)\n - [b]FFmpeg[/b] (GPL3 License)\n - [b]MD2bbcode[/b] (GPL3 License)"
|
||||
str_comps += "\n - [b]GeoidHeight[/b] (LGPL License)\n - [b]Paho MQTT[/b] (EPL2 License)\n - [b]Python[/b] (PSF License)"
|
||||
str_comps += "\n\nGo to [u][ref=link]https://unsigned.io/donate[/ref][/u] to support the project.\n\nThe Sideband app is Copyright © 2025 Mark Qvist / unsigned.io\n\nPermission is granted to freely share and distribute binary copies of "+self.root.ids.app_version_info.text+", so long as no payment or compensation is charged for said distribution or sharing.\n\nIf you were charged or paid anything for this copy of Sideband, please report it to [b]license@unsigned.io[/b].\n\nTHIS IS EXPERIMENTAL SOFTWARE - SIDEBAND COMES WITH ABSOLUTELY NO WARRANTY - USE AT YOUR OWN RISK AND RESPONSIBILITY"
|
||||
info = "This is "+self.root.ids.app_version_info.text+", on RNS v"+RNS.__version__+" and LXMF v"+LXMF.__version__+".\n\nHumbly build using the following open components:\n\n"+str_comps
|
||||
|
||||
@ -113,8 +113,10 @@ class SidebandCore():
|
||||
SERVICE_JOB_INTERVAL = 1
|
||||
PERIODIC_JOBS_INTERVAL = 60
|
||||
PERIODIC_SYNC_RETRY = 360
|
||||
TELEMETRY_KEEP = 60*60*24*7
|
||||
TELEMETRY_INTERVAL = 60
|
||||
SERVICE_TELEMETRY_INTERVAL = 300
|
||||
TELEMETRY_CLEAN_INTERVAL = 3600
|
||||
|
||||
IF_CHANGE_ANNOUNCE_MIN_INTERVAL = 3.5 # In seconds
|
||||
AUTO_ANNOUNCE_RANDOM_MIN = 90 # In minutes
|
||||
@ -174,6 +176,8 @@ class SidebandCore():
|
||||
self.pending_telemetry_send_try = 0
|
||||
self.pending_telemetry_send_maxtries = 2
|
||||
self.telemetry_send_blocked_until = 0
|
||||
self.telemetry_clean_interval = self.TELEMETRY_CLEAN_INTERVAL
|
||||
self.last_telemetry_clean = 0
|
||||
self.pending_telemetry_request = False
|
||||
self.telemetry_request_max_history = 7*24*60*60
|
||||
self.live_tracked_objects = {}
|
||||
@ -2717,7 +2721,7 @@ class SidebandCore():
|
||||
db.commit()
|
||||
|
||||
def _db_clean_messages(self):
|
||||
RNS.log("Purging stale messages... "+str(self.db_path))
|
||||
RNS.log("Purging stale messages... ", RNS.LOG_DEBUG)
|
||||
with self.db_lock:
|
||||
db = self.__db_connect()
|
||||
dbc = db.cursor()
|
||||
@ -2726,6 +2730,20 @@ class SidebandCore():
|
||||
dbc.execute(query, {"outbound_state": LXMF.LXMessage.OUTBOUND, "sending_state": LXMF.LXMessage.SENDING})
|
||||
db.commit()
|
||||
|
||||
def _db_clean_telemetry(self):
|
||||
RNS.log("Cleaning telemetry... ", RNS.LOG_DEBUG)
|
||||
clean_time = time.time()-self.TELEMETRY_KEEP
|
||||
with self.db_lock:
|
||||
db = self.__db_connect()
|
||||
dbc = db.cursor()
|
||||
|
||||
query = f"delete from telemetry where (ts < {clean_time});"
|
||||
dbc.execute(query, {"outbound_state": LXMF.LXMessage.OUTBOUND, "sending_state": LXMF.LXMessage.SENDING})
|
||||
db.commit()
|
||||
|
||||
self.last_telemetry_clean = time.time()
|
||||
|
||||
|
||||
def _db_message_set_state(self, lxm_hash, state, is_retry=False, ratchet_id=None, originator_stamp=None):
|
||||
msg_extras = None
|
||||
if ratchet_id != None:
|
||||
@ -3525,6 +3543,9 @@ class SidebandCore():
|
||||
self.setpersistent("lxmf.syncretrying", False)
|
||||
|
||||
if self.config["telemetry_enabled"]:
|
||||
if time.time()-self.last_telemetry_clean > self.telemetry_clean_interval:
|
||||
self._db_clean_telemetry()
|
||||
|
||||
if self.config["telemetry_send_to_collector"]:
|
||||
if self.config["telemetry_collector"] != None and self.config["telemetry_collector"] != self.lxmf_destination.hash:
|
||||
try:
|
||||
@ -4783,6 +4804,7 @@ class SidebandCore():
|
||||
|
||||
def start(self):
|
||||
self._db_clean_messages()
|
||||
self._db_clean_telemetry()
|
||||
self.__start_jobs_immediate()
|
||||
|
||||
thread = threading.Thread(target=self.__start_jobs_deferred)
|
||||
@ -5008,6 +5030,10 @@ class SidebandCore():
|
||||
RNS.log("Error while handling commands: "+str(e), RNS.LOG_ERROR)
|
||||
|
||||
def create_telemetry_collector_response(self, to_addr, timebase, is_authorized_telemetry_request=False):
|
||||
if self.getstate(f"telemetry.{RNS.hexrep(to_addr, delimit=False)}.update_sending") == True:
|
||||
RNS.log("Not sending new telemetry collector response, since an earlier transfer is already in progress", RNS.LOG_DEBUG)
|
||||
return "in_progress"
|
||||
|
||||
added_sources = {}
|
||||
sources = self.list_telemetry(after=timebase)
|
||||
only_latest = self.config["telemetry_requests_only_send_latest"]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user