cd /opt git clone https://github.com/gabrielmittag/NISQA.git cd NISQA python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install torch torchaudio librosa numpy scipy matplotlib pip install -e . 

cd /opt/NISQA python3 -c "from nisqa.NISQA_model import NISQAModel; m = NISQAModel(pretrained_model_dir='nisqa_pretrained_model'); print('NISQA ready')" 

pip install silero-vad python3 -c "import torch; torch.hub.load('snakers4/silero-vad', 'get_speech_timestamps', trust_repo=True); print('Silero VAD ready')" 

apt-get install -y ffmpeg ffmpeg -i call_20250115_120000_5551234567.wav -acodec pcm_s16le -ar 16000 decoded.wav 

CREATE TABLE vicidial_qa_scores ( qa_id INT AUTO_INCREMENT PRIMARY KEY, call_date DATETIME DEFAULT CURRENT_TIMESTAMP, vicidial_log_id BIGINT NOT NULL, filename VARCHAR(255) NOT NULL, nisqa_score FLOAT, nisqa_raw FLOAT, nisqa_dim_col FLOAT, nisqa_dim_noi FLOAT, nisqa_dim_mos FLOAT, speech_percentage FLOAT, silence_duration_sec FLOAT, noise_floor_db FLOAT, clipping_detected BOOLEAN DEFAULT 0, processing_status ENUM('pending', 'processed', 'error') DEFAULT 'pending', error_message TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY unique_file (filename), KEY idx_vicidial_log_id (vicidial_log_id), KEY idx_processing_status (processing_status) ); 

#!/usr/bin/env python3 """ ViciDial Call Recording QA Scorer using NISQA and Silero VAD Monitors recording directory, scores audio, updates database """ import os import sys import argparse import logging import json import subprocess from pathlib import Path from datetime import datetime import hashlib import MySQLdb import torch import torchaudio import librosa import numpy as np from nisqa.NISQA_model import NISQAModel logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/asterisk/qa_scorer.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class QAScorer: def __init__(self, db_host='localhost', db_user='asteriskuser', db_pass='asterisk', db_name='asterisk', recordings_dir='/var/spool/asterisk/monitor'): self.db_host = db_host self.db_user = db_user self.db_pass = db_pass self.db_name = db_name self.recordings_dir = Path(recordings_dir) # Initialize NISQA model  logger.info("Loading NISQA model...") self.nisqa_model = NISQAModel( pretrained_model_dir='/opt/NISQA/nisqa_pretrained_model' ) # Load Silero VAD  logger.info("Loading Silero VAD...") self.silero_vad = torch.hub.load( 'snakers4/silero-vad', 'get_speech_timestamps', trust_repo=True ) self.silero_get_speech_ts = torch.hub.load( 'snakers4/silero-vad', 'get_speech_timestamps', trust_repo=True ) self.db_conn = None self.connect_db() def connect_db(self): try: self.db_conn = MySQLdb.connect( host=self.db_host, user=self.db_user, passwd=self.db_pass, db=self.db_name ) logger.info("Database connected") except MySQLdb.Error as e: logger.error(f"Database connection failed: {e}") sys.exit(1) def decode_audio(self, filepath, target_sr=16000): """Decode any audio format to 16kHz mono PCM""" if filepath.suffix.lower() == '.wav': try: waveform, sr = torchaudio.load(filepath) if sr != target_sr: resampler = torchaudio.transforms.Resample(sr, target_sr) waveform = resampler(waveform) if waveform.shape[0] > 1: waveform = waveform.mean(dim=0, keepdim=True) return waveform.squeeze().numpy(), target_sr except Exception as e: logger.error(f"Decode error {filepath}: {e}") return None, None else: # Use ffmpeg for other formats  try: temp_wav = f"/tmp/{filepath.stem}_temp.wav" cmd = [ 'ffmpeg', '-i', str(filepath), '-ar', str(target_sr), '-ac', '1', '-f', 'wav', temp_wav, '-y' ] subprocess.run(cmd, check=True, capture_output=True, timeout=30) waveform, sr = torchaudio.load(temp_wav) os.remove(temp_wav) return waveform.squeeze().numpy(), sr except Exception as e: logger.error(f"FFmpeg decode error {filepath}: {e}") return None, None def score_nisqa(self, waveform, sr): """Score audio with NISQA""" try: # NISQA expects dict input with 'speech' key  result = self.nisqa_model.predict( torch.from_numpy(waveform).float().unsqueeze(0), sr=sr ) return { 'mos_score': float(result[0]['mos']), 'mos_raw': float(result[0]['mos_raw']), 'dim_col': float(result[0]['dim_col']), 'dim_noi': float(result[0]['dim_noi']), 'dim_mos': float(result[0]['dim_mos']) } except Exception as e: logger.error(f"NISQA scoring error: {e}") return None def score_vad(self, waveform, sr): """Score voice activity with Silero VAD""" try: # Silero VAD expects float32 tensor  wav_tensor = torch.from_numpy(waveform).float() if wav_tensor.dim() == 1: wav_tensor = wav_tensor.unsqueeze(0) # Get speech timestamps (list of dicts with 'start' and 'end' in samples)  speech_ts = self.silero_get_speech_ts(wav_tensor, sr) if not speech_ts: return { 'speech_percentage': 0.0, 'silence_duration_sec': len(waveform) / sr, 'num_speech_segments': 0, 'clipping_detected': False, 'noise_floor_db': -80.0 } total_speech_samples = sum( ts['end'] - ts['start'] for ts in speech_ts ) speech_percentage = (total_speech_samples / len(waveform)) * 100.0 silence_duration = (len(waveform) - total_speech_samples) / sr # Detect clipping (samples at near max amplitude)  max_val = np.max(np.abs(waveform)) clipping_threshold = 0.99 clipping_detected = max_val > clipping_threshold # Estimate noise floor from silent segments  if len(speech_ts) > 0: silent_regions = [] if speech_ts[0]['start'] > 0: silent_regions.append(waveform[:speech_ts[0]['start']]) for i in range(len(speech_ts) - 1): gap = waveform[speech_ts[i]['end']:speech_ts[i+1]['start']] if len(gap) > sr * 0.05: # Gaps larger than 50ms  silent_regions.append(gap) if silent_regions: noise = np.concatenate(silent_regions) rms = np.sqrt(np.mean(noise ** 2)) noise_db = 20 * np.log10(rms + 1e-10) else: noise_db = -80.0 else: noise_db = -80.0 return { 'speech_percentage': float(speech_percentage), 'silence_duration_sec': float(silence_duration), 'num_speech_segments': len(speech_ts), 'clipping_detected': bool(clipping_detected), 'noise_floor_db': float(noise_db) } except Exception as e: logger.error(f"VAD scoring error: {e}") return None def get_vicidial_log_id(self, filename): """Extract or look up vicidial_log_id from filename or database""" # ViciDial filenames are typically YYYYMMDD-HHMMSS_UNIQUEID.wav  # Try to find in vicidial_log by matching filename  try: cursor = self.db_conn.cursor() cursor.execute(""" SELECT vicidial_log_id FROM vicidial_log WHERE recording LIKE CONCAT('%', %s, '%') ORDER BY call_date DESC LIMIT 1 """, (filename,)) row = cursor.fetchone() cursor.close() return row[0] if row else None except MySQLdb.Error as e: logger.error(f"Query error: {e}") return None def process_file(self, filepath): """Score a single recording file""" filename = filepath.name # Check if already processed  try: cursor = self.db_conn.cursor() cursor.execute( "SELECT qa_id FROM vicidial_qa_scores WHERE filename = %s", (filename,) ) if cursor.fetchone(): logger.info(f"Skipping {filename} (already scored)") cursor.close() return cursor.close() except MySQLdb.Error as e: logger.error(f"Lookup error: {e}") return logger.info(f"Processing {filename}") # Decode audio  waveform, sr = self.decode_audio(filepath) if waveform is None: self.insert_qa_result( filename, None, processing_status='error', error_message='Audio decode failed' ) return # Run NISQA  nisqa_result = self.score_nisqa(waveform, sr) if nisqa_result is None: self.insert_qa_result( filename, None, processing_status='error', error_message='NISQA scoring failed' ) return # Run Silero VAD  vad_result = self.score_vad(waveform, sr) if vad_result is None: self.insert_qa_result( filename, nisqa_result, processing_status='error', error_message='VAD scoring failed' ) return # Insert results  self.insert_qa_result( filename, nisqa_result, vad_result, processing_status='processed' ) logger.info(f"Scored {filename}: NISQA={nisqa_result['mos_score']:.2f}, Speech={vad_result['speech_percentage']:.1f}%") def insert_qa_result(self, filename, nisqa_result=None, vad_result=None, processing_status='pending', error_message=None): """Insert QA scores into database""" try: vicidial_log_id = self.get_vicidial_log_id(filename) cursor = self.db_conn.cursor() cursor.execute(""" INSERT INTO vicidial_qa_scores ( filename, vicidial_log_id, nisqa_score, nisqa_raw, nisqa_dim_col, nisqa_dim_noi, nisqa_dim_mos, speech_percentage, silence_duration_sec, clipping_detected, processing_status, error_message ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( filename, vicidial_log_id, nisqa_result['mos_score'] if nisqa_result else None, nisqa_result['mos_raw'] if nisqa_result else None, nisqa_result[Source：https://dev.to/moisi_trungu_31647b7ac300/auto-qa-call-recording-quality-with-nisqa-silero-vad-1c3b
My favorites