import RNS import os import sys import time from LXST._version import __version__ from LXST.Primitives.Telephony import Telephone from RNS.vendor.configobj import ConfigObj import RNS.vendor.umsgpack as msgpack class ReticulumTelephone(): STATE_AVAILABLE = 0x00 STATE_CONNECTING = 0x01 STATE_RINGING = 0x02 STATE_IN_CALL = 0x03 HW_SLEEP_TIMEOUT = 15 HW_STATE_IDLE = 0x00 HW_STATE_DIAL = 0x01 HW_STATE_SLEEP = 0xFF RING_TIME = 30 WAIT_TIME = 60 PATH_TIME = 10 CALL_LOG_KEEP = 14*24*60*60 def __init__(self, identity, owner = None, service = False, speaker=None, microphone=None, ringer=None, logpath=None): self.identity = identity self.service = service self.owner = owner self.config = None self.should_run = False self.telephone = None self.state = self.STATE_AVAILABLE self.hw_state = self.HW_STATE_IDLE self.hw_last_event = time.time() self.hw_input = "" self.direction = None self.last_input = None self.first_run = False self.user_hung_up = False self.ringtone_path = None self.speaker_device = speaker self.microphone_device = microphone self.ringer_device = ringer self.logpath = logpath self.call_log = None self.phonebook = {} self.aliases = {} self.names = {} self.telephone = Telephone(self.identity, ring_time=self.RING_TIME, wait_time=self.WAIT_TIME) self.telephone.set_ringing_callback(self.ringing) self.telephone.set_established_callback(self.call_established) self.telephone.set_ended_callback(self.call_ended) self.telephone.set_speaker(self.speaker_device) self.telephone.set_microphone(self.microphone_device) self.telephone.set_ringer(self.ringer_device) self.telephone.set_allowed(self.__is_allowed) RNS.log(f"{self} initialised", RNS.LOG_DEBUG) def set_ringtone(self, ringtone_path): if os.path.isfile(ringtone_path): self.ringtone_path = ringtone_path self.telephone.set_ringtone(self.ringtone_path) def set_speaker(self, device): self.speaker_device = device self.telephone.set_speaker(self.speaker_device) def set_microphone(self, device): self.microphone_device = device self.telephone.set_microphone(self.microphone_device) def set_ringer(self, device): self.ringer_device = device self.telephone.set_ringer(self.ringer_device) def announce(self, attached_interface=None): self.telephone.announce(attached_interface=attached_interface) @property def is_available(self): return self.state == self.STATE_AVAILABLE @property def is_in_call(self): return self.state == self.STATE_IN_CALL @property def is_ringing(self): return self.state == self.STATE_RINGING @property def call_is_connecting(self): return self.state == self.STATE_CONNECTING @property def hw_is_idle(self): return self.hw_state == self.HW_STATE_IDLE @property def hw_is_dialing(self): return self.hw_state == self.HW_STATE_DIAL @property def active_profile(self): return self.telephone.active_profile def start(self): if not self.should_run: self.should_run = True self.run() def stop(self): self.should_run = False self.telephone.teardown() self.telephone = None def hangup(self): self.user_hung_up = True self.telephone.hangup() def answer(self): self.telephone.answer(self.caller) def set_busy(self, busy): self.telephone.set_busy(busy) def set_low_latency_output(self, enabled): self.telephone.set_low_latency_output(enabled) def clear_call_log(self): self.call_log = [] try: with open(self.logpath, "wb") as logfile: logfile.write(msgpack.packb([])) except Exception as e: raise OSError("Could not clear call log file") def get_call_log(self): if self.call_log: return self.call_log else: call_log = [] try: if os.path.isfile(self.logpath): with open(self.logpath, "rb") as logfile: read_call_log = msgpack.unpackb(logfile.read()) for entry in read_call_log: age = time.time()-entry["time"] if age < self.CALL_LOG_KEEP: call_log.append(entry) except Exception as e: RNS.log(f"Could not read call log file: {e}", RNS.LOG_ERROR) self.call_log = call_log return self.call_log def log_call(self, event, identity): if identity: RNS.log(f"Logging call event {event} for {RNS.prettyhexrep(identity.hash)}", RNS.LOG_DEBUG) if self.logpath: try: if not os.path.isfile(self.logpath): try: with open(self.logpath, "wb") as logfile: logfile.write(msgpack.packb([])) except Exception as e: raise OSError("Could not create call log file") call_log = [] read_call_log = [] try: with open(self.logpath, "rb") as logfile: read_call_log = msgpack.unpackb(logfile.read()) except Exception as e: RNS.log(f"Error while reading call log file: {e}", RNS.LOG_ERROR) RNS.log(f"Call log file will be re-created", RNS.LOG_ERROR) for entry in read_call_log: age = time.time()-entry["time"] if age < self.CALL_LOG_KEEP: call_log.append(entry) entry = {"time": time.time(), "event": event, "identity": identity.hash} call_log.append(entry) with open(self.logpath, "wb") as logfile: logfile.write(msgpack.packb(call_log)) self.call_log = call_log except Exception as e: RNS.log(f"An error occurred while updating call log: {e}", RNS.LOG_ERROR) RNS.trace_exception(e) def dial(self, identity_hash, profile=None): self.last_dialled_identity_hash = identity_hash destination_hash = RNS.Destination.hash_from_name_and_identity("lxst.telephony", identity_hash) if RNS.Transport.has_path(destination_hash): call_hops = RNS.Transport.hops_to(destination_hash) cs = "" if call_hops == 1 else "s" RNS.log(f"Connecting call over {call_hops} hop{cs}...", RNS.LOG_DEBUG) identity = RNS.Identity.recall(destination_hash) self.call(identity, profile=profile) else: return "no_path" def redial(self, args=None): if self.last_dialled_identity_hash: self.dial(self.last_dialled_identity_hash) def call(self, remote_identity, profile=None): RNS.log(f"Calling {RNS.prettyhexrep(remote_identity.hash)}...", RNS.LOG_DEBUG) self.state = self.STATE_CONNECTING self.caller = remote_identity self.direction = "to" self.telephone.call(self.caller, profile=profile) def switch_profile(self, profile): self.telephone.switch_profile(profile) def ringing(self, remote_identity): if self.hw_state == self.HW_STATE_SLEEP: self.hw_state = self.HW_STATE_IDLE self.state = self.STATE_RINGING self.caller = remote_identity self.direction = "from" if self.direction == None else "to" RNS.log(f"Incoming call from {RNS.prettyhexrep(self.caller.hash)}", RNS.LOG_DEBUG) if self.owner: self.owner.incoming_call(remote_identity) def call_ended(self, remote_identity): call_was_connecting = self.call_is_connecting was_ringing = self.is_ringing was_in_call = self.is_in_call user_hung_up = self.user_hung_up self.user_hung_up = False if self.is_in_call or self.is_ringing or self.call_is_connecting: if self.is_in_call: RNS.log(f"Call with {RNS.prettyhexrep(self.caller.hash)} ended\n", RNS.LOG_DEBUG) if self.is_ringing: RNS.log(f"Call {self.direction} {RNS.prettyhexrep(self.caller.hash)} was not answered\n", RNS.LOG_DEBUG) if self.call_is_connecting: RNS.log(f"Call to {RNS.prettyhexrep(self.caller.hash)} could not be connected\n", RNS.LOG_DEBUG) self.direction = None self.state = self.STATE_AVAILABLE if call_was_connecting: self.log_call("outgoing-failure", remote_identity) if not user_hung_up: self.owner.setstate("voice.connection_failure", True) elif was_in_call: self.log_call("ongoing-ended", remote_identity) self.owner.ended_call(remote_identity) elif was_ringing: self.log_call("incoming-missed", remote_identity) self.owner.missed_call(remote_identity) def call_established(self, remote_identity): call_was_connecting = self.call_is_connecting was_ringing = self.is_ringing if self.call_is_connecting or self.is_ringing: self.state = self.STATE_IN_CALL RNS.log(f"Call established with {RNS.prettyhexrep(self.caller.hash)}", RNS.LOG_DEBUG) if call_was_connecting: self.log_call("outgoing-success", remote_identity) elif was_ringing: self.log_call("incoming-success", remote_identity) def __is_allowed(self, identity_hash): if self.owner.config["voice_trusted_only"]: return self.owner.voice_is_trusted(identity_hash) else: return True def __spin(self, until=None, msg=None, timeout=None): if msg: RNS.log(msg, RNS.LOG_DEBUG) if timeout != None: timeout = time.time()+timeout while (timeout == None or time.time() timeout: return False else: return True class CallerProxy(): def __init__(self, hash=None): self.hash = hash class ReticulumTelephoneProxy(): PATH_TIME = ReticulumTelephone.PATH_TIME def __init__(self, owner=None): self.owner = owner @property def is_available(self): return self.owner.service_rpc_request({"telephone_is_available": True }) @property def is_in_call(self): return self.owner.service_rpc_request({"telephone_is_in_call": True }) @property def call_is_connecting(self): return self.owner.service_rpc_request({"telephone_call_is_connecting": True }) @property def is_ringing(self): return self.owner.service_rpc_request({"telephone_is_ringing": True }) @property def caller(self): return CallerProxy(hash=self.owner.service_rpc_request({"telephone_caller_info": True })) @property def active_profile(self): return self.owner.service_rpc_request({"telephone_active_profile": True }) def set_busy(self, busy): return self.owner.service_rpc_request({"telephone_set_busy": busy }) def dial(self, dial_target, profile=None): return self.owner.service_rpc_request({"telephone_dial": dial_target, "profile": profile }) def hangup(self): return self.owner.service_rpc_request({"telephone_hangup": True }) def answer(self): return self.owner.service_rpc_request({"telephone_answer": True }) def set_speaker(self, speaker): return self.owner.service_rpc_request({"telephone_set_speaker": speaker }) def set_microphone(self, microphone): return self.owner.service_rpc_request({"telephone_set_microphone": microphone }) def set_ringer(self, ringer): return self.owner.service_rpc_request({"telephone_set_ringer": ringer }) def set_low_latency_output(self, enabled): return self.owner.service_rpc_request({"telephone_set_low_latency_output": enabled}) def announce(self): return self.owner.service_rpc_request({"telephone_announce": True}) def get_call_log(self): return self.owner.service_rpc_request({"telephone_get_call_log": True}) def clear_call_log(self): return self.owner.service_rpc_request({"telephone_clear_call_log": True}) def switch_profile(self, profile): return self.owner.service_rpc_request({"telephone_switch_profile": profile})