tayler6000
how to record customer voice in sip phone in python with vicidial running in azure VM
import pjsua2 as pj import time import threading import sys import os # For handling file paths
Logging callback
def log_cb(level, str, len): print(str.strip())
Subclass to extend the Account class
class MyAccount(pj.Account): def init(self, sip_phone): pj.Account.init(self) self.sip_phone = sip_phone
def onRegState(self, prm): print("Registration state changed:") print(f" Status code: {prm.code}") print(f" Status text: {prm.reason}") print(f" Expiration: {prm.expiration}") if prm.code != 200: print(f" Registration failed. Please check your credentials and connection.")
def onIncomingCall(self, prm): call = MyCall(self, prm.callId) call_info = call.getInfo() print(f"Incoming call from: {call_info.remoteUri}") self.sip_phone.current_call = call
call_prm = pj.CallOpParam()
call_prm.statusCode = 200 # Answer immediately with "OK"
call.answer(call_prm)
print("Call answered immediately.")
Subclass to extend the Call class
class MyCall(pj.Call): def init(self, acc, call_id=pj.PJSUA_INVALID_ID): pj.Call.init(self, acc, call_id) self.connected = False self.keep_alive_timer = None self.recorder = None # AudioMediaRecorder for customer voice
def onCallState(self, prm): ci = self.getInfo() print(f"Call state changed to {ci.stateText}") print(f"Call duration: {ci.totalDuration.sec} seconds") if ci.state == pj.PJSIP_INV_STATE_DISCONNECTED: print(f"Call disconnected. Last status: {ci.lastStatusCode}") self.connected = False if self.keep_alive_timer: self.keep_alive_timer.cancel()
# Stop recording when the call is disconnected
if self.recorder:
print("Stopping the recording.")
try:
if self.getMedia(0): # Ensure media still exists before stopping transmission
am = pj.AudioMedia.typecastFromMedia(self.getMedia(0)) # Assume media index 0
self.recorder.stopTransmit(am) # Pass the remote media (am) to stopTransmit
self.recorder = None
except pj.Error as e:
print(f"Error stopping the recording: {e}")
except ValueError as e:
print(f"Invalid media reference: {e}")
elif ci.state == pj.PJSIP_INV_STATE_CONFIRMED:
print("Call is now connected.")
self.connected = True
self.start_keep_alive()
def onCallMediaState(self, prm): ci = self.getInfo() print("Call media state changed.") for mi in ci.media: if mi.type == pj.PJMEDIA_TYPE_AUDIO and mi.status == pj.PJSUA_CALL_MEDIA_ACTIVE: m = self.getMedia(mi.index) am = pj.AudioMedia.typecastFromMedia(m)
# Check if this is the remote media (customer's voice)
if mi.dir == pj.PJMEDIA_DIR_DECODING or mi.dir == pj.PJMEDIA_DIR_ENCODING_DECODING:
print("Remote media detected, recording customer's voice.")
# Record only the customer's voice (remote media)
if not self.recorder: # Avoid multiple recorders for the same call
self.recorder = pj.AudioMediaRecorder()
filename = f"customer_voice_{ci.id}_{time.strftime('%Y%m%d-%H%M%S')}.wav"
try:
# Ensure the directory exists
if not os.path.exists('recordings'):
os.makedirs('recordings')
filepath = os.path.join('recordings', filename)
self.recorder.createRecorder(filepath)
am.startTransmit(self.recorder) # Start transmitting only remote media to recorder
print(f"Recording customer's voice to file: {filepath}")
except pj.Error as e:
print(f"Error setting up recorder: {e}")
else:
print("Skipping local media.")
def start_keep_alive(self): def send_keep_alive(): pj.Endpoint.instance().libRegisterThread('keep_alive_thread') if self.connected: try: # Send a re-INVITE self.reinvite(pj.CallOpParam()) print("Sent keep-alive (re-INVITE) to Asterisk") except pj.Error as e: print(f"Error sending keep-alive: {e}") # Schedule the next re-INVITE after 20 seconds self.keep_alive_timer = threading.Timer(20, send_keep_alive) self.keep_alive_timer.start()
# Schedule the first re-INVITE after 20 seconds
print("Starting keep-alive process (sending re-INVITE every 20 seconds)...")
self.keep_alive_timer = threading.Timer(20, send_keep_alive)
self.keep_alive_timer.start()
class SipPhone: def init(self): self.ep = None self.acc = None self.current_call = None
def init_lib(self): self.ep = pj.Endpoint() self.ep.libCreate()
ep_cfg = pj.EpConfig()
ep_cfg.uaConfig.userAgent = "PJSUA2 Python SIP Phone"
ep_cfg.logConfig.level = 3 # Increased log level for debugging
ep_cfg.logConfig.consoleLevel = 3
ep_cfg.medConfig.noVad = False # Enable VAD (Voice Activity Detection)
# Custom logic to disable audio device (playback/capture)
ep_cfg.medConfig.hasIoqueue = False # Ensure we are not using sound I/O
ep_cfg.medConfig.clockRate = 16000 # Optional: Set the desired clock rate for RTP stream
ep_cfg.medConfig.audioFramePtime = 20 # Set RTP packetization period
self.ep.libInit(ep_cfg)
# Transport configuration
tcfg = pj.TransportConfig()
tcfg.port = 5060
try:
self.ep.transportCreate(pj.PJSIP_TRANSPORT_UDP, tcfg)
except pj.Error as e:
print(f"Error creating transport: {e}")
self.ep.libDestroy()
sys.exit(1)
self.ep.libStart()
def create_account(self, username, password, domain): acc_cfg = pj.AccountConfig() acc_cfg.idUri = f"sip:{username}@{domain}" acc_cfg.regConfig.registrarUri = f"sip:{domain}" cred = pj.AuthCredInfo("digest", "*", username, 0, password) acc_cfg.sipConfig.authCreds.append(cred)
self.acc = MyAccount(self)
try:
self.acc.create(acc_cfg)
print(f"Account created: {acc_cfg.idUri}")
except pj.Error as e:
print(f"Error creating account: {e}")
def hang_up(self): if self.current_call: try: self.current_call.hangup(pj.CallOpParam()) except pj.Error as e: print(f"Error hanging up call: {e}") self.current_call = None
def run(self): print("SIP Phone is running. Waiting for incoming calls...") try: while True: self.ep.libHandleEvents(10) except KeyboardInterrupt: print("Exiting...") finally: self.hang_up() if self.acc: try: self.acc.shutdown() # Use shutdown instead of delete except pj.Error as e: print(f"Error shutting down account: {e}")
i want to record the coustomer voice in vicidial and want to save in my sip phone vm
