how to record customer voice in sip phone in python with vicidial running in azure VM

[jawad097]

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}")

[jawad097]

i want to record the coustomer voice in vicidial and want to save in my sip phone vm

[AntiChrist-Coder]

Wrong github mate, this is pyVOIP, not pjsua2