у меня pjsua2

ошибка
python3: ../src/pj/os_core_unix.c:857: pj_thread_this: Assertion `!"Calling pjlib from unknown/external thread. You must " "register external threads with pj_thread_register() " "before calling any pjlib functions."' failed.
Аварийный останов


КОД
from param import Constants
from loggers import logger

import os
import threading
import time
import pjsua2 as pj
import uuid
import logging
import numpy as np
import webrtcvad

#import wave
#import struct

import queue
from queue import Queue

class MyAudioMediaPort(pj.AudioMediaPort):
    def __init__(self, call, stt, tts, company):
        super().__init__()
        pj.AudioMediaPort.__init__(self)  # Вызываем базовый конструктор без параметров
        self.call_id = str(uuid.uuid4())
        self.stt = stt
        self.tts = tts
        self.call = call
        self.call_active = 0
        #
        self.audio_buffer = bytearray()
        self.vad = webrtcvad.Vad(2)
        self.is_vad = 0
        self.is_vad_start = 0
        #

        self.sample_rate = Constants.SIP_SAMPLERATE  # используемый sample rate
        self.frame_size = int(self.sample_rate * 0.02 * 2)  # размер буфера за 20 мс
        
        self.tts_audio = np.array([], dtype=np.int16)     # numpy array с PCM int16
        self.tts_pos = 0          # текущая позиция чтения TTS аудио
        self.pos = 0  # Текущая позиция в генерации тона  
        self.send = 0
        self.is_first = 0
        self.is_firstc = 0
        self.tts.param(self, self.call_id, company)
  
    def reset(self):
        self.call_active = 0

        self.audio_buffer = bytearray()
        self.is_vad = 0
        self.is_vad_start = 0
        self.tts_audio = np.array([], dtype=np.int16)
        self.tts_pos = 0
        self.pos = 0
        self.send = 0
        self.is_first = 0
        self.is_firstc = 0

    def hangup_call(self):
        if self.call is None:
            print("Call object is None, cannot hangup.")
            return

        try:
            call_info = self.call.getInfo()
            logger.warning(f"Call state changed: {call_info.stateText}")
            if call_info.state != pj.PJSIP_INV_STATE_DISCONNECTED:
                prm = pj.CallOpParam()
                prm.statusCode = 200
                self.call.hangup(prm)
                print("Hangup signal sent.")
            else:
                print("Call already disconnected.")
        except pj.Error as e:
            print(f"Error hanging up call: {e}")
        finally:
            self.reset()

    def onFrameRequested(self, frame):

      if self.send != 1:
          frame.buf[:] = bytes(frame.size)
          return
          
      frame.type = pj.PJMEDIA_FRAME_TYPE_AUDIO
    
      # Защита от выхода за границы буфера
      available = max(0, len(self.tts_audio) - self.tts_pos)
      
      if available <= 0:
          # Можно от человека принимать 
          self.send = 2
          frame.buf[:] = bytes(frame.size)
          self.tts_pos = 0
          self.tts.allow(self.call_id)
          return
          
      samples_needed = self.frame_size // 2  # 160 samples * 2 bytes = 320 bytes
      
      # Вычисляем сколько можно взять
      take_samples = min(samples_needed, available)
      
      # Берем данные из буфера
      chunk = self.tts_audio[self.tts_pos : self.tts_pos + take_samples]
      self.tts_pos += take_samples
      
      # Если не хватает - добиваем тишиной
      if take_samples < samples_needed:
          silence = np.zeros(samples_needed - take_samples, dtype=np.int16)
          chunk = np.concatenate([chunk, silence])
          
      # Проверка данных перед отправкой
      #if np.max(np.abs(chunk)) > 32767:
      #    logger.error("Обнаружены клиппированные samples!")
      #    chunk = np.clip(chunk, -32767, 32767)
      
      frame.buf[:] = chunk.tobytes()
      
      if self.is_first == 0:
        self.is_first = 1
        logger.warning(f" Человеку - разрешено говорить")
    
    # Запуск разговора
    def runvoice(self):
        self.call_active = 1
        logger.warning(f"SETUP VOICE {id(self)} self.call_active {self.call_active}")
      
    def onFrameReceived(self, frame):

        #logger.warning(f"onFrameReceived {id(self)} self.call_active {self.call_active}")

        if self.send == -1:
          if self.call_active == 1:
            self.hangup_call()

        if self.call_active == 0:
            return
            
        if frame.size != self.frame_size:
            return

        is_speech = self.vad.is_speech(bytes(frame.buf), self.sample_rate)

        if is_speech:
            self.audio_buffer.extend(frame.buf)
            self.is_vad = 0   # сброс счётчика тишины
        else:
            self.is_vad += 1
            
        self.is_vad_start +=1

        # В байтах (sampleRate * seconds * 2 байта на сэмпл)
        min_audio_len1 = int(self.sample_rate * Constants.STT_COUNT_SAMPLE_SEC1 * 2)
        min_audio_len2 = int(self.sample_rate * Constants.STT_COUNT_SAMPLE_SEC2 * 2)
      
        #logger.warning(f"Audio : {len(self.audio_buffer)} >= {min_audio_len1} vad {self.is_vad} vads {self.is_vad_start} ")

        #logger.info(f"Audio buffer length: {len(self.audio_buffer)} >= {min_audio_len1} {min_audio_len2}")
        # например 2 секунды аудио (16kHz*2 байта*2сек)
        #if len(self.audio_buffer) >= self.sample_rate*2 and self.is_empty >= 3:

        # Запускаем обработку, если накоплен минимум аудио и после этого стоит достаточная пауза (тишина)
        #if (len(self.audio_buffer) >= min_audio_len1 and self.is_vad >= Constants.STT_COUNT_SAMPLE_NOICE1) or \
        #   (len(self.audio_buffer) >= min_audio_len2 and self.is_vad >= Constants.STT_COUNT_SAMPLE_NOICE2):
        if self.is_vad >=2 and (len(self.audio_buffer) >= 6080) or (self.is_vad > 30 and len(self.audio_buffer) >= 2880):
            if self.is_first == 1 and self.send == 2:
              logger.warning(f"  ЧЕЛОВЕК - завершил говорить")
              # Добавление тишины
              tail_samples = int(self.sample_rate * 0.6)
              silence_tail = np.zeros(tail_samples, dtype=np.int16)

              # Добавляем к bytearray байтовое представление
              self.audio_buffer.extend(silence_tail.tobytes())

              self.stt.start(self.call_id, self.audio_buffer)

              self.audio_buffer = bytearray()
              self.is_vad_start = 0
            
        # Долгую тишину сбрасываем звонок
        if self.is_vad_start > 700:
          self.send = -1

        if self.is_first == 0 and self.is_vad_start >= 6:
            if self.call_active == 1:
              logger.warning(f" РОБОТ = ПРИВЕТСТВИЕ {self.call_id}")

              self.is_first = 1
              self.tts.allow(self.call_id)


class Call(pj.Call):
  def __init__(self, acc, stt, tts, company,call_id=pj.PJSUA_INVALID_ID):
      super().__init__(acc, call_id)

      self.stt = stt
      self.tts = tts
      self.company = company
      self.prm = None
      self.acc = acc
      self.aud_med = None
      #
      self.call_status = None
      self.call_code = None
      self.duration = 0
      self.call_record = ""
      # Обработка входящего
      self.med_port = None 
   
      self.completed = threading.Event()

  def onCallState(self, prm):
      ci = self.getInfo()
      logger.info(f"Call state changed: {ci.stateText}")
      if ci.state == pj.PJSIP_INV_STATE_CONFIRMED:
          logger.info("Call confirmed")
          self.med_port.runvoice()
          
      elif ci.state == pj.PJSIP_INV_STATE_DISCONNECTED:
          logger.warning("Call disconnected")

          if not self.call_status:
            if ci.lastStatusCode == 200:
                self.call_status = "answered"
                self.duration = ci.connectDuration.sec if ci.connectDuration.sec > 0 else 0
            elif ci.lastStatusCode == 486:
                self.call_status = "busy"
            elif ci.lastStatusCode in (480, 408):
                self.call_status = "no_answer"
            elif ci.lastStatusCode == 487:
                self.call_status = "cancelled"
            elif ci.lastStatusCode == 603:
                self.call_status = "declined"
            else:
                self.call_status = "failed"
               
          self.call_code = ci.lastStatusCode
          self.call_record = self.stt.get_text()

          if self.med_port:
              try:
                  self.aud_med.stopTransmit(self.med_port)
                  self.med_port.stopTransmit(self.aud_med)

              except Exception as e:
                  logger.warning(f"Error stopping media port: {e}")
     
          self.med_port = None
          self.aud_med = None
  
          self.completed.set()


  def onCallMediaState(self, prm):

      ci = self.getInfo()
      for media_info in ci.media:
          logger.info(f"STATUS {media_info.status}")
          if media_info.status == pj.PJSUA_CALL_MEDIA_ACTIVE and media_info.type == pj.PJMEDIA_TYPE_AUDIO:

            if not self.med_port:
              logger.debug(f"RUN MEDIA {media_info.status}")
              self.aud_med = self.getAudioMedia(-1)
              self.med_port = MyAudioMediaPort(self, self.stt, self.tts, self.company)
                 
              self.fmt = pj.MediaFormatAudio()

              self.fmt.type = pj.PJMEDIA_TYPE_AUDIO
              self.fmt.clockRate = Constants.SIP_SAMPLERATE
              self.fmt.channelCount = 1
              self.fmt.bitsPerSample = 16
              self.fmt.frameTimeUsec = 20000
             
              self.med_port.createPort("med_port", self.fmt)

              # обработка -> звонок
              self.med_port.startTransmit(self.aud_med) 
              # звонок -> обработка
              self.aud_med.startTransmit(self.med_port)
          
            
  def wait_for_completion(self, timeout=None):
      self.completed.wait(timeout)
      return {
          'code': self.call_code,
          'status': self.call_status,
          'duration': self.duration,
          'record': self.call_record
      }

class Account(pj.Account):
    def __init__(self, stt, tts):
      pj.Account.__init__(self)
      #self.calls = []
      self.is_reg = False
      self.stt = stt
      self.tts = tts
   
    def onRegState(self, prm):
      self.is_reg = prm.code == 200
      status = "Зарегистрирован" if self.is_reg else "Не зарегистрирован"
      logger.info(f"onRegState {status} (код: {prm.code})")

    def onIncomingCall(self, iprm):
      logger.info(f"onIncomingCall")
      

      call = Call(self, stt=self.stt, tts=self.tts, company="", call_id=iprm.callId)
      call_prm = pj.CallOpParam()
      ci = call.getInfo()
      
      #self.calls.append(call)
      time.sleep(0.5)  
      call_prm.statusCode = 200
      call.answer(call_prm)
      
    """   
    def removeCall(self, call):
        for c in self.calls[:]:
          
          if c == call:
            self.calls.remove(c)
            logger.warning(f"Удален вызов {call.getInfo().remoteUri}")
            break
    """
    
class SIPWorker:
    def __init__(self, sip_server, sip_user, sip_pass, stt, tts):
      self.sip_server = sip_server
      self.sip_user = sip_user
      self.sip_pass = sip_pass
      #self.acc_cfg = acc_cfg
      self.command_queue = queue.Queue()
      self.result_queue = queue.Queue()
      self.thread = None
      self.running = False
      self.company = None
      self.stt = stt
      self.tts = tts
      self.call = None
      self.last_call_time = 0
      
    def start(self):
      if self.running:
          return
          
      self.running = True
      self.thread = threading.Thread(target=self._run, daemon=True)
      self.thread.start()
      
      # Ждем инициализации
      while not hasattr(self, 'ep'):
          time.sleep(0.1)

    def _run(self):

      try:
        # Всё в одном потоке - регистрация не нужна
        self.ep = pj.Endpoint()
        self.ep.libCreate()
        
        # Конфигурация endpoint
        ep_cfg = pj.EpConfig()
        ep_cfg.medConfig.maxCalls = 1
        ep_cfg.medConfig.threadCnt = 0

        ep_cfg.medConfig.threadCnt = 0  # No worker threads for audio
        #ep_cfg.medConfig.noVad = True   # Disable VAD (Voice Activity Detection) if not needed
        ep_cfg.medConfig.noAudioDev = True

        ep_cfg.logConfig.level = 0
        ep_cfg.logConfig.consoleLevel = 0
         
        self.ep.libInit(ep_cfg)
        
        tcfg = pj.TransportConfig()
        tcfg.port = 5060
        tcfg.qosType = pj.PJ_QOS_TYPE_VOICE
        tcfg.portRange = 0
        self.ep.transportCreate(pj.PJSIP_TRANSPORT_UDP, tcfg)
   
        acc_cfg = pj.AccountConfig()
        acc_cfg.idUri = f"sip:{self.sip_user}@{self.sip_server}"
        acc_cfg.regConfig.registrarUri = f"sip:{self.sip_server};expires=3600"
        
        cred = pj.AuthCredInfo("digest", "*", self.sip_user, 0, self.sip_pass)
        acc_cfg.sipConfig.authCreds.append(cred)

        # Устанавливаем интервал регистрации (в секундах)
        acc_cfg.regConfig.retryIntervalSec = 60
        acc_cfg.regConfig.firstRetryIntervalSec = 60
        acc_cfg.regConfig.delayBeforeRefreshSec = 120

        # Если сервер требует определенный Expires
        acc_cfg.regConfig.registrarTimeoutSec = 3600
    
        self.acc = Account(self.stt, self.tts)
        self.acc.create(acc_cfg)
        
        self.ep.libStart()

        self.ep.instance().audDevManager().setNullDev()

        # поддержка регистрации
        CHECK_INTERVAL = 60  # раз в минуту проверяем
        REREGISTER_TIMEOUT = 5*60  # 5 минут
        last_check = 0

        # Основной цикл обработки
        while self.running:
          # Обработка команд
          
          now = time.time()
          if now - last_check > CHECK_INTERVAL:
              last_check = now

              if self.acc.is_reg:
                  idle = now - self.last_call_time
                  if idle > REREGISTER_TIMEOUT:
                      logger.warning("No calls for 5 minutes, forcing re-registration")
                      try:
                          # Отключаем регистрацию
                          self.acc.setRegistration(False)
                          time.sleep(2)  # небольшая пауза
                          # Включаем регистрацию заново
                          self.acc.setRegistration(True)
                      except Exception as e:
                          logger.error(f"Failed to re-register: {e}")


          try:
              cmd = self.command_queue.get_nowait()
              if cmd[0] == 'call':
                  self._make_call(cmd[1])
              elif cmd[0] == 'stop':
                  break
          except queue.Empty:
              pass
          # Обработка событий
          self.ep.libHandleEvents(10)
          time.sleep(0.1)
              
      except Exception as e:
          logger.error(f"SIP error: {e}")
      finally:
          self._cleanup()
          
    def reg_company(self, company):
      self.company = company
      
    def _make_call(self, phone):
      
        result = {'status': 'error', 'message': 'noreg'}
        
        if self.acc.is_reg == True:
          
          try:
     
              call = Call(acc=self.acc, stt=self.stt, tts=self.tts, company=self.company)
              
              call_prm = pj.CallOpParam(True)
              call_prm.opt.audioCount = 1
              call_prm.opt.videoCount = 0
              call_prm.opt.timeoutSec = 10
              
     
              call.makeCall(f"sip:{phone}@{self.sip_server}", call_prm)
              
              self.last_call_time = time.time()
      
              result = call.wait_for_completion()
           
              logger.warning(f"self.call END")
          
             
              del call

          except Exception as e:
              result = {'status': 'error', 'message': str(e)}
              
              
        return self.result_queue.put(result)
         
    def make_call(self, phone):
      self.command_queue.put(('call', phone))
      return self.result_queue.get()

    def make_calloff(self, phone):
      self.command_queue.put(('call', phone))
      # Отключить звонок
      return self.result_queue.get()

    def _cleanup(self):
      if hasattr(self, 'ep') and self.ep:
        self.ep.libDestroy()

    def stop(self):
      self.command_queue.put(('stop', None))
      self.running = False
      if self.thread:
        self.thread.join()

class k_SIP:
    def __init__(self, sip_server, sip_user, sip_pass, stt, tts):
        self.worker = SIPWorker(sip_server, sip_user, sip_pass, stt, tts)
        
    def start(self):
        self.worker.start()
        
    def reg_company(self, company):
        return self.worker.reg_company(company)
        
    def make_call(self, phone):
        return self.worker.make_call(phone)
        
    def stop(self):
        self.worker.stop()