Upload 73 files

__init__.py

@@ -0,0 +1,169 @@
+class Metric:
+    def __init__(self, name, window, hop=None, verbose=False):
+        # the metric operates on some fixed rate only or only on mono ?
+        self.fixed_rate = None
+        self.mono = False
+
+        # is the metric absolute or relative ?
+        self.absolute = False
+
+        # length and hop of windows
+        self.window = window
+        if hop is None:
+            hop = window
+        self.hop = hop
+        self.name = name
+        self.verbose = verbose
+
+    def test_window(self, audios, rate):
+        raise NotImplementedError
+
+    def test(self, *test_files, array_rate=None):
+        """loading sound files and making sure they all have the same lengths
+        (zero-padding to the largest). Also works with numpy arrays.
+        Then, calling the `test_window` function that should be specialised
+        depending on the metric."""
+
+        # imports
+        import soundfile as sf
+        import resampy
+        from museval.metrics import Framing
+        import numpy as np
+
+        audios = []
+        maxlen = 0
+        if isinstance(test_files, str):
+            test_files = [test_files]
+        if self.absolute and len(test_files) > 1:
+            if self.verbose:
+                print('  [%s] is absolute. Processing first file only'
+                      % self.name)
+            test_files = [test_files[0],]
+
+        for file in test_files:
+            # Loading sound file
+            if isinstance(file, str):
+                audio, rate = sf.read(file, always_2d=True)
+            else:
+                rate = array_rate
+                if rate is None:
+                    raise ValueError('Sampling rate needs to be specified '
+                                     'when feeding numpy arrays.')
+                audio = file
+                # Standardize shapes
+                if len(audio.shape) == 1:
+                    audio = audio[:, None]
+                if len(audio.shape) != 2:
+                    raise ValueError('Please provide 1D or 2D array, received '
+                                     '{}D array'.format(len(audio.shape)))
+
+            if self.fixed_rate is not None and rate != self.fixed_rate:
+                if self.verbose:
+                    print('  [%s] preferred is %dkHz rate. resampling'
+                          % (self.name, self.fixed_rate))
+                audio = resampy.resample(audio, rate, self.fixed_rate, axis=0)
+                rate = self.fixed_rate
+            if self.mono and audio.shape[1] > 1:
+                if self.verbose:
+                    print('  [%s] only supports mono. Will use first channel'
+                          % self.name)
+                audio = audio[..., 0, None]
+            if self.mono:
+                audio = audio[..., 0]
+            maxlen = max(maxlen, audio.shape[0])
+            audios += [audio]
+
+        for index, audio in enumerate(audios):
+            if audio.shape[0] != maxlen:
+                new = np.zeros((maxlen,) + audio.shape[1:])
+                new[:audio.shape[0]] = audio
+                audios[index] = new
+
+        if self.window is not None:
+            framer = Framing(self.window * rate,
+                             self.hop * rate, maxlen)
+            nwin = framer.nwin
+            result = {}
+            for (t, win) in enumerate(framer):
+                result_t = self.test_window([audio[win] for audio in audios],
+                                            rate)
+                for metric in result_t.keys():
+                    if metric not in result.keys():
+                        result[metric] = np.empty(nwin)
+                    result[metric][t] = result_t[metric]
+        else:
+            result = self.test_window(audios, rate)
+        return result
+
+
+import absolute
+import relative
+
+
+class MetricsList:
+    def __init__(self):
+        self.metrics = []
+
+    def __add__(self, metric):
+        self.metrics += [metric]
+        return self
+
+    def __str__(self):
+        return 'Metrics: ' + ' '.join([x.name for x in self.metrics])
+
+    def __call__(self, *files, rate=None):
+        result = {}
+        for metric in self.metrics:
+            result_metric = metric.test(*files, array_rate=rate)
+            for name in result_metric.keys():
+                result[name] = result_metric[name]
+        return result
+
+
+def load(metrics='', window=2, verbose=False):
+    """ Load the desired metrics inside a Metrics object that can then
+    be called to compute all the desired metrics.
+
+    Parameters:
+    ----------
+    metrics: str or list of str
+        the metrics matching any of these will be automatically loaded. this
+        match is relative to the structure of the speechmetrics package.
+        For instance:
+        * 'absolute' will match all absolute metrics
+        * 'absolute.srmr' or 'srmr' will only match SRMR
+        * '' will match all
+
+    window: float
+        the window length to use for testing the files.
+
+    verbose: boolean
+        will display information during computations
+
+    Returns:
+    --------
+
+    A MetricsList object, that can be run to get the desired metrics
+    """
+    import pkgutil
+    import importlib
+
+    result = MetricsList()
+
+    found_modules = []
+    iterator = pkgutil.walk_packages(__path__, __name__ + '.')
+
+    if isinstance(metrics, str):
+        metrics = [metrics]
+    for module_info in iterator:
+        if any([metric in module_info.name for metric in metrics]):
+            module = importlib.import_module(module_info.name)
+            if module not in found_modules:
+                found_modules += [module],
+                if hasattr(module, 'load'):
+                    load_function = getattr(module, 'load')
+                    new_metric = load_function(window)
+                    new_metric.verbose = verbose
+                    result += new_metric
+                    print('Loaded ', module_info.name)
+    return result

basis.py

@@ -0,0 +1,113 @@
+class ScoreBasis:
+    def __init__(self, name=None):
+        # the score operates on the specified rate
+        self.score_rate = None
+        # is the score intrusive or non-intrusive ?
+        self.intrusive = True #require a reference
+        self.name = name
+
+    def windowed_scoring(self, audios, score_rate):
+        raise NotImplementedError(f'In {self.name}, windowed_scoring is not yet implemented')
+
+    def scoring(self, data, window=None, score_rate=None):
+        """ calling the `windowed_scoring` function that should be specialised
+        depending on the score."""
+
+        # imports
+        #import soundfile as sf
+        import resampy
+        from museval.metrics import Framing
+
+        #checking rate
+        audios = data['audio']
+        score_rate = data['rate']
+
+        if self.score_rate is not None:
+            score_rate = self.score_rate
+
+        if score_rate != data['rate']:
+            for index, audio in enumerate(audios):
+                audio = resampy.resample(audio, data['rate'], score_rate, axis=0)
+                audios[index] = audio
+
+        if window is not None:
+            framer = Framing(window * score_rate, window * score_rate, maxlen)
+            nwin = framer.nwin
+            result = {}
+            for (t, win) in enumerate(framer):
+                result_t = self.windowed_scoring([audio[win] for audio in audios], score_rate)
+                result[t] = result_t
+        else:
+            result = self.windowed_scoring(audios, score_rate)
+        return result
+        """
+        audios = []
+        maxlen = 0
+        if isinstance(test_files, str):
+            test_files = [test_files]
+        print(f'test_files: {test_files}')
+        if not self.intrusive and len(test_files) > 1:
+            if self.verbose:
+                print('  [%s] is non-intrusive. Processing first file only'
+                      % self.name)
+            test_files = [test_files[0],]
+        for file in test_files:
+            # Loading sound file
+            if isinstance(file, str):
+                audio, rate = sf.read(file, always_2d=True)
+            else:
+                rate = array_rate
+                if rate is None:
+                    raise ValueError('Sampling rate needs to be specified '
+                                     'when feeding numpy arrays.')
+                audio = file
+                # Standardize shapes
+                if len(audio.shape) == 1:
+                    audio = audio[:, None]
+                if len(audio.shape) != 2:
+                    raise ValueError('Please provide 1D or 2D array, received '
+                                     '{}D array'.format(len(audio.shape)))
+
+            if self.fixed_rate is not None and rate != self.fixed_rate:
+                if self.verbose:
+                    print('  [%s] preferred is %dkHz rate. resampling'
+                          % (self.name, self.fixed_rate))
+                audio = resampy.resample(audio, rate, self.fixed_rate, axis=0)
+                rate = self.fixed_rate
+            if self.mono and audio.shape[1] > 1:
+                if self.verbose:
+                    print('  [%s] only supports mono. Will use first channel'
+                          % self.name)
+                audio = audio[..., 0, None]
+            if self.mono:
+                audio = audio[..., 0]
+            maxlen = max(maxlen, audio.shape[0])
+            audios += [audio]
+        audio = audios[1]
+        audio[:maxlen-320] = audio[320:]
+        audios[1] = audio
+        for index, audio in enumerate(audios):
+            if audio.shape[0] != maxlen:
+                new = np.zeros((maxlen,) + audio.shape[1:])
+                new[:audio.shape[0]] = audio
+                audios[index] = new
+
+        if self.window is not None:
+            framer = Framing(self.window * rate,
+                             self.hop * rate, maxlen)
+            nwin = framer.nwin
+            result = {}
+            for (t, win) in enumerate(framer):
+                result_t = self.test_window([audio[win] for audio in audios],
+                                            rate)
+                #or metric in result_t.keys():
+                #   if metric not in result.keys():
+                #       result[metric] = np.empty(nwin)
+                #   result[metric][t] = result_t[metric]
+                result[t] = result_t
+        else:
+            result = self.test_window(audios, rate)
+        return result
+        """
+
+

demo.py

@@ -0,0 +1,29 @@
+# Import pprint for pretty-printing the results in a more readable format
+import pprint
+# Import the SpeechScore class to evaluate speech quality metrics
+from speechscore import SpeechScore 
+
+# Main block to ensure the code runs only when executed directly
+if __name__ == '__main__':
+    # Initialize a SpeechScore object with a list of score metrics to be evaluated
+    # Supports any subsets of the list
+    mySpeechScore = SpeechScore([
+        'SRMR', 'PESQ', 'NB_PESQ', 'STOI', 'SISDR', 
+        'FWSEGSNR', 'LSD', 'BSSEval', 'DNSMOS', 
+        'SNR', 'SSNR', 'LLR', 'CSIG', 'CBAK', 
+        'COVL', 'MCD'
+    ])
+
+    # Call the SpeechScore object to evaluate the speech metrics between 'noisy' and 'clean' audio
+    # Arguments:
+    # - {test_path, reference_path} supports audio directories or audio paths (.wav or .flac)
+    # - window (float): seconds, set None to specify no windowing (process the full audio)
+    # - score_rate (int): specifies the sampling rate at which the metrics should be computed
+    # - return_mean (bool): set True to specify that the mean score for each metric should be returned
+    scores = mySpeechScore(test_path='audios/noisy/', reference_path='audios/clean/', window=None, score_rate=16000, return_mean=True)
+
+    # Pretty-print the resulting scores in a readable format
+    pprint.pprint(scores)
+
+    # Print only the resulting mean scores in a readable format
+    pprint.pprint(scores['Mean_Score'])

requirement.txt

@@ -0,0 +1,5 @@
+pysptk
+pymcd
+pyworld
+fastdtw
+museval

scores/bsseval.py

@@ -0,0 +1,21 @@
+import numpy as np
+from basis import ScoreBasis
+
+
+class BSSEval(ScoreBasis):
+    def __init__(self):
+        super(BSSEval, self).__init__(name='BSSEval')
+        self.intrusive = False
+
+    def windowed_scoring(self, audios, score_rate):
+        bss_window = np.inf
+        bss_hop = np.inf
+        from museval.metrics import bss_eval
+        if len(audios) != 2:
+            raise ValueError('BSSEval needs a reference and a test signals.')
+        
+        result = bss_eval(reference_sources=audios[1][None,...], # shape: [nsrc, nsample, nchannels]
+                        estimated_sources=audios[0][None,...],
+                        window=bss_window * score_rate,
+                        hop=bss_hop * score_rate)
+        return {'SDR': result[0][0][0], 'ISR': result[1][0][0], 'SAR': result[3][0][0]}

scores/cbak.py

@@ -0,0 +1,37 @@
+from basis import ScoreBasis
+import numpy as np
+from pesq import pesq
+from scores.helper import wss, llr, SSNR, trim_mos
+
+class CBAK(ScoreBasis):
+    def __init__(self):
+        super(CBAK, self).__init__(name='CBAK')
+        self.score_rate = 16000
+        self.intrusive = False
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('CBAK needs a reference and a test signals.')
+        return cal_CBAK(audios[0], audios[1], score_rate)
+
+def cal_CBAK(target_wav, pred_wav, fs):
+    alpha   = 0.95
+
+    # Compute WSS measure
+    wss_dist_vec = wss(target_wav, pred_wav, fs)
+    wss_dist_vec = sorted(wss_dist_vec, reverse=False)
+    wss_dist     = np.mean(wss_dist_vec[:int(round(len(wss_dist_vec) * alpha))])
+
+    # Compute the SSNR
+    snr_mean, segsnr_mean = SSNR(target_wav, pred_wav, fs)
+    segSNR = np.mean(segsnr_mean)
+
+    # Compute the PESQ
+    pesq_raw = pesq(fs, target_wav, pred_wav, 'wb')
+
+    # Cbak
+    Cbak = 1.634 + 0.478 * pesq_raw - 0.007 * wss_dist + 0.063 * segSNR
+    Cbak = trim_mos(Cbak)
+
+    return Cbak
+

scores/covl.py

@@ -0,0 +1,39 @@
+from basis import ScoreBasis
+import numpy as np
+from pesq import pesq
+from scores.helper import wss, llr, SSNR, trim_mos
+
+class COVL(ScoreBasis):
+    def __init__(self):
+        super(COVL, self).__init__(name='COVL')
+        self.score_rate = 16000
+        self.intrusive = False
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('COVL needs a reference and a test signals.')
+        return cal_COVL(audios[0], audios[1], score_rate)
+
+def cal_COVL(target_wav, pred_wav, fs):
+    alpha   = 0.95
+
+    # Compute WSS measure
+    wss_dist_vec = wss(target_wav, pred_wav, fs)
+    wss_dist_vec = sorted(wss_dist_vec, reverse=False)
+    wss_dist     = np.mean(wss_dist_vec[:int(round(len(wss_dist_vec) * alpha))])
+
+    # Compute LLR measure
+    LLR_dist = llr(target_wav, pred_wav, fs)
+    LLR_dist = sorted(LLR_dist, reverse=False)
+    LLRs     = LLR_dist
+    LLR_len  = round(len(LLR_dist) * alpha)
+    llr_mean = np.mean(LLRs[:LLR_len])
+
+    # Compute the PESQ
+    pesq_raw = pesq(fs, target_wav, pred_wav, 'wb')
+
+    # Covl
+    Covl = 1.594 + 0.805 * pesq_raw - 0.512 * llr_mean - 0.007 * wss_dist
+    Covl = trim_mos(Covl)
+
+    return Covl

scores/csig.py

@@ -0,0 +1,38 @@
+from basis import ScoreBasis
+import numpy as np
+from pesq import pesq
+from scores.helper import wss, llr, SSNR, trim_mos
+
+class CSIG(ScoreBasis):
+    def __init__(self):
+        super(CSIG, self).__init__(name='CSIG')
+        self.score_rate = 16000
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('CSIG needs a reference and a test signals.')
+        return cal_CSIG(audios[0], audios[1], score_rate)
+
+def cal_CSIG(target_wav, pred_wav, fs):
+    alpha   = 0.95
+
+    # Compute WSS measure
+    wss_dist_vec = wss(target_wav, pred_wav, fs)
+    wss_dist_vec = sorted(wss_dist_vec, reverse=False)
+    wss_dist     = np.mean(wss_dist_vec[:int(round(len(wss_dist_vec) * alpha))])
+
+    # Compute LLR measure
+    LLR_dist = llr(target_wav, pred_wav, fs)
+    LLR_dist = sorted(LLR_dist, reverse=False)
+    LLRs     = LLR_dist
+    LLR_len  = round(len(LLR_dist) * alpha)
+    llr_mean = np.mean(LLRs[:LLR_len])
+
+    # Compute the PESQ
+    pesq_raw = pesq(fs, target_wav, pred_wav, 'wb')
+
+    # Csig
+    Csig = 3.093 - 1.029 * llr_mean + 0.603 * pesq_raw - 0.009 * wss_dist
+    Csig = float(trim_mos(Csig))
+    
+    return Csig

scores/dnsmos/DNSMOS/bak_ovr.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5f335c90994618150192a656a474bcf8a9cbcedbc47965494ba8da79605d1308
+size 742375

scores/dnsmos/DNSMOS/model_v8.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9246480c58567bc6affd4200938e77eef49468c8bc7ed3776d109c07456f6e91
+size 224860

scores/dnsmos/DNSMOS/sig.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d2fbdb293bc2366dfbae2b7477c490f981d24a8b4405efd3c11787569c6549d7
+size 742203

scores/dnsmos/DNSMOS/sig_bak_ovr.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:269fbebdb513aa23cddfbb593542ecc540284a91849ac50516870e1ac78f6edd
+size 1157965

scores/dnsmos/dnsmos.py

@@ -0,0 +1,94 @@
+import os
+
+import librosa
+import numpy as np
+import numpy.polynomial.polynomial as poly
+import onnxruntime as ort
+import soundfile as sf
+
+SAMPLING_RATE = 16000
+INPUT_LENGTH = 9.01
+
+from basis import ScoreBasis
+
+
+class DNSMOS(ScoreBasis):
+    def __init__(self):
+        super(DNSMOS, self).__init__(name='DNSMOS')
+        self.intrusive = True
+        self.score_rate = 16000
+        self.p808_model_path = os.path.join('scores/dnsmos/DNSMOS', 'model_v8.onnx')    
+        self.primary_model_path = os.path.join('scores/dnsmos/DNSMOS', 'sig_bak_ovr.onnx')
+        self.compute_score = ComputeScore(self.primary_model_path, self.p808_model_path)
+
+    def windowed_scoring(self, audios, rate):
+        if len(audios) == 2:
+            return self.compute_score.cal_mos(audios[1], rate)
+        else:
+            return self.compute_score.cal_mos(audios[0], rate)
+
+class ComputeScore:
+    def __init__(self, primary_model_path, p808_model_path) -> None:
+        self.onnx_sess = ort.InferenceSession(primary_model_path)
+        self.p808_onnx_sess = ort.InferenceSession(p808_model_path)
+        
+    def audio_melspec(self, audio, n_mels=120, frame_size=320, hop_length=160, sr=16000, to_db=True):
+        mel_spec = librosa.feature.melspectrogram(y=audio, sr=sr, n_fft=frame_size+1, hop_length=hop_length, n_mels=n_mels)
+        if to_db:
+            mel_spec = (librosa.power_to_db(mel_spec, ref=np.max)+40)/40
+        return mel_spec.T
+
+    def get_polyfit_val(self, sig, bak, ovr):
+        p_ovr = np.poly1d([-0.06766283,  1.11546468,  0.04602535])
+        p_sig = np.poly1d([-0.08397278,  1.22083953,  0.0052439 ])
+        p_bak = np.poly1d([-0.13166888,  1.60915514, -0.39604546])
+
+        sig_poly = p_sig(sig)
+        bak_poly = p_bak(bak)
+        ovr_poly = p_ovr(ovr)
+
+        return sig_poly, bak_poly, ovr_poly
+
+    def cal_mos(self, audio, sampling_rate):
+        fs = sampling_rate
+        actual_audio_len = len(audio)
+        len_samples = int(INPUT_LENGTH*fs)
+        while len(audio) < len_samples:
+            audio = np.append(audio, audio)
+        
+        num_hops = int(np.floor(len(audio)/fs) - INPUT_LENGTH)+1
+        hop_len_samples = fs
+        predicted_mos_sig_seg_raw = []
+        predicted_mos_bak_seg_raw = []
+        predicted_mos_ovr_seg_raw = []
+        predicted_mos_sig_seg = []
+        predicted_mos_bak_seg = []
+        predicted_mos_ovr_seg = []
+        predicted_p808_mos = []
+
+        for idx in range(num_hops):
+            audio_seg = audio[int(idx*hop_len_samples) : int((idx+INPUT_LENGTH)*hop_len_samples)]
+            if len(audio_seg) < len_samples:
+                continue
+
+            input_features = np.array(audio_seg).astype('float32')[np.newaxis,:]
+            p808_input_features = np.array(self.audio_melspec(audio=audio_seg[:-160])).astype('float32')[np.newaxis, :, :]
+            oi = {'input_1': input_features}
+            p808_oi = {'input_1': p808_input_features}
+            p808_mos = self.p808_onnx_sess.run(None, p808_oi)[0][0][0]
+            mos_sig_raw,mos_bak_raw,mos_ovr_raw = self.onnx_sess.run(None, oi)[0][0]
+            mos_sig,mos_bak,mos_ovr = self.get_polyfit_val(mos_sig_raw,mos_bak_raw,mos_ovr_raw)
+            predicted_mos_sig_seg_raw.append(mos_sig_raw)
+            predicted_mos_bak_seg_raw.append(mos_bak_raw)
+            predicted_mos_ovr_seg_raw.append(mos_ovr_raw)
+            predicted_mos_sig_seg.append(mos_sig)
+            predicted_mos_bak_seg.append(mos_bak)
+            predicted_mos_ovr_seg.append(mos_ovr)
+            predicted_p808_mos.append(p808_mos)
+
+        results = {}
+        results['OVRL'] = np.mean(predicted_mos_ovr_seg)
+        results['SIG'] = np.mean(predicted_mos_sig_seg)
+        results['BAK'] = np.mean(predicted_mos_bak_seg)
+        results['P808_MOS'] = np.mean(predicted_p808_mos)
+        return results

scores/fwsegsnr.py

@@ -0,0 +1,49 @@
+import librosa
+import numpy as np
+from basis import ScoreBasis
+
+class FWSEGSNR(ScoreBasis):
+    def __init__(self):
+        super(FWSEGSNR, self).__init__(name='FWSEGSNR')
+        self.intrusive = False
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('FWSEGSNR needs a reference and a test signals.')
+        return fwsegsnr(audios[1], audios[0], score_rate)
+
+def fwsegsnr(x, y, fs, frame_sz = 0.025, shift_sz= 0.01, win='hann', numband=23):
+    epsilon = np.finfo(np.float32).eps
+    frame = int(np.fix(frame_sz * fs))
+    shift = int(np.fix(shift_sz * fs))
+    window = win
+    nband = numband
+    noverlap = frame - shift
+    fftpt = int(2**np.ceil(np.log2(np.abs(frame))))
+    x = x / np.sqrt(sum(np.power(x, 2)))
+    y = y / np.sqrt(sum(np.power(y, 2)))
+
+    assert len(x) == len(y), print('Wav length are not matched!')
+    X_stft = np.abs(librosa.stft(x, n_fft=fftpt, hop_length=shift, win_length=frame, window=window, center=False))
+    Y_stft = np.abs(librosa.stft(y, n_fft=fftpt, hop_length=shift, win_length=frame, window=window, center=False))
+
+    num_freq = X_stft.shape[0]
+    num_frame = X_stft.shape[1]
+
+    X_mel = librosa.feature.melspectrogram(S=X_stft, sr=fs, n_mels=nband, fmin=0, fmax=fs/2)
+    Y_mel = librosa.feature.melspectrogram(S=Y_stft, sr=fs, n_mels=nband, fmin=0, fmax=fs/2)
+
+    # Calculate SNR.
+
+    W = np.power(Y_mel, 0.2)
+    E = X_mel - Y_mel
+    E[E == 0.0] = epsilon
+    E_power = np.power(E, 2)
+    Y_div_E = np.divide((np.power(Y_mel,2)), (np.power(E,2)))
+    Y_div_E[Y_div_E==0] = epsilon
+    ds = 10 * np.divide(np.sum(np.multiply(W, np.log10(Y_div_E)), 1), np.sum(W, 1))
+    ds[ds > 35] = 35
+    ds[ds < -10] = -10
+    d = np.mean(ds)
+    return d
+

scores/helper.py

@@ -0,0 +1,307 @@
+"""
+Modifications in Metrics
+
+# Original copyright:
+# Copyright (c) Facebook, Inc. and its affiliates.
+# Demucs (https://github.com/facebookresearch/denoiser) / author: adefossez
+"""
+import numpy as np
+from scipy.linalg import toeplitz
+
+# ----------------------------- HELPERS ------------------------------------ #
+def trim_mos(val):
+    return min(max(val, 1), 5)
+
+def lpcoeff(speech_frame, model_order):
+    # (1) Compute Autocor lags
+    winlength = speech_frame.shape[0]
+    R = []
+    for k in range(model_order + 1):
+        first  = speech_frame[:(winlength - k)]
+        second = speech_frame[k:winlength]
+        R.append(np.sum(first * second))
+
+    # (2) Lev-Durbin
+    a = np.ones((model_order,))
+    E = np.zeros((model_order + 1,))
+    rcoeff = np.zeros((model_order,))
+    E[0] = R[0]
+    for i in range(model_order):
+        if i == 0:
+            sum_term = 0
+        else:
+            a_past = a[:i]
+            sum_term = np.sum(a_past * np.array(R[i:0:-1]))
+        rcoeff[i] = (R[i+1] - sum_term)/E[i]
+        a[i] = rcoeff[i]
+        if i > 0:
+            a[:i] = a_past[:i] - rcoeff[i] * a_past[::-1]
+        E[i+1] = (1-rcoeff[i]*rcoeff[i])*E[i]
+    acorr    = np.array(R, dtype=np.float32)
+    refcoeff = np.array(rcoeff, dtype=np.float32)
+    a = a * -1
+    lpparams = np.array([1] + list(a), dtype=np.float32)
+    acorr    = np.array(acorr, dtype=np.float32)
+    refcoeff = np.array(refcoeff, dtype=np.float32)
+    lpparams = np.array(lpparams, dtype=np.float32)
+
+    return acorr, refcoeff, lpparams
+# -------------------------------------------------------------------------- #
+
+
+def SSNR(ref_wav, deg_wav, srate=16000, eps=1e-10):
+    """ Segmental Signal-to-Noise Ratio Objective Speech Quality Measure
+        This function implements the segmental signal-to-noise ratio
+        as defined in [1, p. 45] (see Equation 2.12).
+    """
+    clean_speech     = ref_wav
+    processed_speech = deg_wav
+    clean_length     = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+    
+    # scale both to have same dynamic range. Remove DC too.
+    clean_speech     -= clean_speech.mean()
+    processed_speech -= processed_speech.mean()
+    processed_speech *= (np.max(np.abs(clean_speech)) / np.max(np.abs(processed_speech)))
+   
+    # Signal-to-Noise Ratio 
+    dif = ref_wav - deg_wav
+    overall_snr = 10 * np.log10(np.sum(ref_wav ** 2) / (np.sum(dif ** 2) +
+                                                        10e-20))
+    # global variables
+    winlength = int(np.round(30 * srate / 1000)) # 30 msecs
+    skiprate  = winlength // 4
+    MIN_SNR   = -10
+    MAX_SNR   = 35
+
+    # For each frame, calculate SSNR
+    num_frames    = int(clean_length / skiprate - (winlength/skiprate))
+    start         = 0
+    time          = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window        = 0.5 * (1 - np.cos(2 * np.pi * time))
+    segmental_snr = []
+
+    for frame_count in range(int(num_frames)):
+        # (1) get the frames for the test and ref speech.
+        # Apply Hanning Window
+        clean_frame     = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame     = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Compute Segmental SNR
+        signal_energy = np.sum(clean_frame ** 2)
+        noise_energy  = np.sum((clean_frame - processed_frame) ** 2)
+        segmental_snr.append(10 * np.log10(signal_energy / (noise_energy + eps)+ eps))
+        segmental_snr[-1] = max(segmental_snr[-1], MIN_SNR)
+        segmental_snr[-1] = min(segmental_snr[-1], MAX_SNR)
+        start += int(skiprate)
+    return overall_snr, segmental_snr
+
+
+def wss(ref_wav, deg_wav, srate):
+    clean_speech     = ref_wav
+    processed_speech = deg_wav
+    clean_length     = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+
+    assert clean_length == processed_length, clean_length
+
+    winlength = round(30 * srate / 1000.) # 240 wlen in samples
+    skiprate  = np.floor(winlength / 4)
+    max_freq  = srate / 2
+    num_crit  = 25 # num of critical bands
+
+    USE_FFT_SPECTRUM = 1
+    n_fft    = int(2 ** np.ceil(np.log(2*winlength)/np.log(2)))
+    n_fftby2 = int(n_fft / 2)
+    Kmax     = 20
+    Klocmax  = 1
+
+    # Critical band filter definitions (Center frequency and BW in Hz)
+    cent_freq = [50., 120, 190, 260, 330, 400, 470, 540, 617.372,
+                 703.378, 798.717, 904.128, 1020.38, 1148.30, 
+                 1288.72, 1442.54, 1610.70, 1794.16, 1993.93, 
+                 2211.08, 2446.71, 2701.97, 2978.04, 3276.17,
+                 3597.63]
+    bandwidth = [70., 70, 70, 70, 70, 70, 70, 77.3724, 86.0056,
+                 95.3398, 105.411, 116.256, 127.914, 140.423, 
+                 153.823, 168.154, 183.457, 199.776, 217.153, 
+                 235.631, 255.255, 276.072, 298.126, 321.465,
+                 346.136]
+
+    bw_min = bandwidth[0] # min critical bandwidth
+
+    # set up critical band filters. Note here that Gaussianly shaped filters
+    # are used. Also, the sum of the filter weights are equivalent for each
+    # critical band filter. Filter less than -30 dB and set to zero.
+    min_factor = np.exp(-30. / (2 * 2.303)) # -30 dB point of filter
+
+    crit_filter = np.zeros((num_crit, n_fftby2))
+    all_f0 = []
+    for i in range(num_crit):
+        f0 = (cent_freq[i] / max_freq) * (n_fftby2)
+        all_f0.append(np.floor(f0))
+        bw = (bandwidth[i] / max_freq) * (n_fftby2)
+        norm_factor = np.log(bw_min) - np.log(bandwidth[i])
+        j = list(range(n_fftby2))
+        crit_filter[i, :] = np.exp(-11 * (((j - np.floor(f0)) / bw) ** 2) + \
+                                   norm_factor)
+        crit_filter[i, :] = crit_filter[i, :] * (crit_filter[i, :] > \
+                                                 min_factor)
+
+    # For each frame of input speech, compute Weighted Spectral Slope Measure
+    num_frames = int(clean_length / skiprate - (winlength / skiprate))
+    start = 0 # starting sample
+    time = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window = 0.5 * (1 - np.cos(2 * np.pi * time))
+    distortion = []
+
+    for frame_count in range(num_frames):
+        # (1) Get the Frames for the test and reference speeech.
+        # Multiply by Hanning window.
+        clean_frame = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Compuet Power Spectrum of clean and processed
+        clean_spec = (np.abs(np.fft.fft(clean_frame, n_fft)) ** 2)
+        processed_spec = (np.abs(np.fft.fft(processed_frame, n_fft)) ** 2)
+        clean_energy = [None] * num_crit
+        processed_energy = [None] * num_crit
+
+        # (3) Compute Filterbank output energies (in dB)
+        for i in range(num_crit):
+            clean_energy[i] = np.sum(clean_spec[:n_fftby2] * \
+                                     crit_filter[i, :])
+            processed_energy[i] = np.sum(processed_spec[:n_fftby2] * \
+                                         crit_filter[i, :])
+        clean_energy = np.array(clean_energy).reshape(-1, 1)
+        eps = np.ones((clean_energy.shape[0], 1)) * 1e-10
+        clean_energy = np.concatenate((clean_energy, eps), axis=1)
+        clean_energy = 10 * np.log10(np.max(clean_energy, axis=1))
+        processed_energy = np.array(processed_energy).reshape(-1, 1)
+        processed_energy = np.concatenate((processed_energy, eps), axis=1)
+        processed_energy = 10 * np.log10(np.max(processed_energy, axis=1))
+
+        # (4) Compute Spectral Shape (dB[i+1] - dB[i])
+        clean_slope = clean_energy[1:num_crit] - clean_energy[:num_crit-1]
+        processed_slope = processed_energy[1:num_crit] - \
+                processed_energy[:num_crit-1]
+
+        # (5) Find the nearest peak locations in the spectra to each
+        # critical band. If the slope is negative, we search
+        # to the left. If positive, we search to the right.
+        clean_loc_peak = []
+        processed_loc_peak = []
+        for i in range(num_crit - 1):
+            if clean_slope[i] > 0:
+                # search to the right
+                n = i
+                while n < num_crit - 1 and clean_slope[n] > 0:
+                    n += 1
+                clean_loc_peak.append(clean_energy[n - 1])
+            else:
+                # search to the left
+                n = i
+                while n >= 0 and clean_slope[n] <= 0:
+                    n -= 1
+                clean_loc_peak.append(clean_energy[n + 1])
+            # find the peaks in the processed speech signal
+            if processed_slope[i] > 0:
+                n = i
+                while n < num_crit - 1 and processed_slope[n] > 0:
+                    n += 1
+                processed_loc_peak.append(processed_energy[n - 1])
+            else:
+                n = i
+                while n >= 0 and processed_slope[n] <= 0:
+                    n -= 1
+                processed_loc_peak.append(processed_energy[n + 1])
+
+        # (6) Compuet the WSS Measure for this frame. This includes
+        # determination of the weighting functino
+        dBMax_clean = max(clean_energy)
+        dBMax_processed = max(processed_energy)
+
+        # The weights are calculated by averaging individual
+        # weighting factors from the clean and processed frame.
+        # These weights W_clean and W_processed should range
+        # from 0 to 1 and place more emphasis on spectral 
+        # peaks and less emphasis on slope differences in spectral
+        # valleys.  This procedure is described on page 1280 of
+        # Klatt's 1982 ICASSP paper.
+        clean_loc_peak = np.array(clean_loc_peak)
+        processed_loc_peak = np.array(processed_loc_peak)
+        Wmax_clean = Kmax / (Kmax + dBMax_clean - clean_energy[:num_crit-1])
+        Wlocmax_clean = Klocmax / (Klocmax + clean_loc_peak - \
+                                   clean_energy[:num_crit-1])
+        W_clean = Wmax_clean * Wlocmax_clean
+        Wmax_processed = Kmax / (Kmax + dBMax_processed - \
+                                processed_energy[:num_crit-1])
+        Wlocmax_processed = Klocmax / (Klocmax + processed_loc_peak - \
+                                      processed_energy[:num_crit-1])
+        W_processed = Wmax_processed * Wlocmax_processed
+        W = (W_clean + W_processed) / 2
+        distortion.append(np.sum(W * (clean_slope[:num_crit - 1] - \
+                                     processed_slope[:num_crit - 1]) ** 2))
+
+        # this normalization is not part of Klatt's paper, but helps
+        # to normalize the meaasure. Here we scale the measure by the sum of the
+        # weights
+        distortion[frame_count] = distortion[frame_count] / np.sum(W)
+        start += int(skiprate)
+    return distortion
+
+
+def llr(ref_wav, deg_wav, srate):
+    clean_speech = ref_wav
+    processed_speech = deg_wav
+    clean_length = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+    assert clean_length == processed_length, clean_length
+
+    winlength = round(30 * srate / 1000.) # 240 wlen in samples
+    skiprate = np.floor(winlength / 4)
+    if srate < 10000:
+        # LPC analysis order
+        P = 10
+    else:
+        P = 16
+
+    # For each frame of input speech, calculate the Log Likelihood Ratio
+    num_frames = int(clean_length / skiprate - (winlength / skiprate))
+    start = 0
+    time = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window = 0.5 * (1 - np.cos(2 * np.pi * time))
+    distortion = []
+
+    for frame_count in range(num_frames):
+        # (1) Get the Frames for the test and reference speeech.
+        # Multiply by Hanning window.
+        clean_frame = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Get the autocorrelation logs and LPC params used
+        # to compute the LLR measure
+        R_clean, Ref_clean, A_clean = lpcoeff(clean_frame, P)
+        R_processed, Ref_processed, A_processed = lpcoeff(processed_frame, P)
+        A_clean = A_clean[None, :]
+        A_processed = A_processed[None, :]
+
+        # (3) Compute the LLR measure
+        numerator = A_processed.dot(toeplitz(R_clean)).dot(A_processed.T)
+        denominator = A_clean.dot(toeplitz(R_clean)).dot(A_clean.T)
+
+        if (numerator/denominator) <= 0:
+            print(f'Numerator: {numerator}')
+            print(f'Denominator: {denominator}')
+
+        log_ = np.log(numerator / denominator)
+        distortion.append(np.squeeze(log_))
+        start += int(skiprate)
+    return np.nan_to_num(np.array(distortion))
+# -------------------------------------------------------------------------- #

scores/helper_bk.py

@@ -0,0 +1,438 @@
+"""
+Modifications in Metrics
+
+# Original copyright:
+# Copyright (c) Facebook, Inc. and its affiliates.
+# Demucs (https://github.com/facebookresearch/denoiser) / author: adefossez
+"""
+import numpy as np
+from scipy.linalg import toeplitz
+
+# ----------------------------- HELPERS ------------------------------------ #
+def trim_mos(val):
+    return min(max(val, 1), 5)
+
+def lpcoeff(speech_frame, model_order):
+    # (1) Compute Autocor lags
+    winlength = speech_frame.shape[0]
+    R = []
+    for k in range(model_order + 1):
+        first  = speech_frame[:(winlength - k)]
+        second = speech_frame[k:winlength]
+        R.append(np.sum(first * second))
+
+    # (2) Lev-Durbin
+    a = np.ones((model_order,))
+    E = np.zeros((model_order + 1,))
+    rcoeff = np.zeros((model_order,))
+    E[0] = R[0]
+    for i in range(model_order):
+        if i == 0:
+            sum_term = 0
+        else:
+            a_past = a[:i]
+            sum_term = np.sum(a_past * np.array(R[i:0:-1]))
+        rcoeff[i] = (R[i+1] - sum_term)/E[i]
+        a[i] = rcoeff[i]
+        if i > 0:
+            a[:i] = a_past[:i] - rcoeff[i] * a_past[::-1]
+        E[i+1] = (1-rcoeff[i]*rcoeff[i])*E[i]
+    acorr    = np.array(R, dtype=np.float32)
+    refcoeff = np.array(rcoeff, dtype=np.float32)
+    a = a * -1
+    lpparams = np.array([1] + list(a), dtype=np.float32)
+    acorr    = np.array(acorr, dtype=np.float32)
+    refcoeff = np.array(refcoeff, dtype=np.float32)
+    lpparams = np.array(lpparams, dtype=np.float32)
+
+    return acorr, refcoeff, lpparams
+# -------------------------------------------------------------------------- #
+
+
+def SSNR(ref_wav, deg_wav, srate=16000, eps=1e-10):
+    """ Segmental Signal-to-Noise Ratio Objective Speech Quality Measure
+        This function implements the segmental signal-to-noise ratio
+        as defined in [1, p. 45] (see Equation 2.12).
+    """
+    clean_speech     = ref_wav
+    processed_speech = deg_wav
+    clean_length     = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+    
+    # scale both to have same dynamic range. Remove DC too.
+    clean_speech     -= clean_speech.mean()
+    processed_speech -= processed_speech.mean()
+    processed_speech *= (np.max(np.abs(clean_speech)) / np.max(np.abs(processed_speech)))
+   
+    # Signal-to-Noise Ratio 
+    dif = ref_wav - deg_wav
+    overall_snr = 10 * np.log10(np.sum(ref_wav ** 2) / (np.sum(dif ** 2) +
+                                                        10e-20))
+    # global variables
+    winlength = int(np.round(30 * srate / 1000)) # 30 msecs
+    skiprate  = winlength // 4
+    MIN_SNR   = -10
+    MAX_SNR   = 35
+
+    # For each frame, calculate SSNR
+    num_frames    = int(clean_length / skiprate - (winlength/skiprate))
+    start         = 0
+    time          = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window        = 0.5 * (1 - np.cos(2 * np.pi * time))
+    segmental_snr = []
+
+    for frame_count in range(int(num_frames)):
+        # (1) get the frames for the test and ref speech.
+        # Apply Hanning Window
+        clean_frame     = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame     = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Compute Segmental SNR
+        signal_energy = np.sum(clean_frame ** 2)
+        noise_energy  = np.sum((clean_frame - processed_frame) ** 2)
+        segmental_snr.append(10 * np.log10(signal_energy / (noise_energy + eps)+ eps))
+        segmental_snr[-1] = max(segmental_snr[-1], MIN_SNR)
+        segmental_snr[-1] = min(segmental_snr[-1], MAX_SNR)
+        start += int(skiprate)
+    return overall_snr, segmental_snr
+
+
+def wss(ref_wav, deg_wav, srate):
+    clean_speech     = ref_wav
+    processed_speech = deg_wav
+    clean_length     = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+
+    assert clean_length == processed_length, clean_length
+
+    winlength = round(30 * srate / 1000.) # 240 wlen in samples
+    skiprate  = np.floor(winlength / 4)
+    max_freq  = srate / 2
+    num_crit  = 25 # num of critical bands
+
+    USE_FFT_SPECTRUM = 1
+    n_fft    = int(2 ** np.ceil(np.log(2*winlength)/np.log(2)))
+    n_fftby2 = int(n_fft / 2)
+    Kmax     = 20
+    Klocmax  = 1
+
+    # Critical band filter definitions (Center frequency and BW in Hz)
+    cent_freq = [50., 120, 190, 260, 330, 400, 470, 540, 617.372,
+                 703.378, 798.717, 904.128, 1020.38, 1148.30, 
+                 1288.72, 1442.54, 1610.70, 1794.16, 1993.93, 
+                 2211.08, 2446.71, 2701.97, 2978.04, 3276.17,
+                 3597.63]
+    bandwidth = [70., 70, 70, 70, 70, 70, 70, 77.3724, 86.0056,
+                 95.3398, 105.411, 116.256, 127.914, 140.423, 
+                 153.823, 168.154, 183.457, 199.776, 217.153, 
+                 235.631, 255.255, 276.072, 298.126, 321.465,
+                 346.136]
+
+    bw_min = bandwidth[0] # min critical bandwidth
+
+    # set up critical band filters. Note here that Gaussianly shaped filters
+    # are used. Also, the sum of the filter weights are equivalent for each
+    # critical band filter. Filter less than -30 dB and set to zero.
+    min_factor = np.exp(-30. / (2 * 2.303)) # -30 dB point of filter
+
+    crit_filter = np.zeros((num_crit, n_fftby2))
+    all_f0 = []
+    for i in range(num_crit):
+        f0 = (cent_freq[i] / max_freq) * (n_fftby2)
+        all_f0.append(np.floor(f0))
+        bw = (bandwidth[i] / max_freq) * (n_fftby2)
+        norm_factor = np.log(bw_min) - np.log(bandwidth[i])
+        j = list(range(n_fftby2))
+        crit_filter[i, :] = np.exp(-11 * (((j - np.floor(f0)) / bw) ** 2) + \
+                                   norm_factor)
+        crit_filter[i, :] = crit_filter[i, :] * (crit_filter[i, :] > \
+                                                 min_factor)
+
+    # For each frame of input speech, compute Weighted Spectral Slope Measure
+    num_frames = int(clean_length / skiprate - (winlength / skiprate))
+    start = 0 # starting sample
+    time = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window = 0.5 * (1 - np.cos(2 * np.pi * time))
+    distortion = []
+
+    for frame_count in range(num_frames):
+        # (1) Get the Frames for the test and reference speeech.
+        # Multiply by Hanning window.
+        clean_frame = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Compuet Power Spectrum of clean and processed
+        clean_spec = (np.abs(np.fft.fft(clean_frame, n_fft)) ** 2)
+        processed_spec = (np.abs(np.fft.fft(processed_frame, n_fft)) ** 2)
+        clean_energy = [None] * num_crit
+        processed_energy = [None] * num_crit
+
+        # (3) Compute Filterbank output energies (in dB)
+        for i in range(num_crit):
+            clean_energy[i] = np.sum(clean_spec[:n_fftby2] * \
+                                     crit_filter[i, :])
+            processed_energy[i] = np.sum(processed_spec[:n_fftby2] * \
+                                         crit_filter[i, :])
+        clean_energy = np.array(clean_energy).reshape(-1, 1)
+        eps = np.ones((clean_energy.shape[0], 1)) * 1e-10
+        clean_energy = np.concatenate((clean_energy, eps), axis=1)
+        clean_energy = 10 * np.log10(np.max(clean_energy, axis=1))
+        processed_energy = np.array(processed_energy).reshape(-1, 1)
+        processed_energy = np.concatenate((processed_energy, eps), axis=1)
+        processed_energy = 10 * np.log10(np.max(processed_energy, axis=1))
+
+        # (4) Compute Spectral Shape (dB[i+1] - dB[i])
+        clean_slope = clean_energy[1:num_crit] - clean_energy[:num_crit-1]
+        processed_slope = processed_energy[1:num_crit] - \
+                processed_energy[:num_crit-1]
+
+        # (5) Find the nearest peak locations in the spectra to each
+        # critical band. If the slope is negative, we search
+        # to the left. If positive, we search to the right.
+        clean_loc_peak = []
+        processed_loc_peak = []
+        for i in range(num_crit - 1):
+            if clean_slope[i] > 0:
+                # search to the right
+                n = i
+                while n < num_crit - 1 and clean_slope[n] > 0:
+                    n += 1
+                clean_loc_peak.append(clean_energy[n - 1])
+            else:
+                # search to the left
+                n = i
+                while n >= 0 and clean_slope[n] <= 0:
+                    n -= 1
+                clean_loc_peak.append(clean_energy[n + 1])
+            # find the peaks in the processed speech signal
+            if processed_slope[i] > 0:
+                n = i
+                while n < num_crit - 1 and processed_slope[n] > 0:
+                    n += 1
+                processed_loc_peak.append(processed_energy[n - 1])
+            else:
+                n = i
+                while n >= 0 and processed_slope[n] <= 0:
+                    n -= 1
+                processed_loc_peak.append(processed_energy[n + 1])
+
+        # (6) Compuet the WSS Measure for this frame. This includes
+        # determination of the weighting functino
+        dBMax_clean = max(clean_energy)
+        dBMax_processed = max(processed_energy)
+
+        # The weights are calculated by averaging individual
+        # weighting factors from the clean and processed frame.
+        # These weights W_clean and W_processed should range
+        # from 0 to 1 and place more emphasis on spectral 
+        # peaks and less emphasis on slope differences in spectral
+        # valleys.  This procedure is described on page 1280 of
+        # Klatt's 1982 ICASSP paper.
+        clean_loc_peak = np.array(clean_loc_peak)
+        processed_loc_peak = np.array(processed_loc_peak)
+        Wmax_clean = Kmax / (Kmax + dBMax_clean - clean_energy[:num_crit-1])
+        Wlocmax_clean = Klocmax / (Klocmax + clean_loc_peak - \
+                                   clean_energy[:num_crit-1])
+        W_clean = Wmax_clean * Wlocmax_clean
+        Wmax_processed = Kmax / (Kmax + dBMax_processed - \
+                                processed_energy[:num_crit-1])
+        Wlocmax_processed = Klocmax / (Klocmax + processed_loc_peak - \
+                                      processed_energy[:num_crit-1])
+        W_processed = Wmax_processed * Wlocmax_processed
+        W = (W_clean + W_processed) / 2
+        distortion.append(np.sum(W * (clean_slope[:num_crit - 1] - \
+                                     processed_slope[:num_crit - 1]) ** 2))
+
+        # this normalization is not part of Klatt's paper, but helps
+        # to normalize the meaasure. Here we scale the measure by the sum of the
+        # weights
+        distortion[frame_count] = distortion[frame_count] / np.sum(W)
+        start += int(skiprate)
+    return distortion
+
+
+def llr(ref_wav, deg_wav, srate):
+    clean_speech = ref_wav
+    processed_speech = deg_wav
+    clean_length = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+    assert clean_length == processed_length, clean_length
+
+    winlength = round(30 * srate / 1000.) # 240 wlen in samples
+    skiprate = np.floor(winlength / 4)
+    if srate < 10000:
+        # LPC analysis order
+        P = 10
+    else:
+        P = 16
+
+    # For each frame of input speech, calculate the Log Likelihood Ratio
+    num_frames = int(clean_length / skiprate - (winlength / skiprate))
+    start = 0
+    time = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window = 0.5 * (1 - np.cos(2 * np.pi * time))
+    distortion = []
+
+    for frame_count in range(num_frames):
+        # (1) Get the Frames for the test and reference speeech.
+        # Multiply by Hanning window.
+        clean_frame = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Get the autocorrelation logs and LPC params used
+        # to compute the LLR measure
+        R_clean, Ref_clean, A_clean = lpcoeff(clean_frame, P)
+        R_processed, Ref_processed, A_processed = lpcoeff(processed_frame, P)
+        A_clean = A_clean[None, :]
+        A_processed = A_processed[None, :]
+
+        # (3) Compute the LLR measure
+        numerator = A_processed.dot(toeplitz(R_clean)).dot(A_processed.T)
+        denominator = A_clean.dot(toeplitz(R_clean)).dot(A_clean.T)
+
+        if (numerator/denominator) <= 0:
+            print(f'Numerator: {numerator}')
+            print(f'Denominator: {denominator}')
+
+        log_ = np.log(numerator / denominator)
+        distortion.append(np.squeeze(log_))
+        start += int(skiprate)
+    return np.nan_to_num(np.array(distortion))
+# -------------------------------------------------------------------------- #
+
+#!/usr/bin/env python3
+
+# Copyright 2020 Wen-Chin Huang and Tomoki Hayashi
+#  Apache 2.0  (http://www.apache.org/licenses/LICENSE-2.0)
+# ported from https://github.com/espnet/espnet/blob/master/utils/mcd_calculate.py
+
+"""Evaluate MCD between generated and groundtruth audios with SPTK-based mcep."""
+
+from typing import Tuple
+
+import numpy as np
+import pysptk
+from fastdtw import fastdtw
+from scipy import spatial
+
+
+def sptk_extract(
+    x: np.ndarray,
+    fs: int,
+    n_fft: int = 512,
+    n_shift: int = 256,
+    mcep_dim: int = 25,
+    mcep_alpha: float = 0.41,
+    is_padding: bool = False,
+) -> np.ndarray:
+    """Extract SPTK-based mel-cepstrum.
+
+    Args:
+        x (ndarray): 1D waveform array.
+        fs (int): Sampling rate
+        n_fft (int): FFT length in point (default=512).
+        n_shift (int): Shift length in point (default=256).
+        mcep_dim (int): Dimension of mel-cepstrum (default=25).
+        mcep_alpha (float): All pass filter coefficient (default=0.41).
+        is_padding (bool): Whether to pad the end of signal (default=False).
+
+    Returns:
+        ndarray: Mel-cepstrum with the size (N, n_fft).
+
+    """
+    # perform padding
+    if is_padding:
+        n_pad = n_fft - (len(x) - n_fft) % n_shift
+        x = np.pad(x, (0, n_pad), "reflect")
+
+    # get number of frames
+    n_frame = (len(x) - n_fft) // n_shift + 1
+
+    # get window function
+    win = pysptk.sptk.hamming(n_fft)
+
+    # check mcep and alpha
+    if mcep_dim is None or mcep_alpha is None:
+        mcep_dim, mcep_alpha = _get_best_mcep_params(fs)
+
+    # calculate spectrogram
+    mcep = [
+        pysptk.mcep(
+            x[n_shift * i : n_shift * i + n_fft] * win,
+            mcep_dim,
+            mcep_alpha,
+            eps=1e-6,
+            etype=1,
+        )
+        for i in range(n_frame)
+    ]
+
+    return np.stack(mcep)
+
+
+def _get_best_mcep_params(fs: int) -> Tuple[int, float]:
+    # https://sp-nitech.github.io/sptk/latest/main/mgcep.html#_CPPv4N4sptk19MelCepstralAnalysisE
+    if fs == 8000:
+        return 13, 0.31
+    elif fs == 16000:
+        return 23, 0.42
+    elif fs == 22050:
+        return 34, 0.45
+    elif fs == 24000:
+        return 34, 0.46
+    elif fs == 32000:
+        return 36, 0.50
+    elif fs == 44100:
+        return 39, 0.53
+    elif fs == 48000:
+        return 39, 0.55
+    else:
+        raise ValueError(f"Not found the setting for {fs}.")
+
+
+def calculate_mcd(
+    inf_audio,
+    ref_audio,
+    fs,
+    n_fft=1024,
+    n_shift=256,
+    mcep_dim=None,
+    mcep_alpha=None,
+):
+    """Calculate MCD."""
+
+    # extract ground truth and converted features
+    gen_mcep = sptk_extract(
+        x=inf_audio,
+        fs=fs,
+        n_fft=n_fft,
+        n_shift=n_shift,
+        mcep_dim=mcep_dim,
+        mcep_alpha=mcep_alpha,
+    )
+    gt_mcep = sptk_extract(
+        x=ref_audio,
+        fs=fs,
+        n_fft=n_fft,
+        n_shift=n_shift,
+        mcep_dim=mcep_dim,
+        mcep_alpha=mcep_alpha,
+    )
+
+    # DTW
+    _, path = fastdtw(gen_mcep, gt_mcep, dist=spatial.distance.euclidean)
+    twf = np.array(path).T
+    gen_mcep_dtw = gen_mcep[twf[0]]
+    gt_mcep_dtw = gt_mcep[twf[1]]
+
+    # MCD
+    diff2sum = np.sum((gen_mcep_dtw - gt_mcep_dtw) ** 2, 1)
+    mcd = np.mean(10.0 / np.log(10.0) * np.sqrt(2 * diff2sum), 0)
+
+    return mcd

scores/llr.py

@@ -0,0 +1,66 @@
+from basis import ScoreBasis
+import numpy as np
+from scipy.linalg import toeplitz
+from scores.helper import lpcoeff
+
+class LLR(ScoreBasis):
+    def __init__(self):
+        super(LLR, self).__init__(name='LLR')
+        self.intrusive = False
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('LLR needs a reference and a test signals.')
+        return cal_LLR(audios[0], audios[1], score_rate)
+
+def cal_LLR(ref_wav, deg_wav, srate):
+    # obtained from https://github.com/wooseok-shin/MetricGAN-plus-pytorch/blob/main/metric_functions/metric_helper.py
+    clean_speech = ref_wav
+    processed_speech = deg_wav
+    clean_length = ref_wav.shape[0]
+    processed_length = deg_wav.shape[0]
+    assert clean_length == processed_length, clean_length
+
+    winlength = round(30 * srate / 1000.) # 240 wlen in samples
+    skiprate = np.floor(winlength / 4)
+    if srate < 10000:
+        # LPC analysis order
+        P = 10
+    else:
+        P = 16
+
+    # For each frame of input speech, calculate the Log Likelihood Ratio
+    num_frames = int(clean_length / skiprate - (winlength / skiprate))
+    start = 0
+    time = np.linspace(1, winlength, winlength) / (winlength + 1)
+    window = 0.5 * (1 - np.cos(2 * np.pi * time))
+    distortion = []
+
+    for frame_count in range(num_frames):
+        # (1) Get the Frames for the test and reference speeech.
+        # Multiply by Hanning window.
+        clean_frame = clean_speech[start:start+winlength]
+        processed_frame = processed_speech[start:start+winlength]
+        clean_frame = clean_frame * window
+        processed_frame = processed_frame * window
+
+        # (2) Get the autocorrelation logs and LPC params used
+        # to compute the LLR measure
+        R_clean, Ref_clean, A_clean = lpcoeff(clean_frame, P)
+        R_processed, Ref_processed, A_processed = lpcoeff(processed_frame, P)
+        A_clean = A_clean[None, :]
+        A_processed = A_processed[None, :]
+
+        # (3) Compute the LLR measure
+        numerator = A_processed.dot(toeplitz(R_clean)).dot(A_processed.T)
+        denominator = A_clean.dot(toeplitz(R_clean)).dot(A_clean.T)
+
+        if (numerator/denominator) <= 0:
+            print(f'Numerator: {numerator}')
+            print(f'Denominator: {denominator}')
+
+        log_ = np.log(numerator / denominator)
+        distortion.append(np.squeeze(log_))
+        start += int(skiprate)
+    return np.mean(np.nan_to_num(np.array(distortion)))
+

scores/lsd.py

@@ -0,0 +1,30 @@
+from basis import ScoreBasis
+import numpy as np
+import librosa
+
+EPS = 1e-12
+
+class LSD(ScoreBasis):
+    def __init__(self):
+        super(LSD, self).__init__(name='LSD')
+        self.intrusive = False
+        self.mono = True
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('NB_PESQ needs a reference and a test signals.')
+        est = wav_to_spectrogram(audios[1], score_rate)
+        target = wav_to_spectrogram(audios[0], score_rate)
+        return cal_LSD(est, target)
+
+def wav_to_spectrogram(wav, rate):
+    hop_length = int(rate / 100)
+    n_fft = int(2048 / (48000 / rate)) 
+    spec = np.abs(librosa.stft(wav, hop_length=hop_length, n_fft=n_fft))
+    spec = np.transpose(spec, (1, 0))
+    return spec
+
+def cal_LSD(est, target):
+    log_ratio = np.log10(target**2 / ((est + EPS) ** 2) + EPS) ** 2
+    lsd_ = np.mean(np.mean(log_ratio, axis=1) ** 0.5, axis=0)
+    return lsd_

scores/mcd.py

@@ -0,0 +1,136 @@
+from basis import ScoreBasis
+import librosa
+import math
+import numpy as np
+import pyworld
+import pysptk
+from fastdtw import fastdtw
+from scipy.spatial.distance import euclidean
+#from scores.helper import calculate_mcd
+#from pymcd.mcd import Calculate_MCD
+#refer to : https://github.com/chenqi008/pymcd/blob/main/pymcd/mcd.py
+class MCD(ScoreBasis):
+    def __init__(self):
+        super(MCD, self).__init__(name='MCD')
+        self.intrusive = False
+        # three different modes "plain", "dtw" and "dtw_sl" for the above three MCD metrics 
+        self.mcd_toolbox = Calculate_MCD(MCD_mode="plain")
+
+    def windowed_scoring(self, audios, score_rate):
+        if len(audios) != 2:
+            raise ValueError('MCD needs a reference and a test signals.')
+        return self.mcd_toolbox.calculate_mcd(audios[1], audios[0], score_rate)
+
+# ================================================= #
+# calculate the Mel-Cepstral Distortion (MCD) value #
+# ================================================= #
+#refer to : https://github.com/chenqi008/pymcd/blob/main/pymcd/mcd.py
+class Calculate_MCD(object):
+    """docstring for Calculate_MCD"""
+    def __init__(self, MCD_mode):
+        super(Calculate_MCD, self).__init__()
+        self.MCD_mode = MCD_mode
+        #self.SAMPLING_RATE = 22050
+        self.FRAME_PERIOD = 5.0
+        self.log_spec_dB_const = 10.0 / math.log(10.0) * math.sqrt(2.0) # 6.141851463713754
+	
+    def load_wav(self, wav_file, sample_rate):
+        """
+        Load a wav file with librosa.
+        :param wav_file: path to wav file
+        :param sr: sampling rate
+        :return: audio time series numpy array
+        """
+        wav, _ = librosa.load(wav_file, sr=sample_rate, mono=True)
+        return wav
+
+    # distance metric
+    def log_spec_dB_dist(self, x, y):
+        # log_spec_dB_const = 10.0 / math.log(10.0) * math.sqrt(2.0)
+        diff = x - y
+        return self.log_spec_dB_const * math.sqrt(np.inner(diff, diff))
+
+    # calculate distance (metric)
+    # def calculate_mcd_distance(self, x, y, distance, path):
+    def calculate_mcd_distance(self, x, y, path):
+        '''
+        param path: pairs between x and y
+        '''
+        pathx = list(map(lambda l: l[0], path))
+        pathy = list(map(lambda l: l[1], path))
+        x, y = x[pathx], y[pathy]
+        frames_tot = x.shape[0]       # length of pairs
+
+        z = x - y
+        min_cost_tot = np.sqrt((z * z).sum(-1)).sum()
+
+        return frames_tot, min_cost_tot
+
+    # extract acoustic features
+    # alpha = 0.65  # commonly used at 22050 Hz
+    def wav2mcep_numpy(self, loaded_wav, score_rate=22050, alpha=0.65, fft_size=512):
+
+        # Use WORLD vocoder to spectral envelope
+        _, sp, _ = pyworld.wav2world(loaded_wav.astype(np.double), fs=score_rate,
+                                     frame_period=self.FRAME_PERIOD, fft_size=fft_size)
+        # Extract MCEP features
+        mcep = pysptk.sptk.mcep(sp, order=13, alpha=alpha, maxiter=0,
+                                etype=1, eps=1.0E-8, min_det=0.0, itype=3)
+
+        return mcep
+
+    # calculate the Mel-Cepstral Distortion (MCD) value
+    #def average_mcd(self, ref_audio_file, syn_audio_file, cost_function, MCD_mode):
+    def average_mcd(self, loaded_ref_wav, loaded_syn_wav, cost_function, MCD_mode, score_rate):
+        """
+        Calculate the average MCD.
+        :param ref_mcep_files: list of strings, paths to MCEP target reference files
+        :param synth_mcep_files: list of strings, paths to MCEP converted synthesised files
+        :param cost_function: distance metric used
+        :param plain: if plain=True, use Dynamic Time Warping (dtw)
+        :returns: average MCD, total frames processed
+        """
+        # load wav from given wav file
+        #loaded_ref_wav = self.load_wav(ref_audio_file, sample_rate=self.SAMPLING_RATE)
+        #loaded_syn_wav = self.load_wav(syn_audio_file, sample_rate=self.SAMPLING_RATE)
+
+        if MCD_mode == "plain":
+            # pad 0
+            if len(loaded_ref_wav)<len(loaded_syn_wav):
+                loaded_ref_wav = np.pad(loaded_ref_wav, (0, len(loaded_syn_wav)-len(loaded_ref_wav)))
+            else:
+                loaded_syn_wav = np.pad(loaded_syn_wav, (0, len(loaded_ref_wav)-len(loaded_syn_wav)))
+
+            # extract MCEP features (vectors): 2D matrix (num x mcep_size)
+            ref_mcep_vec = self.wav2mcep_numpy(loaded_ref_wav, score_rate)
+            syn_mcep_vec = self.wav2mcep_numpy(loaded_syn_wav, score_rate)
+
+            if MCD_mode == "plain":
+                # print("Calculate plain MCD ...")
+                path = []
+                # for i in range(num_temp):
+                for i in range(len(ref_mcep_vec)):
+                    path.append((i, i))
+            elif MCD_mode == "dtw":
+                # print("Calculate MCD-dtw ...")
+                _, path = fastdtw(ref_mcep_vec[:, 1:], syn_mcep_vec[:, 1:], dist=euclidean)
+            elif MCD_mode == "dtw_sl":
+                # print("Calculate MCD-dtw-sl ...")
+                cof = len(ref_mcep_vec)/len(syn_mcep_vec) if len(ref_mcep_vec)>len(syn_mcep_vec) else len(syn_mcep_vec)/len(ref_mcep_vec)
+                _, path = fastdtw(ref_mcep_vec[:, 1:], syn_mcep_vec[:, 1:], dist=euclidean)
+
+            frames_tot, min_cost_tot = self.calculate_mcd_distance(ref_mcep_vec, syn_mcep_vec, path)
+
+            if MCD_mode == "dtw_sl":
+                mean_mcd = cof * self.log_spec_dB_const * min_cost_tot / frames_tot
+            else:
+                mean_mcd = self.log_spec_dB_const * min_cost_tot / frames_tot
+
+            return mean_mcd
+
+    # calculate mcd
+    def calculate_mcd(self, reference_audio, synthesized_audio, score_rate):
+        # extract acoustic features
+        mean_mcd = self.average_mcd(reference_audio, synthesized_audio, self.log_spec_dB_dist, self.MCD_mode, score_rate)
+
+        return mean_mcd

scores/mosnet/__init__.py

@@ -0,0 +1,21 @@
+def load(window, hop=None):
+    import tensorflow as tf
+    from .model import MOSNet
+    tf.debugging.set_log_device_placement(False)
+    # set memory growth
+    gpus = tf.config.experimental.list_physical_devices('GPU')
+    if gpus:
+        try:
+            # Currently, memory growth needs to be the same across GPUs
+            for gpu in gpus:
+                tf.config.experimental.set_memory_growth(gpu, True)
+
+            logical_gpus = tf.config.experimental.list_logical_devices('GPU')
+            print(len(gpus), "Physical GPUs,",
+                  len(logical_gpus), "Logical GPUs")
+        except RuntimeError as e:
+            # Memory growth must be set before GPUs have been initialized
+            print(e)
+
+    mosnet = MOSNet(window, hop)
+    return mosnet

scores/mosnet/cnn_blstm.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:78b75e7d76ee6074ea7d57dcffa56d0c90be9d3d8dedc2217e25e259423cb756
+size 14248464