diff --git a/docs/utilities/rns_audio_call_calc.py b/docs/utilities/rns_audio_call_calc.py deleted file mode 100644 index 6bb2ccb..0000000 --- a/docs/utilities/rns_audio_call_calc.py +++ /dev/null @@ -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") diff --git a/sbapp/main.py b/sbapp/main.py index 8efdd4d..7bc46d4 100644 --- a/sbapp/main.py +++ b/sbapp/main.py @@ -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 diff --git a/sbapp/sideband/core.py b/sbapp/sideband/core.py index 65ab389..36b8162 100644 --- a/sbapp/sideband/core.py +++ b/sbapp/sideband/core.py @@ -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"]