initial code

Author: wq2012

.github/workflows/python-package.yml

@@ -0,0 +1,31 @@
+name: Python Package
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.11"]
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+        pip install pytest
+    - name: Test with pytest
+      run: |
+        pytest

.gitignore

@@ -1 +1,4 @@
 downloads/*
+debug/*
+*.pyc
+dist/*

LICENSE

@@ -0,0 +1,7 @@
+Copyright (c) 2025 Quan Wang
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,211 @@
+# mdeval
+
+A Python implementation of the NIST `md-eval.pl` script for evaluating rich transcription and speaker diarization accuracy. This tool mimics the core functionality and scoring logic of the standard Perl script used in NIST evaluations (e.g., RT-0x), focusing on Diarization Error Rate (DER).
+
+## Table of Contents
+
+- [Overview](#overview)
+- [Installation](#installation)
+- [Usage](#usage)
+  - [Command Line Interface](#command-line-interface)
+  - [Python API](#python-api)
+- [Input Formats](#input-formats)
+  - [RTTM (Rich Transcription Time Marked)](#rttm-rich-transcription-time-marked)
+  - [UEM (Un-partitioned Evaluation Map)](#uem-un-partitioned-evaluation-map)
+- [Core Algorithms](#core-algorithms)
+  - [Scoring Logic](#scoring-logic)
+  - [Optimal Speaker Mapping](#optimal-speaker-mapping)
+  - [Collars](#collars)
+  - [Overlap Exclusion](#overlap-exclusion)
+- [Testing](#testing)
+- [Citation](#citation)
+
+## Overview
+
+`mdeval` calculates the Diarization Error Rate (DER) by comparing a system hypothesis (SYS) against a ground truth reference (REF). It supports:
+- **Missed Speech**: Speech present in REF but not in SYS.
+- **False Alarm**: Speech present in SYS but not in REF.
+- **Speaker Error**: Speech assigned to the wrong speaker (after optimal mapping).
+- **Collars**: Optional no-score zones around reference segment boundaries.
+- **Overlap handling**: Option to exclude regions where multiple reference speakers talk simultaneously.
+
+The goal is to provide a pure Python, dependency-free (or minimal dependency) alternative to the legacy Perl script for modern pipelines.
+
+## Installation
+
+You can install the package via pip:
+
+```bash
+pip install mdeval
+```
+
+## Usage
+
+### Command Line Interface
+
+The package provides a CLI entry point `mdeval`.
+
+```bash
+python3 -m mdeval.cli -r <ref_rttm> -s <sys_rttm> [options]
+```
+
+**Arguments:**
+
+- `-r, --ref`: Path to the Reference RTTM file (Required).
+- `-s, --sys`: Path to the System/Hypothesis RTTM file (Required).
+- `-u, --uem`: Path to the UEM file defining evaluation regions (Optional. If omitted, the valid region is inferred from the Reference RTTM).
+- `-c, --collar`: Collar size in seconds (Float, default: 0.0). A "no-score" zone of +/- `collar` seconds is applied around every reference segment boundary.
+- `-1, --single-speaker`: Limit scoring to single-speaker regions only (ignore overlaps in REF). This is equivalent to "Overlap Exclusion".
+
+**Example:**
+
+```bash
+python3 -m mdeval.cli -r ref.rttm -s hyp.rttm -c 0.25
+```
+
+### Python API
+
+You can use the scoring logic programmatically:
+
+```python
+from mdeval.io import load_rttm, load_uem
+from mdeval.scoring import score_speaker_diarization
+from mdeval.utils import Segment
+
+# Load Data
+ref_data = load_rttm('ref.rttm')
+sys_data = load_rttm('sys.rttm')
+
+# Define Evaluation Map (or infer it)
+# uem_eval = [Segment(0.0, 100.0)]
+# Or load:
+# uem_data = load_uem('test.uem')
+# uem_eval = uem_data['file1']['1']
+
+# Parse specific file/channel data
+ref_spkrs = {} # ... extract from ref_data['file1']['1']['SPEAKER']
+sys_spkrs = {} # ... extract from sys_data['file1']['1']['SPEAKER']
+
+# Score
+stats, mapping = score_speaker_diarization(
+    'file1', '1', 
+    ref_spkrs, sys_spkrs, 
+    uem_eval, 
+    collar=0.25, 
+    ignore_overlap=False
+)
+
+print(f"DER: {stats['MISSED_SPEAKER'] + stats['FALARM_SPEAKER'] + stats['SPEAKER_ERROR']}")
+```
+
+## Input Formats
+
+### RTTM (Rich Transcription Time Marked)
+
+Format used for both Reference and System inputs.
+Space-delimited text file. Lines starting with `;` or `#` are ignored.
+
+**Required Columns (indices 0-8):**
+
+1.  **TYPE**: Segment type (must be `SPEAKER` to be scored).
+2.  **FILE**: File name / Recording ID.
+3.  **CHNL**: Channel ID (e.g., `1`).
+4.  **TBEG**: Start time in seconds (float).
+5.  **TDUR**: Duration in seconds (float).
+6.  **ORTHO**: Orthography field (ignored/placeholder, e.g., `<NA>`).
+7.  **STYPE**: Subtype (ignored/placeholder, e.g., `<NA>`).
+8.  **NAME**: Speaker Name/ID.
+9.  **CONF**: Confidence score (ignored/placeholder, e.g., `<NA>`).
+
+**Example:**
+```
+SPEAKER file1 1 0.00 5.00 <NA> <NA> spk1 <NA> <NA>
+SPEAKER file1 1 5.00 3.00 <NA> <NA> spk2 <NA> <NA>
+```
+
+### UEM (Un-partitioned Evaluation Map)
+
+Defines the time regions that should be evaluated. Regions outside the UEM are ignored.
+Space-delimited text file.
+
+**Required Columns:**
+
+1.  **FILE**: File name.
+2.  **CHNL**: Channel ID.
+3.  **TBEG**: Start time of valid region.
+4.  **TEND**: End time of valid region.
+
+**Example:**
+```
+file1 1 0.00 100.00
+file1 1 120.00 300.00
+```
+
+## Core Algorithms
+
+### Scoring Logic
+
+The scoring is segment-based (time-weighted).
+1.  **Metric**: Diarization Error Rate (DER).
+    $$ DER = \frac{\text{Missed Speaker Time} + \text{False Alarm Speaker Time} + \text{Speaker Error Time}}{\text{Total Scored Speaker Time}} $$
+2.  **Segmentation**: The timeline is split into contiguous segments where the set of reference and system speakers remains constant.
+3.  **Intersection**: For each segment, the number of reference speakers ($N_{ref}$) and system speakers ($N_{sys}$) is compared.
+
+### Optimal Speaker Mapping
+
+Since System speaker labels (e.g., "sys01") do not match Reference labels (e.g., "spk01"), a global 1-to-1 mapping is computed to minimize error.
+-   We compute an overlap matrix between every reference speaker and every system speaker over the entire valid UEM duration.
+-   The **Hungarian Algorithm** (implemented purely in Python, no `scipy` dependency required) is used to find the optimal assignment that maximizes total overlap time.
+
+### Collars
+
+When `collar > 0`, a "no-score" zone is applied.
+-   For every segment boundary in the **Reference** RTTM, a region of $t \pm collar$ is removed from the UEM.
+-   This accounts for human annotation uncertainty boundaries.
+-   **Note**: The Python implementation follows the logic of `md-eval.pl`'s `add_collars_to_uem` subroutine, using a counter-based approach to subtract the union of all collar regions from the scoring UEM.
+
+### Overlap Exclusion
+
+If enabled (via `-1` / `--single-speaker`), regions where **two or more** Reference speakers are speaking simultaneously are removed from the UEM.
+-   This allows evaluation of systems that only output single-speaker segments.
+-   **Note**: Overlap exclusion is applied *before* collars in the perl script logic, but effectively they both just subtract time from the valid UEM.
+
+## Testing
+
+The package includes unit tests using Python's `unittest` framework.
+
+Run tests via:
+```bash
+python3 -m unittest discover tests
+```
+
+## Citation
+
+We developed this package as part of the following work:
+
+```
+@inproceedings{wang2018speaker,
+  title={{Speaker Diarization with LSTM}},
+  author={Wang, Quan and Downey, Carlton and Wan, Li and Mansfield, Philip Andrew and Moreno, Ignacio Lopz},
+  booktitle={2018 IEEE International Conference on Acoustics, Speech and Signal Processing (ICASSP)},
+  pages={5239--5243},
+  year={2018},
+  organization={IEEE}
+}
+
+@inproceedings{xia2022turn,
+  title={{Turn-to-Diarize: Online Speaker Diarization Constrained by Transformer Transducer Speaker Turn Detection}},
+  author={Wei Xia and Han Lu and Quan Wang and Anshuman Tripathi and Yiling Huang and Ignacio Lopez Moreno and Hasim Sak},
+  booktitle={2022 IEEE International Conference on Acoustics, Speech and Signal Processing (ICASSP)},
+  pages={8077--8081},
+  year={2022},
+  organization={IEEE}
+}
+
+@article{wang2022highly,
+  title={Highly Efficient Real-Time Streaming and Fully On-Device Speaker Diarization with Multi-Stage Clustering},
+  author={Quan Wang and Yiling Huang and Han Lu and Guanlong Zhao and Ignacio Lopez Moreno},
+  journal={arXiv:2210.13690},
+  year={2022}
+}
+```

mdeval/__init__.py

@@ -0,0 +1,138 @@
+import argparse
+import sys
+import os
+from typing import List
+from .io import load_rttm, load_uem
+from .scoring import score_speaker_diarization
+from .utils import Segment
+
+def main():
+    parser = argparse.ArgumentParser(description='Python implementation of NIST md-eval.pl')
+    parser.add_argument('-r', '--ref', required=True, help='Reference RTTM file')
+    parser.add_argument('-s', '--sys', required=True, help='System RTTM file')
+    parser.add_argument('-u', '--uem', help='UEM file (Evaluation Partition)')
+    parser.add_argument('-c', '--collar', type=float, default=0.0, help='No-score collar around reference boundaries (seconds)')
+    parser.add_argument('-1', '--single-speaker', action='store_true', dest='single_speaker', help='Limit scoring to single-speaker regions')
+    # Add other flags as needed
+    
+    args = parser.parse_args()
+    
+    # Load Data
+    ref_data = load_rttm(args.ref)
+    sys_data = load_rttm(args.sys)
+    
+    uem_data = None
+    if args.uem:
+        uem_data = load_uem(args.uem)
+    
+    # Process each file found in REF
+    files = sorted(ref_data.keys())
+    
+    # Accumulate global scores
+    total_stats = {
+        'EVAL_TIME': 0.0,
+        'EVAL_SPEECH': 0.0,
+        'SCORED_TIME': 0.0,
+        'SCORED_SPEECH': 0.0,
+        'MISSED_SPEECH': 0.0,
+        'FALARM_SPEECH': 0.0,
+        'SCORED_SPEAKER': 0.0,
+        'MISSED_SPEAKER': 0.0,
+        'FALARM_SPEAKER': 0.0,
+        'SPEAKER_ERROR': 0.0,
+        'SCORED_WORDS': 0, # Placeholder
+        'EVAL_WORDS': 0
+    }
+    
+    # TODO: Output header matching md-eval.pl
+    
+    for file in files:
+        if file not in sys_data:
+            print(f"Warning: File {file} found in REF but not in SYS. Skipping.", file=sys.stderr)
+            continue
+            
+        chnls = sorted(ref_data[file].keys())
+        for chnl in chnls:
+            if chnl not in sys_data[file]:
+                print(f"Warning: Channel {chnl} for file {file} found in REF but not in SYS. Skipping.", file=sys.stderr)
+                continue
+                
+            # Determine UEM
+            if uem_data and file in uem_data and chnl in uem_data[file]:
+                uem_eval = uem_data[file][chnl]
+            else:
+                # Infer UEM from REF RTTM (min TBEG, max TEND)
+                min_t = 1e30
+                max_t = 0
+                found_seg = False
+                for seg in ref_data[file][chnl]['SPEAKER']:
+                    min_t = min(min_t, seg['TBEG'])
+                    max_t = max(max_t, seg['TEND'])
+                    found_seg = True
+                if not found_seg:
+                    # Try other types? For now just SPEAKER
+                    pass
+                if max_t > min_t:
+                    uem_eval = [Segment(min_t, max_t)]
+                else:
+                    uem_eval = []
+
+            # Determine REF/SYS inputs
+            # Group by speaker
+            # Expected format for scoring: {spkr: [{TBEG, TDUR, TEND, ...}]}
+            curr_ref = {}
+            for seg in ref_data[file][chnl]['SPEAKER']:
+                s = seg['SPKR']
+                if s not in curr_ref: curr_ref[s] = []
+                curr_ref[s].append(seg)
+                
+            curr_sys = {}
+            if 'SPEAKER' in sys_data[file][chnl]:
+                for seg in sys_data[file][chnl]['SPEAKER']:
+                    s = seg['SPKR']
+                    if s not in curr_sys: curr_sys[s] = []
+                    curr_sys[s].append(seg)
+            
+            file_stats, _ = score_speaker_diarization(file, chnl, curr_ref, curr_sys, uem_eval, args.collar, args.single_speaker)
+            
+            # Add to totals
+            for k in total_stats:
+                if k in file_stats:
+                    total_stats[k] += file_stats[k]
+    
+    # Print simplified output
+    print_scores("ALL", total_stats)
+
+def print_scores(condition, scores):
+    print(f"\n*** Performance analysis for Speaker Diarization for {condition} ***\n")
+    
+    def p(val): return val
+    
+    eval_time = scores['EVAL_TIME']
+    eval_speech = scores['EVAL_SPEECH']
+    scored_time = scores['SCORED_TIME']
+    scored_speech = scores['SCORED_SPEECH']
+    
+    print(f"    EVAL TIME = {eval_time:10.2f} secs")
+    print(f"  EVAL SPEECH = {eval_speech:10.2f} secs ({100*eval_speech/eval_time if eval_time else 0:5.1f} percent of evaluated time)")
+    print(f"  SCORED TIME = {scored_time:10.2f} secs ({100*scored_time/eval_time if eval_time else 0:5.1f} percent of evaluated time)")
+    print(f"SCORED SPEECH = {scored_speech:10.2f} secs ({100*scored_speech/scored_time if scored_time else 0:5.1f} percent of scored time)")
+    print(f"   EVAL WORDS = {scores['EVAL_WORDS']:7d}        ")
+    print(f" SCORED WORDS = {scores['SCORED_WORDS']:7d}         (100.0 percent of evaluated words)")
+    print("---------------------------------------------")
+    print(f"MISSED SPEECH = {scores['MISSED_SPEECH']:10.2f} secs ({100*scores['MISSED_SPEECH']/scored_time if scored_time else 0:5.1f} percent of scored time)")
+    print(f"FALARM SPEECH = {scores['FALARM_SPEECH']:10.2f} secs ({100*scores['FALARM_SPEECH']/scored_time if scored_time else 0:5.1f} percent of scored time)")
+    print(f" MISSED WORDS =       0         (100.0 percent of scored words)")
+    print("---------------------------------------------")
+    print(f"SCORED SPEAKER TIME = {scores['SCORED_SPEAKER']:10.2f} secs ({100*scores['SCORED_SPEAKER']/scored_speech if scored_speech else 0:5.1f} percent of scored speech)")
+    print(f"MISSED SPEAKER TIME = {scores['MISSED_SPEAKER']:10.2f} secs ({100*scores['MISSED_SPEAKER']/scores['SCORED_SPEAKER'] if scores['SCORED_SPEAKER'] else 0:5.1f} percent of scored speaker time)")
+    print(f"FALARM SPEAKER TIME = {scores['FALARM_SPEAKER']:10.2f} secs ({100*scores['FALARM_SPEAKER']/scores['SCORED_SPEAKER'] if scores['SCORED_SPEAKER'] else 0:5.1f} percent of scored speaker time)")
+    print(f" SPEAKER ERROR TIME = {scores['SPEAKER_ERROR']:10.2f} secs ({100*scores['SPEAKER_ERROR']/scores['SCORED_SPEAKER'] if scores['SCORED_SPEAKER'] else 0:5.1f} percent of scored speaker time)")
+    print(f"SPEAKER ERROR WORDS =       0         (100.0 percent of scored speaker words)")
+    print("---------------------------------------------")
+    der = (scores['MISSED_SPEAKER'] + scores['FALARM_SPEAKER'] + scores['SPEAKER_ERROR']) / scores['SCORED_SPEAKER'] if scores['SCORED_SPEAKER'] else 0
+    print(f" OVERALL SPEAKER DIARIZATION ERROR = {100*der:5.2f} percent of scored speaker time  `({condition})")
+    print("---------------------------------------------")
+
+if __name__ == '__main__':
+    main()