From 0ddfa3de5be7fe3b66efadc14f00d4eb854c2c62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 15:36:18 -0700 Subject: [PATCH] move WakeCheckFilter to processors/filters --- examples/foundational/14-wake-phrase.py | 77 +---------------- .../{filter.py => filters/frame_filter.py} | 2 +- .../processors/filters/wake_check_filter.py | 82 +++++++++++++++++++ src/pipecat/processors/frame_processor.py | 2 +- 4 files changed, 86 insertions(+), 77 deletions(-) rename src/pipecat/processors/{filter.py => filters/frame_filter.py} (96%) create mode 100644 src/pipecat/processors/filters/wake_check_filter.py diff --git a/examples/foundational/14-wake-phrase.py b/examples/foundational/14-wake-phrase.py index b6d19116c..283768460 100644 --- a/examples/foundational/14-wake-phrase.py +++ b/examples/foundational/14-wake-phrase.py @@ -8,14 +8,8 @@ import asyncio import aiohttp import os import sys -import re -import time -from enum import Enum -from pipecat.frames.frames import ( - Frame, SystemFrame, TranscriptionFrame, TextFrame -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.filters.wake_check_filter import WakeCheckFilter from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -37,73 +31,6 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -class WakeCheckFilter(FrameProcessor): - """ - This filter looks for wake phrases in the transcription frames and only passes through frames - after a wake phrase has been detected. It also has a keepalive timeout to allow for a brief - period of continued conversation after a wake phrase has been detected. - """ - class WakeState(Enum): - IDLE = 1 - AWAKE = 2 - - class ParticipantState: - def __init__(self, participant_id: str): - self.participant_id = participant_id - self.state = WakeCheckFilter.WakeState.IDLE - self.wake_timer = 0 - self.accumulator = "" - - def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 2): - super().__init__() - self._participant_states = {} - self._keepalive_timeout = keepalive_timeout - self._wake_patterns = [] - for name in wake_phrases: - pattern = re.compile(r'\b' + r'\s*'.join(re.escape(word) - for word in name.split()) + r'\b', re.IGNORECASE) - self._wake_patterns.append(pattern) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - try: - if isinstance(frame, TranscriptionFrame): - p = self._participant_states.get(frame.user_id) - if p is None: - p = WakeCheckFilter.ParticipantState(frame.user_id) - self._participant_states[frame.user_id] = p - - # If we have been AWAKE within the last keepalive_timeout seconds, pass - # the frame through - if p.state == WakeCheckFilter.WakeState.AWAKE: - if time.time() - p.wake_timer < self._keepalive_timeout: - logger.debug( - f"Wake phrase keepalive timeout has not expired. Passing frame through.") - p.wake_timer = time.time() - await self.push_frame(frame) - return - else: - p.state = WakeCheckFilter.WakeState.IDLE - - p.accumulator += frame.text - for pattern in self._wake_patterns: - match = pattern.search(p.accumulator) - if match: - logger.debug(f"Wake phrase triggered: {match.group()}") - # Found the wake word. Discard from the accumulator up to the start of the match - # and modify the frame in place. - p.state = WakeCheckFilter.WakeState.AWAKE - p.wake_timer = time.time() - frame.text = p.accumulator[match.start():] - p.accumulator = "" - await self.push_frame(frame) - else: - pass - else: - await self.push_frame(frame, direction) - except Exception as e: - logger.error(f"Error in wake word filter: {e}") - - async def main(room_url: str, token): async with aiohttp.ClientSession() as session: @@ -152,7 +79,7 @@ async def main(room_url: str, token): task = PipelineTask(pipeline, allow_interruptions=True) - @ transport.event_handler("on_first_participant_joined") + @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): transport.capture_participant_transcription(participant["id"]) await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.") diff --git a/src/pipecat/processors/filter.py b/src/pipecat/processors/filters/frame_filter.py similarity index 96% rename from src/pipecat/processors/filter.py rename to src/pipecat/processors/filters/frame_filter.py index 0f7026b4a..42f0e9dd8 100644 --- a/src/pipecat/processors/filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -10,7 +10,7 @@ from pipecat.frames.frames import AppFrame, ControlFrame, Frame, SystemFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -class Filter(FrameProcessor): +class FrameFilter(FrameProcessor): def __init__(self, types: List[type]): super().__init__() diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py new file mode 100644 index 000000000..459ee5f75 --- /dev/null +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -0,0 +1,82 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import re +import time + +from enum import Enum + +from pipecat.frames.frames import Frame, TranscriptionFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +from loguru import logger + + +class WakeCheckFilter(FrameProcessor): + """ + This filter looks for wake phrases in the transcription frames and only passes through frames + after a wake phrase has been detected. It also has a keepalive timeout to allow for a brief + period of continued conversation after a wake phrase has been detected. + """ + class WakeState(Enum): + IDLE = 1 + AWAKE = 2 + + class ParticipantState: + def __init__(self, participant_id: str): + self.participant_id = participant_id + self.state = WakeCheckFilter.WakeState.IDLE + self.wake_timer = 0.0 + self.accumulator = "" + + def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 2): + super().__init__() + self._participant_states = {} + self._keepalive_timeout = keepalive_timeout + self._wake_patterns = [] + for name in wake_phrases: + pattern = re.compile(r'\b' + r'\s*'.join(re.escape(word) + for word in name.split()) + r'\b', re.IGNORECASE) + self._wake_patterns.append(pattern) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + try: + if isinstance(frame, TranscriptionFrame): + p = self._participant_states.get(frame.user_id) + if p is None: + p = WakeCheckFilter.ParticipantState(frame.user_id) + self._participant_states[frame.user_id] = p + + # If we have been AWAKE within the last keepalive_timeout seconds, pass + # the frame through + if p.state == WakeCheckFilter.WakeState.AWAKE: + if time.time() - p.wake_timer < self._keepalive_timeout: + logger.debug( + "Wake phrase keepalive timeout has not expired. Passing frame through.") + p.wake_timer = time.time() + await self.push_frame(frame) + return + else: + p.state = WakeCheckFilter.WakeState.IDLE + + p.accumulator += frame.text + for pattern in self._wake_patterns: + match = pattern.search(p.accumulator) + if match: + logger.debug(f"Wake phrase triggered: {match.group()}") + # Found the wake word. Discard from the accumulator up to the start of the match + # and modify the frame in place. + p.state = WakeCheckFilter.WakeState.AWAKE + p.wake_timer = time.time() + frame.text = p.accumulator[match.start():] + p.accumulator = "" + await self.push_frame(frame) + else: + pass + else: + await self.push_frame(frame, direction) + except Exception as e: + logger.error(f"Error in wake word filter: {e}") diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index a79352f7a..3bb750218 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,7 +8,7 @@ import asyncio from asyncio import AbstractEventLoop from enum import Enum -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame +from pipecat.frames.frames import ErrorFrame, Frame from pipecat.utils.utils import obj_count, obj_id from loguru import logger