Renaming the example to p2p-webrtc

examples/p2p-webrtc/video-transform/server/bot.py

@@ -0,0 +1,156 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+import os
+import sys
+
+import cv2
+import numpy as np
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame, InputImageRawFrame, OutputImageRawFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
+from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+class EdgeDetectionProcessor(FrameProcessor):
+    def __init__(self, camera_out_width, camera_out_height: int):
+        super().__init__()
+        self._camera_out_width = camera_out_width
+        self._camera_out_height = camera_out_height
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, InputImageRawFrame):
+            # Convert bytes to NumPy array
+            img = np.frombuffer(frame.image, dtype=np.uint8).reshape(
+                (frame.size[1], frame.size[0], 3)
+            )
+
+            # perform edge detection
+            img = cv2.cvtColor(cv2.Canny(img, 100, 200), cv2.COLOR_GRAY2BGR)
+
+            # convert the size if needed
+            desired_size = (self._camera_out_width, self._camera_out_height)
+            if frame.size != desired_size:
+                resized_image = cv2.resize(img, desired_size)
+                frame = OutputImageRawFrame(resized_image.tobytes(), desired_size, frame.format)
+                await self.push_frame(frame)
+            else:
+                await self.push_frame(
+                    OutputImageRawFrame(image=img.tobytes(), size=frame.size, format=frame.format)
+                )
+        else:
+            await self.push_frame(frame, direction)
+
+
+SYSTEM_INSTRUCTION = f"""
+"You are Gemini Chatbot, a friendly, helpful robot.
+
+Your goal is to demonstrate your capabilities in a succinct way.
+
+Your output will be converted to audio so don't include special characters in your answers.
+
+Respond to what the user said in a creative and helpful way. Keep your responses brief. One or two sentences at most.
+"""
+
+
+async def run_bot(webrtc_connection):
+    transport_params = TransportParams(
+        camera_in_enabled=True,
+        camera_out_enabled=True,
+        camera_out_is_live=True,
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+        vad_audio_passthrough=True,
+    )
+
+    pipecat_transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection, params=transport_params
+    )
+
+    llm = GeminiMultimodalLiveLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        voice_id="Puck",  # Aoede, Charon, Fenrir, Kore, Puck
+        transcribe_user_audio=True,
+        transcribe_model_audio=True,
+        system_instruction=SYSTEM_INSTRUCTION,
+    )
+
+    context = OpenAILLMContext(
+        [
+            {
+                "role": "user",
+                "content": "Start by greeting the user warmly and introducing yourself.",
+            }
+        ],
+    )
+    context_aggregator = llm.create_context_aggregator(context)
+
+    # RTVI events for Pipecat client UI
+    rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
+
+    pipeline = Pipeline(
+        [
+            pipecat_transport.input(),
+            context_aggregator.user(),
+            rtvi,
+            llm,  # LLM
+            EdgeDetectionProcessor(
+                transport_params.camera_out_width, transport_params.camera_out_height
+            ),  # Sending the video back to the user
+            pipecat_transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            observers=[RTVIObserver(rtvi)],
+        ),
+    )
+
+    @rtvi.event_handler("on_client_ready")
+    async def on_client_ready(rtvi):
+        logger.info("Pipecat client ready.")
+        await rtvi.set_bot_ready()
+
+    @pipecat_transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info("Pipecat Client connected")
+        # Kick off the conversation.
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @pipecat_transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info("Pipecat Client disconnected")
+
+    @pipecat_transport.event_handler("on_client_closed")
+    async def on_client_closed(transport, client):
+        logger.info("Pipecat Client closed")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=False)
+
+    await runner.run(task)

examples/p2p-webrtc/video-transform/server/env.example

@@ -0,0 +1 @@
+GOOGLE_API_KEY=

examples/p2p-webrtc/video-transform/server/requirements.txt

@@ -0,0 +1,6 @@
+python-dotenv
+fastapi[all]
+uvicorn
+aiortc
+opencv-python
+pipecat-ai[google,silero]

examples/p2p-webrtc/video-transform/server/server.py

@@ -0,0 +1,77 @@
+import argparse
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from typing import Dict
+
+import uvicorn
+from bot import run_bot
+from dotenv import load_dotenv
+from fastapi import BackgroundTasks, FastAPI
+
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+
+# Load environment variables
+load_dotenv(override=True)
+
+logger = logging.getLogger("pc")
+
+app = FastAPI()
+
+# Store connections by pc_id
+pcs_map: Dict[str, SmallWebRTCConnection] = {}
+
+ice_servers = ["stun:stun.l.google.com:19302"]
+
+
+@app.post("/api/offer")
+async def offer(request: dict, background_tasks: BackgroundTasks):
+    pc_id = request.get("pc_id")
+
+    if pc_id and pc_id in pcs_map:
+        pipecat_connection = pcs_map[pc_id]
+        logger.info(f"Reusing existing connection for pc_id: {pc_id}")
+        await pipecat_connection.renegotiate(
+            sdp=request["sdp"], type=request["type"], restart_pc=request.get("restart_pc", False)
+        )
+    else:
+        pipecat_connection = SmallWebRTCConnection(ice_servers)
+        await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"])
+
+        @pipecat_connection.on("closed")
+        async def handle_disconnected(webrtc_connection: SmallWebRTCConnection):
+            logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}")
+            pcs_map.pop(webrtc_connection.pc_id, None)
+
+        background_tasks.add_task(run_bot, pipecat_connection)
+
+    answer = pipecat_connection.get_answer()
+    # Updating the peer connection inside the map
+    pcs_map[answer["pc_id"]] = pipecat_connection
+
+    return answer
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    yield  # Run app
+    coros = [pc.close() for pc in pcs_map.values()]
+    await asyncio.gather(*coros)
+    pcs_map.clear()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="WebRTC demo")
+    parser.add_argument("--host", default="0.0.0.0", help="Host for HTTP server (default: 0.0.0.0)")
+    parser.add_argument(
+        "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)"
+    )
+    parser.add_argument("--verbose", "-v", action="count")
+    args = parser.parse_args()
+
+    if args.verbose:
+        logging.basicConfig(level=logging.DEBUG)
+    else:
+        logging.basicConfig(level=logging.INFO)
+
+    uvicorn.run(app, host=args.host, port=args.port)