Metrics and alignment

kudio.core.evaluator

Speech-enhancement quality metrics: PESQ, STOI, SDR, and the SNR family.

The SNR family (snr, si_sdr, segmental_snr, sdi) is pure numpy. The rest are optional dependencies:

pip install kudio[eval]   # mir_eval + pystoi: SDR, STOI
pip install kudio[pesq]   # pesq: PESQ -- built from source, needs a C compiler

Which PESQ. ITU-T P.862 scores narrowband speech and P.862.2 wideband, and each can be reported raw or mapped to MOS-LQO. They are different numbers for one file: a raw narrowband 2.0 is a MOS-LQO of about 1.6. Papers that just say "PESQ" do not agree on which they mean -- some quote the raw narrowband score the ITU reference binary prints, and much VoiceBank-DEMAND work reports wideband MOS-LQO from the pesq package -- so read how a paper scored before comparing with it. pesq takes the bandwidth and the scale as arguments rather than picking one silently.

Classes:

Name	Description
AudioEvaluate	Evaluate enhanced audio against clean references.

Functions:

Name	Description
snr	Signal-to-noise ratio (dB) of deg relative to ref (pure numpy).
si_sdr	Scale-invariant SDR (dB) — immune to overall gain differences.
segmental_snr	Frame-averaged (segmental) SNR in dB, clipped to [-10, 35] per frame.
sdi	Speech distortion index: sum((ref - deg)**2) / sum(ref**2).
pesq	PESQ of deg against ref (ITU-T P.862 / P.862.2).
score	Score deg against ref: {metric: value}.
eval_metrics	Score a single (reference, degraded) pair with every available metric.

Attributes:

Name	Type	Description
MAX_LENGTH_MISMATCH		How much of the longer signal a comparison may leave unmeasured, as a
METRICS		the names score takes. The first four need nothing installed.

MAX_LENGTH_MISMATCH module-attribute

MAX_LENGTH_MISMATCH = 0.5

How much of the longer signal a comparison may leave unmeasured, as a fraction of it. Every metric here lines the pair up at sample zero and cuts both to the common length, so the tail of the longer one is never looked at; past half of it the score describes the overlap and not the file whose name it is reported under. Half is also well clear of the slop a real pipeline produces -- a window or two lost at the tail, a resampler's rounding -- which is the case truncation is there to absorb in the first place. Raise it to compare a pair that really is that lopsided.

METRICS module-attribute

METRICS = (
    "snr",
    "si_sdr",
    "segsnr",
    "sdi",
    "pesq",
    "stoi",
    "estoi",
    "sdr",
)

the names score takes. The first four need nothing installed.

AudioEvaluate

AudioEvaluate(
    clean_list: Sequence,
    noisy_list: Sequence,
    *,
    pesq_mode: str = "auto",
    pesq_scale: str = "lqo",
)

Evaluate enhanced audio against clean references.

>>> audio_eval = AudioEvaluate(clean_list, noisy_list)
>>> audio_eval.eval('ddae', enhanced_list)
>>> df = audio_eval.get_df()       # pandas.DataFrame, with a Baseline row
>>> per = audio_eval.per_file()    # one row per method x file

File lists must be aligned: clean_list[i] / noisy_list[i] / enhanced_list[i] refer to the same utterance. Files are read with kudio.file_load -- float, mono, and any format soundfile reads, NIST SPHERE included.

Parameters:

Name	Type	Description	Default
pesq_scale	(pesq_mode,)	as in kudio.pesq. The defaults keep what this class has always reported, MOS-LQO and wideband at 16 kHz; pesq_mode='nb', pesq_scale='raw' is the raw P.862 score. A file that cannot be scored -- unreadable, at another rate, too far from its reference in length -- gets nan in every column and a warning naming it, and the rest of the list is still scored. get_df has the means. per_file has every file, so a result can be broken down by noise, SNR or talker: join it to a kudio.save_manifest manifest on the noisy column.	'lqo'

Methods:

Name	Description
eval	Score enhanced_list with every available metric.
get_df	Return the collected scores (plus a Baseline row) as a DataFrame.
per_file	Every file's scores as a DataFrame, one row per method and file.
sdr_score	(baseline, enhanced) mean SDR over enhanced_list.
stoi_score	(baseline, enhanced) mean STOI over enhanced_list.
pesq_score	(baseline, enhanced) mean PESQ over enhanced_list.
pesq_score_exe	Legacy PESQ via the ITU pesq.exe binary on PATH (Windows).

Attributes:

Name	Type	Description
rows	List[dict]	one dict per method and file -- method, clean, noisy,

Source code in kudio/core/evaluator.py

def __init__(self, clean_list: Sequence, noisy_list: Sequence, *,
             pesq_mode: str = 'auto', pesq_scale: str = 'lqo'):
    if len(clean_list) != len(noisy_list):
        raise ValueError(f"{len(clean_list)} clean files but "
                         f"{len(noisy_list)} noisy ones: the lists pair "
                         f"up by position")
    # 16 kHz allows every combination the standards define, so this
    # refuses exactly the undefined ones before any file is read
    _pesq_mode(16000, pesq_mode, pesq_scale)
    self.enhanced_list: Optional[Sequence] = None
    self.clean_list = clean_list
    self.noisy_list = noisy_list
    self.pesq_mode = pesq_mode
    self.pesq_scale = pesq_scale

    self.pesq_status, self.stoi_status, self.sdr_status = check_metrics_install()

    self.score_dict = {"Enh_method": []}
    if self.pesq_status:
        self.score_dict["PESQ"] = []
    if self.stoi_status:
        self.score_dict["STOI"] = []
    if self.sdr_status:
        self.score_dict["SDR"] = []
    self.base_pesq = self.base_stoi = self.base_sdr = None
    #: one dict per method and file -- ``method``, ``clean``, ``noisy``,
    #: ``file`` (the one scored) and a value per metric, NaN where it
    #: failed. The noisy files themselves are scored once, as 'Baseline'.
    self.rows: List[dict] = []
    self._baseline: Optional[List[dict]] = None

rows instance-attribute

rows: List[dict] = []

one dict per method and file -- method, clean, noisy, file (the one scored) and a value per metric, NaN where it failed. The noisy files themselves are scored once, as 'Baseline'.

eval

eval(enhance_method: str, enhanced_list: Sequence) -> None

Score enhanced_list with every available metric.

Source code in kudio/core/evaluator.py

def eval(self, enhance_method: str, enhanced_list: Sequence) -> None:
    """Score *enhanced_list* with every available metric."""
    self._check_length(enhanced_list)
    self.enhanced_list = enhanced_list
    columns = self._columns()
    if self._baseline is None:
        self._baseline = self._score_rows('Baseline', self.noisy_list, columns)
    rows = self._score_rows(enhance_method, enhanced_list, columns)
    self.rows.extend(rows)
    self.score_dict['Enh_method'].append(enhance_method)
    for column in columns:
        base, enhanced = _paired_means(self._baseline, rows, column)
        self.score_dict[column].append(np.around(enhanced, 4))
        setattr(self, f"base_{column.lower()}", base)

get_df

get_df() -> DataFrame

Return the collected scores (plus a Baseline row) as a DataFrame.

Before anything has been scored the DataFrame is empty, with the columns it will have, rather than a list.

Source code in kudio/core/evaluator.py

def get_df(self) -> pd.DataFrame:
    """Return the collected scores (plus a Baseline row) as a DataFrame.

    Before anything has been scored the DataFrame is empty, with the
    columns it will have, rather than a list.
    """
    if not self.score_dict['Enh_method']:
        return _pandas().DataFrame(columns=list(self.score_dict))
    table = {key: list(values) for key, values in self.score_dict.items()}
    table['Enh_method'].append('Baseline')
    if self.base_pesq is not None:
        table['PESQ'].append(np.around(self.base_pesq, 4))
    if self.base_stoi is not None:
        table['STOI'].append(np.around(self.base_stoi, 4))
    if self.base_sdr is not None:
        table['SDR'].append(np.around(self.base_sdr, 4))
    return _pandas().DataFrame.from_dict(table, orient='columns')

per_file

per_file() -> DataFrame

Every file's scores as a DataFrame, one row per method and file.

The 'Baseline' rows score the noisy files; rows holds the method rows as plain dicts, for use without pandas.

Source code in kudio/core/evaluator.py

def per_file(self) -> pd.DataFrame:
    """Every file's scores as a DataFrame, one row per method and file.

    The 'Baseline' rows score the noisy files; :attr:`rows` holds the
    method rows as plain dicts, for use without pandas.
    """
    return _pandas().DataFrame(list(self._baseline or []) + self.rows)

sdr_score

sdr_score() -> Tuple[float, float]

(baseline, enhanced) mean SDR over enhanced_list.

Source code in kudio/core/evaluator.py

def sdr_score(self) -> Tuple[float, float]:
    """``(baseline, enhanced)`` mean SDR over :attr:`enhanced_list`."""
    log.info("==== Evaluate SDR ====")
    return self._means('SDR')

stoi_score

stoi_score() -> Tuple[float, float]

(baseline, enhanced) mean STOI over enhanced_list.

Source code in kudio/core/evaluator.py

def stoi_score(self) -> Tuple[float, float]:
    """``(baseline, enhanced)`` mean STOI over :attr:`enhanced_list`."""
    log.info("==== Evaluate STOI ====")
    return self._means('STOI')

pesq_score

pesq_score() -> Tuple[float, float]

(baseline, enhanced) mean PESQ over enhanced_list.

Source code in kudio/core/evaluator.py

def pesq_score(self) -> Tuple[float, float]:
    """``(baseline, enhanced)`` mean PESQ over :attr:`enhanced_list`."""
    log.info("==== Evaluate PESQ ====")
    return self._means('PESQ')

pesq_score_exe

pesq_score_exe(
    pesq_exe: str = "pesq.exe", rate: int = 16000
) -> Tuple[float, float]

Legacy PESQ via the ITU pesq.exe binary on PATH (Windows).

The binary prints the raw narrowband P.862 score, the same scale as pesq_mode='nb', pesq_scale='raw'.

Source code in kudio/core/evaluator.py

def pesq_score_exe(self, pesq_exe: str = 'pesq.exe', rate: int = 16000
                   ) -> Tuple[float, float]:
    """Legacy PESQ via the ITU ``pesq.exe`` binary on PATH (Windows).

    The binary prints the raw narrowband P.862 score, the same scale as
    ``pesq_mode='nb', pesq_scale='raw'``.
    """
    log.info("==== Evaluate PESQ (pesq.exe) ====")
    if self.enhanced_list is None:
        raise ValueError("no enhanced files yet: call eval(method, files) first")
    if shutil.which(pesq_exe) is None:
        # a wrong path is a mistake in the call, not a file that scored nan
        raise FileNotFoundError(f"pesq binary not found: {pesq_exe!r}")

    def run(clean, other) -> float:
        # an argument list, not a shell line: a file name with a quote or
        # a space in it is passed as it is
        out = subprocess.run([pesq_exe, f'+{rate}', str(clean), str(other)],
                             capture_output=True, text=True).stdout
        try:
            return float(re.sub(r'^.*= ', '', out.strip().splitlines()[-1].strip()))
        except (IndexError, ValueError):     # no score printed for this pair
            return float('nan')

    rows = []
    for cln, noy, enh in tqdm(zip(self.clean_list, self.noisy_list,
                                  self.enhanced_list),
                              total=len(self.clean_list)):
        rows.append({'wave': os.path.basename(str(cln)),
                     'Baseline-PESQ': run(cln, noy),
                     'Enhance-PESQ': run(cln, enh)})
    mean = _pandas().DataFrame(rows).dropna().mean(numeric_only=True)
    log.info("pesq_mean:\n%s", mean)
    return mean['Baseline-PESQ'], mean['Enhance-PESQ']

snr

snr(ref: ndarray, deg: ndarray) -> float

Signal-to-noise ratio (dB) of deg relative to ref (pure numpy).

Source code in kudio/core/evaluator.py

def snr(ref: np.ndarray, deg: np.ndarray) -> float:
    """Signal-to-noise ratio (dB) of *deg* relative to *ref* (pure numpy)."""
    ref, deg = _align(ref, deg)
    noise = ref - deg
    return float(10 * np.log10((np.sum(ref ** 2) + 1e-12) /
                               (np.sum(noise ** 2) + 1e-12)))

si_sdr

si_sdr(ref: ndarray, deg: ndarray) -> float

Scale-invariant SDR (dB) — immune to overall gain differences.

Source code in kudio/core/evaluator.py

def si_sdr(ref: np.ndarray, deg: np.ndarray) -> float:
    """Scale-invariant SDR (dB) — immune to overall gain differences."""
    ref, deg = _align(ref, deg)
    ref = ref - ref.mean()
    deg = deg - deg.mean()
    proj = (np.dot(deg, ref) / (np.dot(ref, ref) + 1e-12)) * ref
    noise = deg - proj
    return float(10 * np.log10((np.sum(proj ** 2) + 1e-12) /
                               (np.sum(noise ** 2) + 1e-12)))

segmental_snr

segmental_snr(
    ref: ndarray,
    deg: ndarray,
    frame_length: int = 512,
    hop_length: int = 256,
) -> float

Frame-averaged (segmental) SNR in dB, clipped to [-10, 35] per frame.

Source code in kudio/core/evaluator.py

def segmental_snr(ref: np.ndarray, deg: np.ndarray, frame_length: int = 512,
                  hop_length: int = 256) -> float:
    """Frame-averaged (segmental) SNR in dB, clipped to [-10, 35] per frame."""
    ref, deg = _align(ref, deg)
    seg = []
    for start in range(0, len(ref) - frame_length + 1, hop_length):
        r = ref[start:start + frame_length]
        d = deg[start:start + frame_length]
        noise = r - d
        val = 10 * np.log10((np.sum(r ** 2) + 1e-12) / (np.sum(noise ** 2) + 1e-12))
        seg.append(np.clip(val, -10, 35))
    return float(np.mean(seg)) if seg else float('nan')

sdi

sdi(ref: ndarray, deg: ndarray) -> float

Speech distortion index: sum((ref - deg)**2) / sum(ref**2).

0 is a perfect copy and 1 is as much error as there is speech; lower is better. It is snr on a linear scale, 10 ** (-snr / 10): the error energy of the whole output against the clean signal. The speech-distortion index of the noise-reduction literature (Chen, Benesty, Huang and Doclo, 2006) is taken differently, on the clean speech passed through the filter alone, so the two are not the same number.

Source code in kudio/core/evaluator.py

def sdi(ref: np.ndarray, deg: np.ndarray) -> float:
    """Speech distortion index: ``sum((ref - deg)**2) / sum(ref**2)``.

    0 is a perfect copy and 1 is as much error as there is speech; lower is
    better. It is :func:`snr` on a linear scale, ``10 ** (-snr / 10)``: the
    error energy of the whole output against the clean signal. The
    speech-distortion index of the noise-reduction literature (Chen, Benesty,
    Huang and Doclo, 2006) is taken differently, on the clean speech passed
    through the filter alone, so the two are not the same number.
    """
    ref, deg = _align(ref, deg)
    return float(np.sum((ref - deg) ** 2) / (np.sum(ref ** 2) + 1e-12))

pesq

pesq(
    ref: ndarray,
    deg: ndarray,
    sr: int,
    *,
    mode: str = "auto",
    scale: str = "lqo",
) -> float

PESQ of deg against ref (ITU-T P.862 / P.862.2).

>>> kudio.pesq(clean, enhanced, 16000)                          # WB MOS-LQO
>>> kudio.pesq(clean, enhanced, 16000, mode='nb', scale='raw')  # raw P.862

Parameters:

Name	Type	Description	Default
mode	str	'nb' -- P.862 narrowband, at 8 or 16 kHz. 'wb' -- P.862.2 wideband, 16 kHz only. 'auto' (default) -- wideband at 16 kHz and narrowband at 8 kHz, which is what kudio has always reported; narrowband whenever scale is 'raw'.	'auto'
scale	str	'lqo' (default) -- MOS-LQO, the mapped score both standards define. 'raw' -- the narrowband P.862 score before the P.862.1 mapping, -0.5 to 4.5: the number the ITU reference binary prints.	'lqo'

Raises:

Type	Description
ValueError	off 8 / 16 kHz, or for a combination the standards do not define (wideband at 8 kHz, a raw wideband score).
DependencyError	without the pesq package (kudio[pesq]). The pair is checked for length like every other metric here: a mismatch warns, and past MAX_LENGTH_MISMATCH it is refused. Runs on the pesq package from PyPI, which returns MOS-LQO; the raw score is that inverted through P.862.1, (4.6607 - ln(4 / (lqo - 0.999) - 1)) / 1.4945.

Source code in kudio/core/evaluator.py

def pesq(ref: np.ndarray, deg: np.ndarray, sr: int, *, mode: str = 'auto',
         scale: str = 'lqo') -> float:
    """PESQ of *deg* against *ref* (ITU-T P.862 / P.862.2).

    >>> kudio.pesq(clean, enhanced, 16000)                          # WB MOS-LQO
    >>> kudio.pesq(clean, enhanced, 16000, mode='nb', scale='raw')  # raw P.862

    :param mode: ``'nb'`` -- P.862 narrowband, at 8 or 16 kHz. ``'wb'`` --
        P.862.2 wideband, 16 kHz only. ``'auto'`` (default) -- wideband at
        16 kHz and narrowband at 8 kHz, which is what kudio has always
        reported; narrowband whenever *scale* is ``'raw'``.
    :param scale: ``'lqo'`` (default) -- MOS-LQO, the mapped score both
        standards define. ``'raw'`` -- the narrowband P.862 score before the
        P.862.1 mapping, -0.5 to 4.5: the number the ITU reference binary
        prints.
    :raises ValueError: off 8 / 16 kHz, or for a combination the standards
        do not define (wideband at 8 kHz, a raw wideband score).
    :raises DependencyError: without the ``pesq`` package (``kudio[pesq]``).

    The pair is checked for length like every other metric here: a mismatch
    warns, and past :data:`MAX_LENGTH_MISMATCH` it is refused.

    Runs on the ``pesq`` package from PyPI, which returns MOS-LQO; the raw
    score is that inverted through P.862.1,
    ``(4.6607 - ln(4 / (lqo - 0.999) - 1)) / 1.4945``.
    """
    mode = _pesq_mode(int(sr), mode, scale)
    try:
        from pesq import pesq as _pesq
    except ImportError as e:
        raise DependencyError('pesq', extra='pesq') from e
    ref, deg = _align(np.asarray(ref).squeeze(), np.asarray(deg).squeeze())
    lqo = float(_pesq(int(sr), ref, deg, mode))
    if scale == 'lqo':
        return lqo
    if not 0.999 < lqo < 4.999:              # outside the mapping's range
        return float('nan')
    return (4.6607 - math.log(4.0 / (lqo - 0.999) - 1.0)) / 1.4945

score

score(
    ref: ndarray,
    deg: ndarray,
    sr: int,
    metrics: Optional[Sequence[str]] = None,
    *,
    pesq_mode: str = "auto",
    pesq_scale: str = "lqo",
) -> Dict[str, float]

Score deg against ref: {metric: value}.

>>> kudio.score(clean, enhanced, 16000, ['pesq', 'stoi', 'sdi'],
...             pesq_mode='nb', pesq_scale='raw')
{'pesq': 2.41, 'stoi': 0.874, 'sdi': 0.105}

Parameters:

Name	Type	Description	Default
metrics	Optional[Sequence[str]]	names from METRICS. None means every one that can run here: the four that need nothing, each optional one whose package is installed, and PESQ only at 8 or 16 kHz. A metric named explicitly that cannot run raises instead of going missing.	None
pesq_scale	(pesq_mode,)	as mode / scale of pesq. The two signals are compared sample for sample over their common length, as the rest of kudio's reference metrics are -- see kudio.align for a pair that may not line up. A length difference is warned about, and refused past MAX_LENGTH_MISMATCH of the longer signal.	'lqo'

Source code in kudio/core/evaluator.py

def score(ref: np.ndarray, deg: np.ndarray, sr: int,
          metrics: Optional[Sequence[str]] = None, *,
          pesq_mode: str = 'auto', pesq_scale: str = 'lqo') -> Dict[str, float]:
    """Score *deg* against *ref*: ``{metric: value}``.

    >>> kudio.score(clean, enhanced, 16000, ['pesq', 'stoi', 'sdi'],
    ...             pesq_mode='nb', pesq_scale='raw')
    {'pesq': 2.41, 'stoi': 0.874, 'sdi': 0.105}

    :param metrics: names from :data:`METRICS`. ``None`` means every one that
        can run here: the four that need nothing, each optional one whose
        package is installed, and PESQ only at 8 or 16 kHz. A metric named
        explicitly that cannot run raises instead of going missing.
    :param pesq_mode, pesq_scale: as *mode* / *scale* of :func:`pesq`.

    The two signals are compared sample for sample over their common length,
    as the rest of kudio's reference metrics are -- see :func:`kudio.align`
    for a pair that may not line up. A length difference is warned about, and
    refused past :data:`MAX_LENGTH_MISMATCH` of the longer signal.
    """
    if metrics is None:
        names = [m for m in METRICS
                 if (m not in _BACKEND or _has_module(_BACKEND[m][0]))
                 and (m != 'pesq' or int(sr) in (8000, 16000))]
    else:
        names = [metrics] if isinstance(metrics, str) else list(metrics)
        unknown = [m for m in names if m not in METRICS]
        if unknown:
            raise ValueError(f"unknown metric(s) {unknown}; choose from {METRICS}")
        for m in names:
            if m in _BACKEND and not _has_module(_BACKEND[m][0]):
                raise DependencyError(*_BACKEND[m])
    ref, deg = _align(np.asarray(ref).squeeze(), np.asarray(deg).squeeze())
    return {m: _compute(m, ref, deg, int(sr), pesq_mode, pesq_scale) for m in names}

check_metrics_install

check_metrics_install() -> Tuple[bool, bool, bool]

Report availability of the optional metric back-ends.

Returns (pesq_ok, stoi_ok, sdr_ok). A missing back-end is logged once per process, with how to install it; nothing is installed automatically.

Source code in kudio/core/evaluator.py

def check_metrics_install() -> Tuple[bool, bool, bool]:
    """Report availability of the optional metric back-ends.

    Returns ``(pesq_ok, stoi_ok, sdr_ok)``. A missing back-end is logged once
    per process, with how to install it; nothing is installed automatically.
    """
    pesq_ok = _has_module("pesq")
    stoi_ok = _has_module("pystoi")
    sdr_ok = _has_module("mir_eval")
    for ok, what, hint in (
            (pesq_ok, "PESQ", "pip install kudio[pesq] (builds from source: "
                              "needs a C compiler -- on Windows, the MSVC "
                              "Build Tools)"),
            (stoi_ok, "STOI", "pip install kudio[eval]"),
            (sdr_ok, "SDR", "pip install kudio[eval]")):
        if not ok and what not in _REPORTED:
            _REPORTED.add(what)
            log.warning("%s unavailable: %s", what, hint)
    return pesq_ok, stoi_ok, sdr_ok

eval_metrics

eval_metrics(
    rate: int,
    ref: ndarray,
    deg: ndarray,
    *,
    pesq_mode: str = "auto",
    pesq_scale: str = "lqo",
) -> Tuple[
    Optional[float], Optional[float], Optional[float]
]

Score a single (reference, degraded) pair with every available metric.

Returns (pesq, stoi, sdr) -- always three values, whatever is installed. A metric comes back as None when its back-end is not installed, when it cannot run on this pair -- PESQ off 8 and 16 kHz, STOI on less than about 0.4 s of speech -- or when it fails on it, which is logged.

Parameters:

Name	Type	Description	Default
pesq_mode	str	'nb', 'wb' or 'auto' (by sample rate).	'auto'
pesq_scale	str	'lqo' for MOS-LQO, or 'raw' for the narrowband P.862 score. The defaults are what kudio has always reported.	'lqo'

Source code in kudio/core/evaluator.py

def eval_metrics(rate: int, ref: np.ndarray, deg: np.ndarray, *,
                 pesq_mode: str = 'auto', pesq_scale: str = 'lqo'
                 ) -> Tuple[Optional[float], Optional[float], Optional[float]]:
    """Score a single (reference, degraded) pair with every available metric.

    Returns ``(pesq, stoi, sdr)`` -- always three values, whatever is
    installed. A metric comes back as ``None`` when its back-end is not
    installed, when it cannot run on this pair -- PESQ off 8 and 16 kHz, STOI
    on less than about 0.4 s of speech -- or when it fails on it, which is
    logged.

    :param pesq_mode: ``'nb'``, ``'wb'`` or ``'auto'`` (by sample rate).
    :param pesq_scale: ``'lqo'`` for MOS-LQO, or ``'raw'`` for the narrowband
        P.862 score. The defaults are what kudio has always reported.
    """
    pesq_status, stoi_status, sdr_status = check_metrics_install()
    ref, deg = _align(np.asarray(ref).squeeze(), np.asarray(deg).squeeze())
    wanted = [('pesq', pesq_status and int(rate) in (8000, 16000)),
              ('stoi', stoi_status), ('sdr', sdr_status)]
    values = []
    for name, ok in wanted:
        value = None
        if ok:
            try:
                value = _compute(name, ref, deg, int(rate),
                                 pesq_mode=pesq_mode, pesq_scale=pesq_scale)
                if value is not None and np.isnan(value):
                    value = None            # STOI on a clip too short for it
            except Exception as e:                          # noqa: BLE001
                log.warning("%s could not be computed: %s", name.upper(), e)
        values.append(value)
    return tuple(values)

kudio.core.align

Line two recordings up before measuring one against the other.

The SNR family in kudio -- kudio.si_sdr, kudio.snr, kudio.segmental_snr, kudio.sdi -- compares sample i of one signal with sample i of the other. Every reference metric trims to the shorter of the two; the SNR family also assumes the rest lines up, which is right for a process that returns what it was given and wrong for anything that introduces a delay. PESQ runs its own time alignment and STOI shrugs off a few milliseconds, so the damage below is the SNR family's. A trim large enough to matter is warned about, and past kudio.core.evaluator.MAX_LENGTH_MISMATCH refused -- but a delay that leaves the lengths equal is invisible to that check, and this is the module for it.

How wrong is worth seeing. A clip scored against itself, shifted:

========= ========== shift SI-SDR ========= ========== 0 +145 dB 1 sample +10.6 dB 1 ms -10.1 dB 10 ms -11.1 dB ========= ==========

One sample costs 134 dB. A millisecond scores worse than not processing at all, and nothing in the number says "these are misaligned" rather than "your denoiser destroyed the signal" -- which is the wrong conclusion to hand somebody about work that may be fine.

>>> shift = kudio.find_delay(clean, processed, sr)
>>> print(shift)
lags by 128 samples (8.0 ms), correlation 0.998
>>> a, b = kudio.align(clean, processed, sr)
>>> kudio.si_sdr(a, b)
18.4

The confidence is half of it. Cross-correlation always has a maximum, so it always returns a delay -- including for two recordings that have nothing to do with each other. Alignment.correlation is what separates "these are the same take, 8 ms apart" from "these are different recordings and the number means nothing", and align refuses rather than lining up two unrelated signals and letting a metric put a number on the result.

Polarity is reported, not silently fixed. An inverted copy correlates at -1: it is the same take, and something in the chain flipped it. That is a finding, so it is named rather than corrected on the way past.

Classes:

Name	Description
Alignment	Where two recordings sit relative to each other.

Functions:

Name	Description
find_delay	Measure how far deg sits after ref, to the nearest sample.
align	Return ref and deg trimmed to the region they share, lined up.

Attributes:

Name	Type	Description
MIN_CORRELATION		Below this, two signals are not the same recording and the delay between

MIN_CORRELATION module-attribute

MIN_CORRELATION = 0.5

Below this, two signals are not the same recording and the delay between them does not mean anything. Chosen low: real pairs -- a denoised take against its source, a resampled copy -- sit well above 0.8 even when the processing is aggressive, and the point of the threshold is to catch "somebody picked the wrong file", not to grade the processing.

Alignment dataclass

Alignment(
    delay: int, correlation: float, sr: int, overlap: int
)

Where two recordings sit relative to each other.

Parameters:

Name	Type	Description	Default
delay	int	samples by which deg sits after ref. Negative means it starts first. Zero means they already line up.	required
correlation	float	normalised cross-correlation at that delay, in [-1, 1]. Negative means one of them is polarity-inverted.	required
overlap	int	samples the two share once the delay is taken out.	required

Methods:

Name	Description
confident	Are these plausibly the same recording? minimum defaults to
apply	Trim both signals to the region they share at this delay.

Attributes:

Name	Type	Description
inverted	bool	Is one of the two polarity-flipped relative to the other?
strength	float	How alike they are, ignoring polarity: abs(correlation).

inverted property

inverted: bool

Is one of the two polarity-flipped relative to the other?

strength property

strength: float

How alike they are, ignoring polarity: abs(correlation).

confident

confident(minimum: Optional[float] = None) -> bool

Are these plausibly the same recording? minimum defaults to MIN_CORRELATION as it is when called, so raising that raises it everywhere.

Source code in kudio/core/align.py

def confident(self, minimum: Optional[float] = None) -> bool:
    """Are these plausibly the same recording? *minimum* defaults to
    :data:`MIN_CORRELATION` as it is when called, so raising that raises
    it everywhere."""
    return self.strength >= (MIN_CORRELATION if minimum is None else minimum)

apply

apply(
    ref: ndarray, deg: ndarray
) -> Tuple[ndarray, ndarray]

Trim both signals to the region they share at this delay.

Separate from find_delay so a delay measured once -- on a loud passage, say -- can be applied to a whole set of files that came through the same chain.

Source code in kudio/core/align.py

def apply(self, ref: np.ndarray,
          deg: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """Trim both signals to the region they share at this delay.

    Separate from :func:`find_delay` so a delay measured once -- on a loud
    passage, say -- can be applied to a whole set of files that came
    through the same chain.
    """
    ref = np.asarray(ref)
    deg = np.asarray(deg)
    if self.delay > 0:
        deg = deg[self.delay:]
    elif self.delay < 0:
        ref = ref[-self.delay:]
    n = min(len(ref), len(deg))
    return ref[:n], deg[:n]

find_delay

find_delay(
    ref: ndarray,
    deg: ndarray,
    sr: int,
    *,
    max_seconds: Optional[float] = None,
) -> Alignment

Measure how far deg sits after ref, to the nearest sample.

Parameters:

Name	Type	Description	Default
max_seconds	Optional[float]	bound the search to this much delay either way. Worth setting when you know the answer is small: an unbounded search over a long file can find a better-correlating match half a minute away in something as repetitive as speech, and be perfectly right about a passage nobody meant.	None

Returns:

Type	Description
Alignment	an Alignment. It always returns one -- ask it whether to believe it. Both signals must already be at sr; loudness and pitch survive a resample and a sample offset does not, so aligning two rates would be measuring a delay in units that do not exist yet.

Source code in kudio/core/align.py

def find_delay(ref: np.ndarray, deg: np.ndarray, sr: int, *,
               max_seconds: Optional[float] = None) -> Alignment:
    """Measure how far *deg* sits after *ref*, to the nearest sample.

    :param max_seconds: bound the search to this much delay either way.
        Worth setting when you know the answer is small: an unbounded search
        over a long file can find a better-correlating match half a minute
        away in something as repetitive as speech, and be perfectly right
        about a passage nobody meant.
    :returns: an :class:`Alignment`. It always returns one -- ask it whether
        to believe it.

    Both signals must already be at *sr*; loudness and pitch survive a
    resample and a sample offset does not, so aligning two rates would be
    measuring a delay in units that do not exist yet.
    """
    a = _mono("find_delay()", ref)
    b = _mono("find_delay()", deg)
    if sr <= 0:
        raise FeatureError(f"find_delay() needs a positive rate, got {sr}")
    if a.size == 0 or b.size == 0:
        raise FeatureError("find_delay() needs audio on both sides, got "
                           f"{a.size} and {b.size} samples")

    # A DC offset correlates with everything, including a step of silence, and
    # would drag the peak toward whichever lag overlaps most.
    a = a - a.mean()
    b = b - b.mean()

    from scipy.signal import correlate, correlation_lags
    corr = correlate(b, a, mode="full", method="fft")
    lags = correlation_lags(len(b), len(a), mode="full")

    if max_seconds is not None:
        limit = int(round(max_seconds * sr))
        if limit < 0:
            raise FeatureError(f"max_seconds must be >= 0, got {max_seconds}")
        keep = np.abs(lags) <= limit
        if not keep.any():                     # limit smaller than one sample
            keep = lags == 0
        corr, lags = corr[keep], lags[keep]

    delay = int(lags[int(np.argmax(np.abs(corr)))])

    # Normalise over the region the two actually share at that delay, rather
    # than over their whole lengths: a genuine match that only overlaps for a
    # second would otherwise score as badly as a mismatch.
    left, right = (a[-delay:], b) if delay < 0 else (a, b[delay:])
    n = min(len(left), len(right))
    left, right = left[:n], right[:n]
    energy = float(np.sqrt(np.dot(left, left) * np.dot(right, right)))
    correlation = float(np.dot(left, right) / energy) if energy > 0 else 0.0
    return Alignment(delay=delay, correlation=correlation, sr=int(sr),
                     overlap=int(n))

align

align(
    ref: ndarray,
    deg: ndarray,
    sr: int,
    *,
    max_seconds: Optional[float] = None,
    min_correlation: Optional[float] = None,
) -> Tuple[ndarray, ndarray]

Return ref and deg trimmed to the region they share, lined up.

The pair comes back ready for kudio.si_sdr and friends.

Parameters:

Name	Type	Description	Default
min_correlation	Optional[float]	refuse below this (MIN_CORRELATION when not given), because two unrelated recordings still have a best lag and lining them up at it produces a confident-looking number about nothing. Pass 0 to align whatever you have anyway.	None

Raises:

Type	Description
FeatureError	when the two do not look like the same recording. >>> a, b = kudio.align(clean, processed, sr) >>> kudio.si_sdr(a, b) 18.4

Source code in kudio/core/align.py

def align(ref: np.ndarray, deg: np.ndarray, sr: int, *,
          max_seconds: Optional[float] = None,
          min_correlation: Optional[float] = None
          ) -> Tuple[np.ndarray, np.ndarray]:
    """Return *ref* and *deg* trimmed to the region they share, lined up.

    The pair comes back ready for :func:`kudio.si_sdr` and friends.

    :param min_correlation: refuse below this (:data:`MIN_CORRELATION` when
        not given), because two unrelated recordings still have a best lag
        and lining them up at it produces a confident-looking number about
        nothing. Pass ``0`` to align whatever you have anyway.
    :raises FeatureError: when the two do not look like the same recording.

    >>> a, b = kudio.align(clean, processed, sr)           # doctest: +SKIP
    >>> kudio.si_sdr(a, b)                                 # doctest: +SKIP
    18.4
    """
    found = find_delay(ref, deg, sr, max_seconds=max_seconds)
    if min_correlation is None:
        min_correlation = MIN_CORRELATION
    if not found.confident(min_correlation):
        raise FeatureError(
            f"these do not look like the same recording: best correlation "
            f"{found.strength:.3f} at {found.delay} samples, below "
            f"{min_correlation:g}. Aligning them would put a number on two "
            f"unrelated signals; pass min_correlation=0 to do it anyway.")
    if found.inverted:
        log.warning("align: the two are polarity-inverted relative to each "
                    "other (correlation %.3f); lining them up as they are",
                    found.correlation)
    return found.apply(ref, deg)