utils(string): add new function parse_start_end_tags()

src/pipecat/utils/string.py

@@ -5,6 +5,7 @@
 #
 
 import re
+from typing import Optional, Sequence, Tuple
 
 ENDOFSENTENCE_PATTERN_STR = r"""
     (?<![A-Z])       # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
@@ -25,8 +26,23 @@ EMAIL_PATTERN = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
 
 NUMBER_PATTERN = re.compile(r"[+-]?(\d+(\.\d*)?|\.\d+)([eE][+-]?\d+)?")
 
+StartEndTags = Tuple[str, str]
+
 
 def replace_match(text: str, match: re.Match, old: str, new: str) -> str:
+    """Replace occurrences of a substring within a matched section of a given
+    text.
+
+    Args:
+        text (str): The input text in which replacements will be made.
+        match (re.Match): A regex match object representing the section of text to modify.
+        old (str): The substring to be replaced.
+        new (str): The substring to replace `old` with.
+
+    Returns:
+        str: The modified text with the specified replacements made within the matched section.
+
+    """
     start = match.start()
     end = match.end()
     replacement = text[start:end].replace(old, new)
@@ -35,6 +51,20 @@ def replace_match(text: str, match: re.Match, old: str, new: str) -> str:
 
 
 def match_endofsentence(text: str) -> int:
+    """Finds the position of the end of a sentence in the provided text string.
+
+    This function processes the input text by replacing periods in email
+    addresses and numbers with ampersands to prevent them from being
+    misidentified as sentence terminals. It then searches for the end of a
+    sentence using a specified regex pattern.
+
+    Args:
+        text (str): The input text in which to find the end of the sentence.
+
+    Returns:
+        int: The position of the end of the sentence if found, otherwise 0.
+
+    """
     text = text.rstrip()
 
     # Replace email dots by ampersands so we can find the end of sentence. For
@@ -52,3 +82,49 @@ def match_endofsentence(text: str) -> int:
     match = ENDOFSENTENCE_PATTERN.search(text)
 
     return match.end() if match else 0
+
+
+def parse_start_end_tags(
+    text: str,
+    tags: Sequence[StartEndTags],
+    current_tag: Optional[StartEndTags],
+    current_tag_index: int,
+) -> Tuple[Optional[StartEndTags], int]:
+    """Parses the given text to identify a pair of start/end tags.
+
+    If a start tag was previously found (i.e. current_tags is valid), wait for
+    the corresponding end tag.  Otherwise, wait for a start tag.
+
+    This function will return the index in the text that we should start parsing
+    in the next call and the current or new tags.
+
+    Parameters:
+    - text (str): The text to be parsed.
+    - tags (Sequence[StartEndTags]): List of tuples containing start and end tags.
+    - current_tags (Optional[StartEndTags]): The currently active tags, if any.
+    - current_tags_index (int): The current index in the text.
+
+    Returns:
+    Tuple[Optional[StartEndTags], int]: A tuple containing None or the current
+    tag and the index of the text.
+
+    """
+    # If we are already inside a tag, check if the end tag is in the text.
+    if current_tag:
+        _, end_tag = current_tag
+        if end_tag in text[current_tag_index:]:
+            return (None, len(text))
+        return (current_tag, current_tag_index)
+
+    # Check if any start tag appears in the text
+    for start_tag, end_tag in tags:
+        start_tag_count = text[current_tag_index:].count(start_tag)
+        end_tag_count = text[current_tag_index:].count(end_tag)
+        if start_tag_count == 0 and end_tag_count == 0:
+            return (None, current_tag_index)
+        elif start_tag_count > end_tag_count:
+            return ((start_tag, end_tag), len(text))
+        elif start_tag_count == end_tag_count:
+            return (None, len(text))
+
+    return (None, current_tag_index)

tests/test_utils_string.py

@@ -6,7 +6,7 @@
 
 import unittest
 
-from pipecat.utils.string import match_endofsentence
+from pipecat.utils.string import match_endofsentence, parse_start_end_tags
 
 
 class TestUtilsString(unittest.IsolatedAsyncioTestCase):
@@ -23,6 +23,7 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
         assert match_endofsentence("My emails are foo@pipecat.ai and bar@pipecat.ai.") == 48
         assert match_endofsentence("My email is foo.bar@pipecat.ai.") == 31
         assert match_endofsentence("My email is spell(foo.bar@pipecat.ai).") == 38
+        assert match_endofsentence("My email is <spell>foo.bar@pipecat.ai</spell>.") == 46
         assert match_endofsentence("The number pi is 3.14159.") == 25
         assert match_endofsentence("Valid scientific notation 1.23e4.") == 33
         assert match_endofsentence("Valid scientific notation 0.e4.") == 31
@@ -60,3 +61,50 @@ class TestUtilsString(unittest.IsolatedAsyncioTestCase):
         for i in hindi_sentences:
             assert match_endofsentence(i)
         assert not match_endofsentence("हैलो，")
+
+
+class TestStartEndTags(unittest.IsolatedAsyncioTestCase):
+    async def test_empty(self):
+        assert parse_start_end_tags("", [], None, 0) == (None, 0)
+        assert parse_start_end_tags("Hello from Pipecat!", [], None, 0) == (None, 0)
+
+    async def test_simple(self):
+        # (<a>, </a>)
+        assert parse_start_end_tags("Hello from <a>Pipecat</a>!", [("<a>", "</a>")], None, 0) == (
+            None,
+            26,
+        )
+        assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 0) == (
+            ("<a>", "</a>"),
+            21,
+        )
+        assert parse_start_end_tags("Hello from <a>Pipecat", [("<a>", "</a>")], None, 6) == (
+            ("<a>", "</a>"),
+            21,
+        )
+
+        # (spell(, ))
+        assert parse_start_end_tags("Hello from spell(Pipecat)!", [("spell(", ")")], None, 0) == (
+            None,
+            26,
+        )
+        assert parse_start_end_tags("Hello from spell(Pipecat", [("spell(", ")")], None, 0) == (
+            ("spell(", ")"),
+            24,
+        )
+
+    async def test_multiple(self):
+        # (<a>, </a>)
+        assert parse_start_end_tags(
+            "Hello from <a>Pipecat</a>! Hello <a>World</a>!", [("<a>", "</a>")], None, 0
+        ) == (
+            None,
+            46,
+        )
+
+        assert parse_start_end_tags(
+            "Hello from <a>Pipecat</a>! Hello <a>World", [("<a>", "</a>")], None, 0
+        ) == (
+            ("<a>", "</a>"),
+            41,
+        )