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