Official PyTorch implementation of "t-EER: Parameter-Free Tandem Evaluation Metric of Countermeasures and Biometric Comparators"
import numpy as np
import sys

def compute_Pmiss_Pfa_Pspoof_curves(tar_scores, non_scores, spf_scores):

    # Concatenate all scores and designate arbitrary labels 1=target, 0=nontarget, -1=spoof
    all_scores = np.concatenate((tar_scores, non_scores, spf_scores))
    labels = np.concatenate((np.ones(tar_scores.size), np.zeros(non_scores.size), -1*np.ones(spf_scores.size)))

    # Sort labels based on scores
    indices = np.argsort(all_scores, kind='mergesort')
    labels = labels[indices]

    # Cumulative sums
    tar_sums    = np.cumsum(labels==1)
    non_sums    = np.cumsum(labels==0)
    spoof_sums  = np.cumsum(labels==-1)

    Pmiss       = np.concatenate((np.atleast_1d(0), tar_sums / tar_scores.size))
    Pfa_non     = np.concatenate((np.atleast_1d(1), 1 - (non_sums / non_scores.size)))
    Pfa_spoof   = np.concatenate((np.atleast_1d(1), 1 - (spoof_sums / spf_scores.size)))
    thresholds  = np.concatenate((np.atleast_1d(all_scores[indices[0]] - 0.001), all_scores[indices]))  # Thresholds are the sorted scores

    return Pmiss, Pfa_non, Pfa_spoof, thresholds


def obtain_asv_error_rates(tar_asv, non_asv, spoof_asv, asv_threshold):

    # False alarm and miss rates for ASV
    Pfa_asv = sum(non_asv >= asv_threshold) / non_asv.size
    Pmiss_asv = sum(tar_asv < asv_threshold) / tar_asv.size

    # Rate of rejecting spoofs in ASV
    if spoof_asv.size == 0:
        Pmiss_spoof_asv = None
        Pfa_spoof_asv = None
    else:
        Pmiss_spoof_asv = np.sum(spoof_asv < asv_threshold) / spoof_asv.size
        Pfa_spoof_asv = np.sum(spoof_asv >= asv_threshold) / spoof_asv.size

    return Pfa_asv, Pmiss_asv, Pmiss_spoof_asv, Pfa_spoof_asv


def compute_det_curve(target_scores, nontarget_scores):

    n_scores = target_scores.size + nontarget_scores.size
    all_scores = np.concatenate((target_scores, nontarget_scores))
    labels = np.concatenate((np.ones(target_scores.size), np.zeros(nontarget_scores.size)))

    # Sort labels based on scores
    indices = np.argsort(all_scores, kind='mergesort')
    labels = labels[indices]

    # Compute false rejection and false acceptance rates
    tar_trial_sums = np.cumsum(labels)
    nontarget_trial_sums = nontarget_scores.size - (np.arange(1, n_scores + 1) - tar_trial_sums)

    frr = np.concatenate((np.atleast_1d(0), tar_trial_sums / target_scores.size))  # false rejection rates
    far = np.concatenate((np.atleast_1d(1), nontarget_trial_sums / nontarget_scores.size))  # false acceptance rates
    thresholds = np.concatenate((np.atleast_1d(all_scores[indices[0]] - 0.001), all_scores[indices]))  # Thresholds are the sorted scores

    return frr, far, thresholds


def compute_eer(target_scores, nontarget_scores):
    """ Returns equal error rate (EER) and the corresponding threshold. """
    frr, far, thresholds = compute_det_curve(target_scores, nontarget_scores)
    abs_diffs = np.abs(frr - far)
    min_index = np.argmin(abs_diffs)
    eer = np.mean((frr[min_index], far[min_index]))
    return eer, thresholds[min_index]


def compute_tDCF(bonafide_score_cm, spoof_score_cm, Pfa_asv, Pmiss_asv, Pfa_spoof_asv, cost_model, print_cost):
    """
    Compute Tandem Detection Cost Function (t-DCF) [1] for a fixed ASV system.
    In brief, t-DCF returns a detection cost of a cascaded system of this form,

      Speech waveform -> [CM] -> [ASV] -> decision

    where CM stands for countermeasure and ASV for automatic speaker
    verification. The CM is therefore used as a 'gate' to decided whether or
    not the input speech sample should be passed onwards to the ASV system.
    Generally, both CM and ASV can do detection errors. Not all those errors
    are necessarily equally cost, and not all types of users are necessarily
    equally likely. The tandem t-DCF gives a principled with to compare
    different spoofing countermeasures under a detection cost function
    framework that takes that information into account.

    INPUTS:

      bonafide_score_cm   A vector of POSITIVE CLASS (bona fide or human)
                          detection scores obtained by executing a spoofing
                          countermeasure (CM) on some positive evaluation trials.
                          trial represents a bona fide case.
      spoof_score_cm      A vector of NEGATIVE CLASS (spoofing attack)
                          detection scores obtained by executing a spoofing
                          CM on some negative evaluation trials.
      Pfa_asv             False alarm (false acceptance) rate of the ASV
                          system that is evaluated in tandem with the CM.
                          Assumed to be in fractions, not percentages.
      Pmiss_asv           Miss (false rejection) rate of the ASV system that
                          is evaluated in tandem with the spoofing CM.
                          Assumed to be in fractions, not percentages.
      Pmiss_spoof_asv     Miss rate of spoof samples of the ASV system that
                          is evaluated in tandem with the spoofing CM. That
                          is, the fraction of spoof samples that were
                          rejected by the ASV system.
      cost_model          A struct that contains the parameters of t-DCF,
                          with the following fields.

                          Ptar        Prior probability of target speaker.
                          Pnon        Prior probability of nontarget speaker (zero-effort impostor)
                          Psoof       Prior probability of spoofing attack.
                          Cmiss       Cost of tandem system falsely rejecting target speaker.
                          Cfa         Cost of tandem system falsely accepting nontarget speaker.
                          Cfa_spoof   Cost of tandem system falsely accepting spoof.

      print_cost          Print a summary of the cost parameters and the
                          implied t-DCF cost function?

    OUTPUTS:

      tDCF_norm           Normalized t-DCF curve across the different CM
                          system operating points; see [2] for more details.
                          Normalized t-DCF > 1 indicates a useless
                          countermeasure (as the tandem system would do
                          better without it). min(tDCF_norm) will be the
                          minimum t-DCF used in ASVspoof 2019 [2].
      CM_thresholds       Vector of same size as tDCF_norm corresponding to
                          the CM threshold (operating point).

    NOTE:
    o     In relative terms, higher detection scores values are assumed to
          indicate stronger support for the bona fide hypothesis.
    o     You should provide real-valued soft scores, NOT hard decisions. The
          recommendation is that the scores are log-likelihood ratios (LLRs)
          from a bonafide-vs-spoof hypothesis based on some statistical model.
          This, however, is NOT required. The scores can have arbitrary range
          and scaling.
    o     Pfa_asv, Pmiss_asv, Pmiss_spoof_asv are in fractions, not percentages.

    References:

      [1] T. Kinnunen, H. Delgado, N. Evans,K.-A. Lee, V. Vestman,
          A. Nautsch, M. Todisco, X. Wang, M. Sahidullah, J. Yamagishi,
          and D.-A. Reynolds, "Tandem Assessment of Spoofing Countermeasures
          and Automatic Speaker Verification: Fundamentals," IEEE/ACM Transaction on
          Audio, Speech and Language Processing (TASLP).

      [2] ASVspoof 2019 challenge evaluation plan
          https://www.asvspoof.org/asvspoof2019/asvspoof2019_evaluation_plan.pdf
    """


    # Sanity check of cost parameters
    if cost_model['Cfa'] < 0 or cost_model['Cmiss'] < 0 or \
            cost_model['Cfa'] < 0 or cost_model['Cmiss'] < 0:
        print('WARNING: Usually the cost values should be positive!')

    if cost_model['Ptar'] < 0 or cost_model['Pnon'] < 0 or cost_model['Pspoof'] < 0 or \
            np.abs(cost_model['Ptar'] + cost_model['Pnon'] + cost_model['Pspoof'] - 1) > 1e-10:
        sys.exit('ERROR: Your prior probabilities should be positive and sum up to one.')

    # Unless we evaluate worst-case model, we need to have some spoof tests against asv
    if Pfa_spoof_asv is None:
        sys.exit('ERROR: you should provide false alarm rate of spoof tests against your ASV system.')

    # Sanity check of scores
    combined_scores = np.concatenate((bonafide_score_cm, spoof_score_cm))
    if np.isnan(combined_scores).any() or np.isinf(combined_scores).any():
        sys.exit('ERROR: Your scores contain nan or inf.')

    # Sanity check that inputs are scores and not decisions
    n_uniq = np.unique(combined_scores).size
    if n_uniq < 3:
        sys.exit('ERROR: You should provide soft CM scores - not binary decisions')

    # Obtain miss and false alarm rates of CM
    Pmiss_cm, Pfa_cm, CM_thresholds = compute_det_curve(bonafide_score_cm, spoof_score_cm)

    # Constants - see ASVspoof 2019 evaluation plan

    C0 = cost_model['Ptar'] * cost_model['Cmiss'] * Pmiss_asv + cost_model['Pnon']*cost_model['Cfa']*Pfa_asv
    C1 = cost_model['Ptar'] * cost_model['Cmiss'] - (cost_model['Ptar'] * cost_model['Cmiss'] * Pmiss_asv + cost_model['Pnon'] * cost_model['Cfa'] * Pfa_asv)
    C2 = cost_model['Pspoof'] * cost_model['Cfa_spoof'] * Pfa_spoof_asv;


    # Sanity check of the weights
    if C0 < 0 or C1 < 0 or C2 < 0:
        sys.exit('You should never see this error but I cannot evalute tDCF with negative weights - please check whether your ASV error rates are correctly computed?')

    # Obtain t-DCF curve for all thresholds
    tDCF = C0 + C1 * Pmiss_cm + C2 * Pfa_cm

    # Obtain default t-DCF
    tDCF_default = C0 + np.minimum(C1, C2)

    # Normalized t-DCF
    tDCF_norm = tDCF / tDCF_default

    # Everything should be fine if reaching here.
    if print_cost:

        print('t-DCF evaluation from [Nbona={}, Nspoof={}] trials\n'.format(bonafide_score_cm.size, spoof_score_cm.size))
        print('t-DCF MODEL')
        print('   Ptar         = {:8.5f} (Prior probability of target user)'.format(cost_model['Ptar']))
        print('   Pnon         = {:8.5f} (Prior probability of nontarget user)'.format(cost_model['Pnon']))
        print('   Pspoof       = {:8.5f} (Prior probability of spoofing attack)'.format(cost_model['Pspoof']))
        print('   Cfa          = {:8.5f} (Cost of tandem system falsely accepting a nontarget)'.format(cost_model['Cfa']))
        print('   Cmiss        = {:8.5f} (Cost of tandem system falsely rejecting target speaker)'.format(cost_model['Cmiss']))
        print('   Cfa_spoof    = {:8.5f} (Cost of tandem sysmte falsely accepting spoof)'.format(cost_model['Cfa_spoof']))
        print('\n   Implied normalized t-DCF function (depends on t-DCF parameters and ASV errors), t_CM=CM threshold)')
        print('   tDCF_norm(t_CM) = {:8.5f} + {:8.5f} x Pmiss_cm(t_CM) + {:8.5f} x Pfa_cm(t_CM)\n'.format(C0/tDCF_default, C1/tDCF_default, C2/tDCF_default))
        print('     * The optimum value is given by the first term (0.06273). This is the normalized t-DCF obtained with an error-free CM system.')
        print('     * The minimum normalized cost (minimum over all possible thresholds) is always <= 1.00.')
        print('')

    return tDCF_norm, CM_thresholds

def compute_tDCF_legacy(bonafide_score_cm, spoof_score_cm, Pfa_asv, Pmiss_asv, Pmiss_spoof_asv, cost_model, print_cost):
    """
    Compute Tandem Detection Cost Function (t-DCF) [1] for a fixed ASV system.
    In brief, t-DCF returns a detection cost of a cascaded system of this form,

      Speech waveform -> [CM] -> [ASV] -> decision

    where CM stands for countermeasure and ASV for automatic speaker
    verification. The CM is therefore used as a 'gate' to decided whether or
    not the input speech sample should be passed onwards to the ASV system.
    Generally, both CM and ASV can do detection errors. Not all those errors
    are necessarily equally cost, and not all types of users are necessarily
    equally likely. The tandem t-DCF gives a principled with to compare
    different spoofing countermeasures under a detection cost function
    framework that takes that information into account.

    INPUTS:

      bonafide_score_cm   A vector of POSITIVE CLASS (bona fide or human)
                          detection scores obtained by executing a spoofing
                          countermeasure (CM) on some positive evaluation trials.
                          trial represents a bona fide case.
      spoof_score_cm      A vector of NEGATIVE CLASS (spoofing attack)
                          detection scores obtained by executing a spoofing
                          CM on some negative evaluation trials.
      Pfa_asv             False alarm (false acceptance) rate of the ASV
                          system that is evaluated in tandem with the CM.
                          Assumed to be in fractions, not percentages.
      Pmiss_asv           Miss (false rejection) rate of the ASV system that
                          is evaluated in tandem with the spoofing CM.
                          Assumed to be in fractions, not percentages.
      Pmiss_spoof_asv     Miss rate of spoof samples of the ASV system that
                          is evaluated in tandem with the spoofing CM. That
                          is, the fraction of spoof samples that were
                          rejected by the ASV system.
      cost_model          A struct that contains the parameters of t-DCF,
                          with the following fields.

                          Ptar        Prior probability of target speaker.
                          Pnon        Prior probability of nontarget speaker (zero-effort impostor)
                          Psoof       Prior probability of spoofing attack.
                          Cmiss_asv   Cost of ASV falsely rejecting target.
                          Cfa_asv     Cost of ASV falsely accepting nontarget.
                          Cmiss_cm    Cost of CM falsely rejecting target.
                          Cfa_cm      Cost of CM falsely accepting spoof.

      print_cost          Print a summary of the cost parameters and the
                          implied t-DCF cost function?

    OUTPUTS:

      tDCF_norm           Normalized t-DCF curve across the different CM
                          system operating points; see [2] for more details.
                          Normalized t-DCF > 1 indicates a useless
                          countermeasure (as the tandem system would do
                          better without it). min(tDCF_norm) will be the
                          minimum t-DCF used in ASVspoof 2019 [2].
      CM_thresholds       Vector of same size as tDCF_norm corresponding to
                          the CM threshold (operating point).

    NOTE:
    o     In relative terms, higher detection scores values are assumed to
          indicate stronger support for the bona fide hypothesis.
    o     You should provide real-valued soft scores, NOT hard decisions. The
          recommendation is that the scores are log-likelihood ratios (LLRs)
          from a bonafide-vs-spoof hypothesis based on some statistical model.
          This, however, is NOT required. The scores can have arbitrary range
          and scaling.
    o     Pfa_asv, Pmiss_asv, Pmiss_spoof_asv are in fractions, not percentages.

    References:

      [1] T. Kinnunen, K.-A. Lee, H. Delgado, N. Evans, M. Todisco,
          M. Sahidullah, J. Yamagishi, D.A. Reynolds: "t-DCF: a Detection
          Cost Function for the Tandem Assessment of Spoofing Countermeasures
          and Automatic Speaker Verification", Proc. Odyssey 2018: the
          Speaker and Language Recognition Workshop, pp. 312--319, Les Sables d'Olonne,
          France, June 2018 (https://www.isca-speech.org/archive/Odyssey_2018/pdfs/68.pdf)

      [2] ASVspoof 2019 challenge evaluation plan
          https://www.asvspoof.org/asvspoof2019/asvspoof2019_evaluation_plan.pdf
    """


    # Sanity check of cost parameters
    if cost_model['Cfa_asv'] < 0 or cost_model['Cmiss_asv'] < 0 or \
            cost_model['Cfa_cm'] < 0 or cost_model['Cmiss_cm'] < 0:
        print('WARNING: Usually the cost values should be positive!')

    if cost_model['Ptar'] < 0 or cost_model['Pnon'] < 0 or cost_model['Pspoof'] < 0 or \
            np.abs(cost_model['Ptar'] + cost_model['Pnon'] + cost_model['Pspoof'] - 1) > 1e-10:
        sys.exit('ERROR: Your prior probabilities should be positive and sum up to one.')

    # Unless we evaluate worst-case model, we need to have some spoof tests against asv
    if Pmiss_spoof_asv is None:
        sys.exit('ERROR: you should provide miss rate of spoof tests against your ASV system.')

    # Sanity check of scores
    combined_scores = np.concatenate((bonafide_score_cm, spoof_score_cm))
    if np.isnan(combined_scores).any() or np.isinf(combined_scores).any():
        sys.exit('ERROR: Your scores contain nan or inf.')

    # Sanity check that inputs are scores and not decisions
    n_uniq = np.unique(combined_scores).size
    if n_uniq < 3:
        sys.exit('ERROR: You should provide soft CM scores - not binary decisions')

    # Obtain miss and false alarm rates of CM
    Pmiss_cm, Pfa_cm, CM_thresholds = compute_det_curve(bonafide_score_cm, spoof_score_cm)

    # Constants - see ASVspoof 2019 evaluation plan
    C1 = cost_model['Ptar'] * (cost_model['Cmiss_cm'] - cost_model['Cmiss_asv'] * Pmiss_asv) - \
         cost_model['Pnon'] * cost_model['Cfa_asv'] * Pfa_asv
    C2 = cost_model['Cfa_cm'] * cost_model['Pspoof'] * (1 - Pmiss_spoof_asv)

    # Sanity check of the weights
    if C1 < 0 or C2 < 0:
        sys.exit('You should never see this error but I cannot evalute tDCF with negative weights - please check whether your ASV error rates are correctly computed?')

    # Obtain t-DCF curve for all thresholds
    tDCF = C1 * Pmiss_cm + C2 * Pfa_cm

    # Normalized t-DCF
    tDCF_norm = tDCF / np.minimum(C1, C2)

    # Everything should be fine if reaching here.
    if print_cost:

        print('t-DCF evaluation from [Nbona={}, Nspoof={}] trials\n'.format(bonafide_score_cm.size, spoof_score_cm.size))
        print('t-DCF MODEL')
        print('   Ptar         = {:8.5f} (Prior probability of target user)'.format(cost_model['Ptar']))
        print('   Pnon         = {:8.5f} (Prior probability of nontarget user)'.format(cost_model['Pnon']))
        print('   Pspoof       = {:8.5f} (Prior probability of spoofing attack)'.format(cost_model['Pspoof']))
        print('   Cfa_asv      = {:8.5f} (Cost of ASV falsely accepting a nontarget)'.format(cost_model['Cfa_asv']))
        print('   Cmiss_asv    = {:8.5f} (Cost of ASV falsely rejecting target speaker)'.format(cost_model['Cmiss_asv']))
        print('   Cfa_cm       = {:8.5f} (Cost of CM falsely passing a spoof to ASV system)'.format(cost_model['Cfa_cm']))
        print('   Cmiss_cm     = {:8.5f} (Cost of CM falsely blocking target utterance which never reaches ASV)'.format(cost_model['Cmiss_cm']))
        print('\n   Implied normalized t-DCF function (depends on t-DCF parameters and ASV errors), s=CM threshold)')

        if C2 == np.minimum(C1, C2):
            print('   tDCF_norm(s) = {:8.5f} x Pmiss_cm(s) + Pfa_cm(s)\n'.format(C1 / C2))
        else:
            print('   tDCF_norm(s) = Pmiss_cm(s) + {:8.5f} x Pfa_cm(s)\n'.format(C2 / C1))

    return tDCF_norm, CM_thresholds


def performance(bona_cm,spoof_cm, Pfa_asv, Pmiss_asv, Pfa_spoof_asv, invert=False):

    Pspoof = 0.05
    cost_model = {
        'Pspoof': Pspoof,  # Prior probability of a spoofing attack
        'Ptar': (1 - Pspoof) * 0.99,  # Prior probability of target speaker
        'Pnon': (1 - Pspoof) * 0.01,  # Prior probability of nontarget speaker
        'Cmiss': 1,  # Cost of tandem system falsely rejecting target speaker
        'Cfa': 10,  # Cost of tandem system falsely accepting nontarget speaker
        'Cfa_spoof': 10,  # Cost of tandem system falsely accepting spoof
    }

    if invert==False:
        eer_cm = compute_eer(bona_cm, spoof_cm)[0]
    else:
        eer_cm = compute_eer(-bona_cm, -spoof_cm)[0]

    if invert==False:
        tDCF_curve, _ = compute_tDCF(bona_cm, spoof_cm, Pfa_asv, Pmiss_asv, Pfa_spoof_asv, cost_model, False)
    else:
        tDCF_curve, _ = compute_tDCF(-bona_cm, -spoof_cm, Pfa_asv, Pmiss_asv, Pfa_spoof_asv, cost_model, False)

    min_tDCF_index = np.argmin(tDCF_curve)
    min_tDCF = tDCF_curve[min_tDCF_index]

    return min_tDCF, eer_cm

eval_metrics.py
TakHemlata/T-EER
import numpy as np
import scipy.stats as stats
import matplotlib.pyplot as plt
from scipy import special
import random
import sys
import os
import eval_metrics as em
import loading_function as lf
import simulated_function as sim

args = sys.argv

SASV = args[1].lower() == 'true'
ASVspoof = args[2].lower() == 'true'

if SASV:
    # Replace CM scores with your own scores.
    cm_score_file = 'scores\SASV\SASV_CM_AASIST_scores.txt'
    # Replace ASV with your own scores.
    asv_score_file = 'scores\SASV\SASV_ASV_ECAPA_scores.txt'

elif ASVspoof:
    # Replace CM scores with your own scores.
    cm_score_file = 'scores\ASVspoof\CM_RawNet2_2021_scores_file.txt'
    # Replace ASV with your own scores.
    asv_score_file = 'scores\ASVspoof\ASV_X_vector_2021_baseline_scores.txt'

else:
    print('Using synthetic scores')

# load scores file
if SASV:
    X_tar, X_non, X_spf = lf.process_sasv_score_files(cm_score_file, asv_score_file)
    
elif ASVspoof:
    cond = 'eval'
    X_tar, X_non, X_spf = lf.process_ASVspoof2021_score_files(cm_score_file, asv_score_file,cond)

else:
    print('Using synthetic scores')

    np.random.seed(seed=233423)

    ## SIMULATOR PARAMETERS
    sim_param = {}
    sim_param['asv_eer_tar_non']    = 0.08
    sim_param['asv_eer_tar_spoof']  = 0.35
    sim_param['cm_eer_tar_spoof']   = 0.10
    sim_param['cm_eer_non_spoof']   = sim_param['cm_eer_tar_spoof']
    sim_param['asv_cm_xcorr_tar']   = 0.00
    sim_param['asv_cm_xcorr_non']   = 0.00
    sim_param['asv_cm_xcorr_spf']   = 0.00

    ## SAMPLE SOME SCORES FROM THE THREE CLASSES
    n_trials_per_class = 10000
    tar_mu, non_mu, spf_mu, tar_cov, non_cov, spf_cov = sim.all_bivariate_gauss_params_from_sim_params(sim_param)
    X_tar = np.random.multivariate_normal(tar_mu, tar_cov, size=n_trials_per_class, check_valid='warn', tol=1e-8)
    X_non = np.random.multivariate_normal(non_mu, non_cov, size=n_trials_per_class, check_valid='warn', tol=1e-8)
    X_spf = np.random.multivariate_normal(spf_mu, spf_cov, size=n_trials_per_class, check_valid='warn', tol=1e-8)


# Obtain t-EER path and t-EER values

plt.close('all')
fig, ax = plt.subplots(1,2, figsize=(16, 8))

if SASV:
    xmin = min([min(X_tar[:,0]), min(X_non[:,0]), min(X_spf[:,0])])  # ASV scores
    xmax = max([max(X_tar[:,0]), max(X_non[:,0]), max(X_spf[:,0])])
    ymin = min([min(X_tar[:,1]), min(X_non[:,1]), min(X_spf[:,1])])  # CM scores
    ymax = max([max(X_tar[:,1]), max(X_non[:,1]), max(X_spf[:,1])])

    ax[0].scatter(X_tar[:,0], X_tar[:,1], c='green', label='target', marker='x', zorder=1, alpha=.5, s=50)
    ax[0].scatter(X_non[:,0], X_non[:,1], c='red', label='non-target', marker='o', zorder=1, alpha=.5, s=50)
    ax[0].scatter(X_spf[:,0], X_spf[:,1], c='black', label='spoof', marker='d', zorder=1, alpha=.5, s=50)

    ax[0].set_xlabel(r'$\mathcal{S}_\mathrm{asv}$',fontsize=34)
    ax[0].set_ylabel(r'$\mathcal{S}_\mathrm{cm}$',fontsize=34)
    ax[0].legend(loc=3,edgecolor='black',fontsize=16);

    # Obtain ASV error curves and ASV thresholds
    Pmiss_ASV, Pfa_non_ASV, Pfa_spf_ASV, tau_ASV = em.compute_Pmiss_Pfa_Pspoof_curves(X_tar[:,0], X_non[:,0], X_spf[:,0])

    # Obtain CM error curves and CM thresholds. Here we use both target and nontarget scores on the bonafide side.
    Pmiss_CM, Pfa_CM, tau_CM = em.compute_det_curve(np.concatenate([X_tar[:,1], X_non[:,1]]), X_spf[:,1])

elif ASVspoof:
    xmin = min([min(X_tar[0]), min(X_non[0]), min(X_spf[0])])  # ASV scores
    xmax = max([max(X_tar[0]), max(X_non[0]), max(X_spf[0])])
    ymin = min([min(X_tar[1]),  min(X_spf[1])])                # CM scores
    ymax = max([max(X_tar[1]),  max(X_spf[1])])

    # Obtain ASV error curves and ASV thresholds
    Pmiss_ASV, Pfa_non_ASV, Pfa_spf_ASV, tau_ASV = em.compute_Pmiss_Pfa_Pspoof_curves(X_tar[0], X_non[0], X_spf[0])

    # Obtain CM error curves and CM thresholds.
    Pmiss_CM, Pfa_CM, tau_CM = em.compute_det_curve(X_tar[1], X_spf[1])

else:
    xmin = min([min(X_tar[:,0]), min(X_non[:,0]), min(X_spf[:,0])])  # ASV scores
    xmax = max([max(X_tar[:,0]), max(X_non[:,0]), max(X_spf[:,0])])
    ymin = min([min(X_tar[:,1]), min(X_non[:,1]), min(X_spf[:,1])])  # CM scores
    ymax = max([max(X_tar[:,1]), max(X_non[:,1]), max(X_spf[:,1])])

    ax[0].scatter(X_tar[:,0], X_tar[:,1], c='green', label='target', marker='x', zorder=1, alpha=.5, s=50)
    ax[0].scatter(X_non[:,0], X_non[:,1], c='red', label='non-target', marker='o', zorder=1, alpha=.5, s=50)
    ax[0].scatter(X_spf[:,0], X_spf[:,1], c='black', label='spoof', marker='d', zorder=1, alpha=.5, s=50)

    ax[0].set_xlabel(r'$\mathcal{S}_\mathrm{asv}$',fontsize=34)
    ax[0].set_ylabel(r'$\mathcal{S}_\mathrm{cm}$',fontsize=34)
    ax[0].legend(loc=3,edgecolor='black',fontsize=16);

    # Obtain ASV error curves and ASV thresholds for synthetic_scores
    Pmiss_ASV, Pfa_non_ASV, Pfa_spf_ASV, tau_ASV = em.compute_Pmiss_Pfa_Pspoof_curves(X_tar[:,0], X_non[:,0], X_spf[:,0])

    # Obtain CM error curves and CM thresholds.
    Pmiss_CM, Pfa_CM, tau_CM = em.compute_det_curve(np.concatenate([X_tar[:,1], X_non[:,1]]), X_spf[:,1])


# Different spoofing prevalence priors (rho) parameters values
rho_vals            = [0,0.2,0.5,0.8,1]
rho_lw              = [0.5,1.3, 1.8,2.2, 2.7]
rho_min_pt_size     = [9,9,9,9,9]
rho_marker          = ['o','o','o','o','o']
ymins               = np.empty(len(rho_vals))

tEER_val    = np.empty([len(rho_vals),len(tau_ASV)], dtype=float)

for rho_idx, rho_spf in enumerate(rho_vals):

    # Table to store the CM threshold index, per each of the ASV operating points
    tEER_idx_CM = np.empty(len(tau_ASV), dtype=int)

    tEER_path   = np.empty([len(rho_vals),len(tau_ASV),2], dtype=float)

    # Tables to store the t-EER, total Pfa and total miss valuees along the t-EER path
    Pmiss_total = np.empty(len(tau_ASV), dtype=float)
    Pfa_total   = np.empty(len(tau_ASV), dtype=float)
    min_tEER    = np.inf
    argmin_tEER = np.empty(2)

    # best intersection point
    xpoint_crit_best = np.inf
    xpoint = np.empty(2)

    # Loop over all possible ASV thresholds
    for tau_ASV_idx, tau_ASV_val in enumerate(tau_ASV):

        # Tandem miss and fa rates as defined in the manuscript
        Pmiss_tdm = Pmiss_CM + (1 - Pmiss_CM) * Pmiss_ASV[tau_ASV_idx]
        Pfa_tdm   = (1 - rho_spf) * (1 - Pmiss_CM) * Pfa_non_ASV[tau_ASV_idx] + rho_spf * Pfa_CM * Pfa_spf_ASV[tau_ASV_idx]

        # Store only the INDEX of the CM threshold (for the current ASV threshold)
        h = Pmiss_tdm - Pfa_tdm
        tmp = np.argmin(abs(h))
        tEER_idx_CM[tau_ASV_idx] = tmp

        if Pmiss_ASV[tau_ASV_idx] < (1 - rho_spf) * Pfa_non_ASV[tau_ASV_idx] + rho_spf * Pfa_spf_ASV[tau_ASV_idx]:
            Pmiss_total[tau_ASV_idx] = Pmiss_tdm[tmp]
            Pfa_total[tau_ASV_idx] = Pfa_tdm[tmp]

            tEER_val[rho_idx,tau_ASV_idx] = np.mean([Pfa_total[tau_ASV_idx], Pmiss_total[tau_ASV_idx]])

            tEER_path[rho_idx,tau_ASV_idx, 0] = tau_ASV_val
            tEER_path[rho_idx,tau_ASV_idx, 1] = tau_CM[tmp]

            if tEER_val[rho_idx,tau_ASV_idx] < min_tEER:
                min_tEER = tEER_val[rho_idx,tau_ASV_idx]
                argmin_tEER[0] = tau_ASV_val
                argmin_tEER[1] = tau_CM[tmp]

            # Check how close we are to the INTERSECTION POINT for different prior (rho) values:
            LHS = Pfa_non_ASV[tau_ASV_idx]/Pfa_spf_ASV[tau_ASV_idx]
            RHS = Pfa_CM[tmp]/(1 - Pmiss_CM[tmp])
            crit = abs(LHS - RHS)

            if crit < xpoint_crit_best:
                xpoint_crit_best = crit
                xpoint[0] = tau_ASV_val
                xpoint[1] = tau_CM[tmp]
                xpoint_tEER = Pfa_spf_ASV[tau_ASV_idx]*Pfa_CM[tmp]
        else:
            # Not in allowed region
            tEER_path[rho_idx,tau_ASV_idx, 0] = np.nan
            tEER_path[rho_idx,tau_ASV_idx, 1] = np.nan
            Pmiss_total[tau_ASV_idx] = np.nan
            Pfa_total[tau_ASV_idx] = np.nan
            tEER_val[rho_idx,tau_ASV_idx] = np.nan


    print("concurrent-teer for [rho :{}] = {:.2f} ".format(rho_spf,xpoint_tEER*100))

    ##======================= t-EER plots =========================##
    ax[0].plot(tEER_path[rho_idx,:,0], tEER_path[rho_idx,:,1], c='blue', lw=rho_lw[rho_idx])
    ax[0].plot(xpoint[0], xpoint[1], marker=rho_marker[rho_idx], markersize=rho_min_pt_size[rho_idx], color='magenta', fillstyle='full', markeredgecolor='yellow')

    ax[0].tick_params(axis='x', labelsize=24)
    ax[0].tick_params(axis='y', labelsize=24)

    ax[1].plot(tau_ASV, tEER_val[rho_idx,:], c='blue', lw=rho_lw[rho_idx],label=r'$\rho={}$'.format(rho_spf), fillstyle='full')
    ax[1].plot(xpoint[0], xpoint_tEER, marker=rho_marker[rho_idx], markersize=rho_min_pt_size[rho_idx], color='magenta', fillstyle='full', markeredgecolor='yellow')

    ax[1].set_xlabel(r'$\tau_\mathrm{asv}$',fontsize=34)
    ax[1].set_ylabel(r't-EER',fontsize=24)
    ax[1].yaxis.set_ticks(np.arange(0.0, 0.7, 0.1))
    ax[1].tick_params(axis='x', labelsize=24)
    ax[1].tick_params(axis='y', labelsize=24)


ax[1].legend(loc=1,edgecolor='black',fontsize=20);
fig.savefig("Teer_plot.pdf", bbox_inches='tight', dpi=600)

evaluate_tEER.py
TakHemlata/T-EER
import numpy as np
import sys

def process_sasv_score_files(cm_score_file, asv_score_file):
    cm_data = np.genfromtxt(cm_score_file, dtype=str)
    cm_utt_id = cm_data[:,1]
    cm_trial_type = cm_data[:, 3]
    cm_scores = cm_data[:,4].astype(float)

    asv_data = np.genfromtxt(asv_score_file, dtype=str)
    asv_utt_id = asv_data[:,1]
    asv_trial_type = asv_data[:, 3]
    asv_scores = asv_data[:,4].astype(float)
    
    # need to check here that the utt_ids are the same in cm and asv files
    X_tar = np.array([asv_scores[asv_trial_type == 'target'], cm_scores[cm_trial_type == 'target']]).T
    X_non = np.array([asv_scores[asv_trial_type == 'nontarget'], cm_scores[cm_trial_type == 'nontarget']]).T
    X_spf = np.array([asv_scores[asv_trial_type == 'spoof'], cm_scores[cm_trial_type == 'spoof']]).T

    return X_tar, X_non, X_spf

def process_ASVspoof2021_score_files(cm_score_file, asv_score_file,cond):
    cm_data = np.genfromtxt(cm_score_file, dtype=str)
    cm_utt_id = cm_data[:,1]
    cm_trial_type = cm_data[:, 5]
    cm_cond = cm_data[:,7]
    cm_scores = cm_data[:,8].astype(float)
    cm_scores = cm_scores[cm_cond == cond]
    cm_trial_type = cm_trial_type[cm_cond == cond]

    asv_data = np.genfromtxt(asv_score_file, dtype=str)
    asv_utt_id = asv_data[:,1]
    asv_trial_type = asv_data[:, 5]
    asv_cond = asv_data[:,7]
    asv_scores = asv_data[:,8].astype(float)
    asv_trial_type = asv_trial_type[asv_cond == cond]
    asv_scores = asv_scores[asv_cond == cond]

    X_tar = np.array([asv_scores[asv_trial_type == 'target'], cm_scores[cm_trial_type == 'bonafide']],dtype=object).T
    X_non = np.array([asv_scores[asv_trial_type == 'nontarget'], cm_scores[cm_trial_type == 'nontarget']],dtype=object).T
    X_spf = np.array([asv_scores[asv_trial_type == 'spoof'], cm_scores[cm_trial_type == 'spoof']],dtype=object).T

    return X_tar, X_non, X_spf
loading_function.py
TakHemlata/T-EER
import numpy as np
import sys

def gauss_params_from_eer(desired_eer):

    m = 2*(stats.norm.ppf(desired_eer)**2)
    v = 2*m

    m_pos = m
    m_neg = -m
    v_pos = v
    v_neg = v

    return m_pos, m_neg, v_pos, v_neg

# -----------------------------------------------------------------------------
def set_bivariate_gauss_params(m1, m2, v1, v2, r):
# m1,m2 - means, v1,v2 - vars, r = correlation

    m       = np.array([m1, m2])
    x_term  = r * np.sqrt(v1) * np.sqrt(v2)
    C       = np.matrix([[v1, x_term], [x_term, v2]])
    return m, C

# -----------------------------------------------------------------------------
def all_bivariate_gauss_params_from_sim_params(sim_param):

    # First, "lock" the target-nontarget gaussians (ASV) and target-spoof Gaussians (CM)
    asv_tar_m, asv_non_m,  asv_tar_v, asv_non_v = gauss_params_from_eer(sim_param['asv_eer_tar_non'])
    cm_tar_m,  cm_spf_m,   cm_tar_v,  cm_spf_v  = gauss_params_from_eer(sim_param['cm_eer_tar_spoof'])

    # ASV - insert spoof distribution with the same shared variance, and with desired EER between (target, spoof)
    F = np.sqrt(2) * special.erfinv(1 - 2*sim_param['asv_eer_tar_spoof'])
    asv_spf_m = asv_tar_m - 2*np.sqrt(asv_tar_v)*F
    asv_spf_v = asv_tar_v

    # CM - similarly, insert nontarget distribution. Note here we've defined the discrimation from spoof (rather than target)
    F = np.sqrt(2) * special.erfinv(1 - 2*sim_param['cm_eer_non_spoof'])
    cm_non_m = cm_spf_m + 2*np.sqrt(cm_tar_v)*F
    cm_non_v = cm_tar_v

    # Now we have all the building blocks!
    tar_mu, tar_cov = set_bivariate_gauss_params(asv_tar_m, cm_tar_m, asv_tar_v, cm_tar_v, sim_param['asv_cm_xcorr_tar'])
    non_mu, non_cov = set_bivariate_gauss_params(asv_non_m, cm_non_m, asv_non_v, cm_non_v, sim_param['asv_cm_xcorr_non'])
    spf_mu, spf_cov = set_bivariate_gauss_params(asv_spf_m, cm_spf_m, asv_spf_v, cm_spf_v, sim_param['asv_cm_xcorr_spf'])

    return tar_mu, non_mu, spf_mu, tar_cov, non_cov, spf_cov
simulated_function.py
TakHemlata/T-EER