From 535514f506874102a1e9ed17b55d845e7756209e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Mon, 1 Jul 2024 14:58:01 -0700
Subject: [PATCH] processors: added new UserIdleProcessor

---
 CHANGELOG.md                                  |  4 +
 src/pipecat/processors/user_idle_processor.py | 77 +++++++++++++++++++
 2 files changed, 81 insertions(+)
 create mode 100644 src/pipecat/processors/user_idle_processor.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e0710a03c..6ca777718 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `UserIdleProcessor`. This processor can be used to wait for any
+  interaction with the user. If the user doesn't say anything within a given
+  timeout a provided callback is called.
+
 - Added `IdleFrameProcessor`. This processor can be used to wait for frames
   within a given timeout. If no frame is received within the timeout a provided
   callback is called.
diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py
new file mode 100644
index 000000000..20e8e7be6
--- /dev/null
+++ b/src/pipecat/processors/user_idle_processor.py
@@ -0,0 +1,77 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+
+from typing import Awaitable, Callable
+
+from pipecat.frames.frames import BotSpeakingFrame, Frame, StartInterruptionFrame, StopInterruptionFrame, SystemFrame
+from pipecat.processors.async_frame_processor import AsyncFrameProcessor
+from pipecat.processors.frame_processor import FrameDirection
+
+
+class UserIdleProcessor(AsyncFrameProcessor):
+    """This class is useful to check if the user is interacting with the bot
+    within a given timeout. If the timeout is reached before any interaction
+    occurred the provided callback will be called.
+
+    The callback can then be used to push frames downstream by using
+    `queue_frame()` (or `push_frame()` for system frames).
+
+    """
+
+    def __init__(
+            self,
+            *,
+            callback: Callable[["UserIdleProcessor"], Awaitable[None]],
+            timeout: float,
+            **kwargs):
+        super().__init__(**kwargs)
+
+        self._callback = callback
+        self._timeout = timeout
+
+        self._interrupted = False
+
+        self._create_idle_task()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+        else:
+            await self.queue_frame(frame, direction)
+
+        # We shouldn't call the idle callback if the user or the bot are speaking.
+        if isinstance(frame, StartInterruptionFrame):
+            self._interrupted = True
+            self._idle_event.set()
+        elif isinstance(frame, StopInterruptionFrame):
+            self._interrupted = False
+            self._idle_event.set()
+        elif isinstance(frame, BotSpeakingFrame):
+            self._idle_event.set()
+
+    async def cleanup(self):
+        self._idle_task.cancel()
+        await self._idle_task
+
+    def _create_idle_task(self):
+        self._idle_event = asyncio.Event()
+        self._idle_task = self.get_event_loop().create_task(self._idle_task_handler())
+
+    async def _idle_task_handler(self):
+        while True:
+            try:
+                await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout)
+            except asyncio.TimeoutError:
+                if not self._interrupted:
+                    await self._callback(self)
+            except asyncio.CancelledError:
+                break
+            finally:
+                self._idle_event.clear()