diff --git a/simulstream/server/speech_processors/base_streamatt.py b/simulstream/server/speech_processors/base_streamatt.py index fa7ccd4..3d31ed8 100644 --- a/simulstream/server/speech_processors/base_streamatt.py +++ b/simulstream/server/speech_processors/base_streamatt.py @@ -297,6 +297,27 @@ def select_text_history(self, text_history: List[str]): return new_history[::-1] +class FixedCharsTextHistory: + """ + Character-count-based textual history selection method that retains a pre-defined number of + tokens in the history (*history_chars*). + + Recommended for character-level languages (e.g., Chinese, Japanese) where word-boundary + markers (▁) are sparse, making :class:`FixedWordsTextHistory` ineffective: when few tokens + carry a BOW prefix, the word counter never reaches *history_words*, so the history is never + trimmed and the audio history grows without bound, causing AlignAtt to cut all new tokens. + + Args: + config (SimpleNamespace): Configuration object with an optional attribute: + - **history_chars (int)**: Number of tokens to retain. Defaults to 20. + """ + def __init__(self, config: SimpleNamespace): + self.history_chars = getattr(config, "history_chars", 20) + + def select_text_history(self, text_history: List[str]) -> List[str]: + return text_history[-self.history_chars:] + + class PunctuationTextHistory: """ Punctuation textual history selection method that retains the sentence diff --git a/uts/speech_processors/test_streamatt.py b/uts/speech_processors/test_streamatt.py index 180c408..a20878f 100644 --- a/uts/speech_processors/test_streamatt.py +++ b/uts/speech_processors/test_streamatt.py @@ -15,7 +15,45 @@ import unittest from types import SimpleNamespace -from simulstream.server.speech_processors.base_streamatt import PunctuationTextHistory +from simulstream.server.speech_processors.base_streamatt import ( + FixedCharsTextHistory, + FixedWordsTextHistory, + PunctuationTextHistory, +) + + +class TestFixedWordsTextHistory(unittest.TestCase): + def setUp(self): + self.history = FixedWordsTextHistory(SimpleNamespace(history_words=3)) + + def test_word_level(self): + """ Trims history to the last 3 BOW-started words in space-separated languages. """ + en_history = ["▁I", "▁am", "▁going", "▁to", "▁New", "▁York"] + self.assertEqual( + self.history.select_text_history(en_history), + ["▁to", "▁New", "▁York"]) + + def test_word_level_with_subwords(self): + """ Subword continuations (no ▁) are included in the retained word. """ + en_history = ["▁inter", "nation", "al", "▁meet", "ing", "▁today"] + self.assertEqual( + self.history.select_text_history(en_history), + ["▁inter", "nation", "al", "▁meet", "ing", "▁today"]) + + +class TestFixedCharsTextHistory(unittest.TestCase): + def setUp(self): + self.history = FixedCharsTextHistory(SimpleNamespace(history_chars=3)) + + def test_char_level(self): + """ Keeps the last 3 tokens for character-level languages. """ + zh_history = ['大', '家', '好', '我', '在', '谈', '论'] + self.assertEqual(self.history.select_text_history(zh_history), ['在', '谈', '论']) + + def test_shorter_than_limit(self): + """ Returns the full history when it is shorter than history_chars. """ + zh_history = ['大', '家'] + self.assertEqual(self.history.select_text_history(zh_history), ['大', '家']) class TestPunctuationTextHistory(unittest.TestCase):