Migrate editdistance metrics to rapidfuzz with empty-reference guard

Replace archived editdistance with rapidfuzz and return None when CER/WER references have zero total length.

Supersedes #3105.

Author: LauraGPT

funasr/metrics/common.py

@@ -9,8 +9,9 @@
 import json
 import logging
 import sys
-
 from itertools import groupby
+
+from rapidfuzz.distance import Levenshtein
 import numpy as np
 import six
 
@@ -155,7 +156,6 @@ def calculate_cer_ctc(self, ys_hat, ys_pad):
         :return: average sentence-level CER score
         :rtype float
         """
-        import editdistance
 
         cers, char_ref_lens = [], []
         for i, y in enumerate(ys_hat):
@@ -175,7 +175,7 @@ def calculate_cer_ctc(self, ys_hat, ys_pad):
             hyp_chars = "".join(seq_hat)
             ref_chars = "".join(seq_true)
             if len(ref_chars) > 0:
-                cers.append(editdistance.eval(hyp_chars, ref_chars))
+                cers.append(Levenshtein.distance(hyp_chars, ref_chars))
                 char_ref_lens.append(len(ref_chars))
 
         cer_ctc = float(sum(cers)) / sum(char_ref_lens) if cers else None
@@ -214,16 +214,16 @@ def calculate_cer(self, seqs_hat, seqs_true):
         :return: average sentence-level CER score
         :rtype float
         """
-        import editdistance
 
         char_eds, char_ref_lens = [], []
         for i, seq_hat_text in enumerate(seqs_hat):
             seq_true_text = seqs_true[i]
             hyp_chars = seq_hat_text.replace(" ", "")
             ref_chars = seq_true_text.replace(" ", "")
-            char_eds.append(editdistance.eval(hyp_chars, ref_chars))
+            char_eds.append(Levenshtein.distance(hyp_chars, ref_chars))
             char_ref_lens.append(len(ref_chars))
-        return float(sum(char_eds)) / sum(char_ref_lens)
+        ref_len = sum(char_ref_lens)
+        return float(sum(char_eds)) / ref_len if ref_len > 0 else None
 
     def calculate_wer(self, seqs_hat, seqs_true):
         """Calculate sentence-level WER score.
@@ -233,13 +233,13 @@ def calculate_wer(self, seqs_hat, seqs_true):
         :return: average sentence-level WER score
         :rtype float
         """
-        import editdistance
 
         word_eds, word_ref_lens = [], []
         for i, seq_hat_text in enumerate(seqs_hat):
             seq_true_text = seqs_true[i]
             hyp_words = seq_hat_text.split()
             ref_words = seq_true_text.split()
-            word_eds.append(editdistance.eval(hyp_words, ref_words))
+            word_eds.append(Levenshtein.distance(hyp_words, ref_words))
             word_ref_lens.append(len(ref_words))
-        return float(sum(word_eds)) / sum(word_ref_lens)
+        ref_len = sum(word_ref_lens)
+        return float(sum(word_eds)) / ref_len if ref_len > 0 else None

setup.py

@@ -35,7 +35,7 @@
         "jaconv",
         # Speaker & evaluation
         "umap_learn",
-        "editdistance>=0.5.2",
+        "rapidfuzz>=3.0.0",
         # Optional (training/enhancement)
         "torch_complex",
         "tensorboardX",
@@ -44,7 +44,7 @@
     ],
     # train: The modules invoked when training only.
     "train": [
-        "editdistance",
+        "rapidfuzz>=3.0.0",
     ],
     # all: The modules should be optionally installled due to some reason.
     #      Please consider moving them to "install" occasionally

tests/test_metrics_common_rapidfuzz.py

@@ -0,0 +1,72 @@
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+
+def _install_fake_rapidfuzz():
+    rapidfuzz = types.ModuleType("rapidfuzz")
+    distance = types.ModuleType("rapidfuzz.distance")
+
+    class Levenshtein:
+        @staticmethod
+        def distance(left, right):
+            left = list(left)
+            right = list(right)
+            dp = list(range(len(right) + 1))
+            for i, lval in enumerate(left, 1):
+                prev = dp[0]
+                dp[0] = i
+                for j, rval in enumerate(right, 1):
+                    old = dp[j]
+                    dp[j] = min(
+                        dp[j] + 1,
+                        dp[j - 1] + 1,
+                        prev + (0 if lval == rval else 1),
+                    )
+                    prev = old
+            return dp[-1]
+
+    distance.Levenshtein = Levenshtein
+    rapidfuzz.distance = distance
+    sys.modules.setdefault("rapidfuzz", rapidfuzz)
+    sys.modules.setdefault("rapidfuzz.distance", distance)
+
+
+def _load_metrics_common():
+    _install_fake_rapidfuzz()
+    module_path = Path(__file__).resolve().parents[1] / "funasr" / "metrics" / "common.py"
+    spec = importlib.util.spec_from_file_location("metrics_common_probe", module_path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules["metrics_common_probe"] = module
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_cer_and_wer_return_none_when_reference_length_is_zero():
+    common = _load_metrics_common()
+    calc = common.ErrorCalculator(
+        char_list=["<blank>", "a", "b", "c", "h", "e", "l", "o", "w", "r", "d", " "],
+        sym_space=" ",
+        sym_blank="<blank>",
+        report_cer=True,
+        report_wer=True,
+    )
+
+    assert calc.calculate_cer(["abc"], ["   "]) is None
+    assert calc.calculate_wer(["hello world"], ["   "]) is None
+
+
+def test_cer_and_wer_keep_normal_levenshtein_semantics():
+    common = _load_metrics_common()
+    calc = common.ErrorCalculator(
+        char_list=["<blank>", "a", "b", "c", "x", "y", " "],
+        sym_space=" ",
+        sym_blank="<blank>",
+        report_cer=True,
+        report_wer=True,
+    )
+
+    assert calc.calculate_cer(["abc"], ["axc"]) == 1 / 3
+    assert calc.calculate_wer(["foo bar baz"], ["foo qux baz"]) == 1 / 3