wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update foundational examples to use the development runner

Browse Source
This commit is contained in:
Mark Backman
2025-07-31 22:00:55 -04:00
parent ff45b77fdf
commit f1b1d5b130
144 changed files with 1903 additions and 958 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
examples/foundational/18-gstreamer-filesrc.py
View File

@@ -13,11 +13,16 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.gstreamer.pipeline_source import GStreamerPipelineSource
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.services.daily import DailyParams
load_dotenv(override=True)
parser = argparse.ArgumentParser(description="Pipecat Video Streaming Bot")
parser.add_argument("-i", "--input", type=str, required=True, help="Input video file")
args = parser.parse_args()
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
@@ -40,7 +45,7 @@ transport_params = {
}
async def run_example(transport: BaseTransport, args: argparse.Namespace, handle_sigint: bool):
async def run_bot(transport: BaseTransport, args: argparse.Namespace):
logger.info(f"Starting bot with video input: {args.input}")
gst = GStreamerPipelineSource(
@@ -60,15 +65,18 @@ async def run_example(transport: BaseTransport, args: argparse.Namespace, handle
task = PipelineTask(pipeline)
runner = PipelineRunner(handle_sigint=handle_sigint)
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport)
if __name__ == "__main__":
from pipecat.examples.run import main
from pipecat.runner.run import main
parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
parser.add_argument("-i", "--input", type=str, required=True, help="Input video file")
main(run_example, parser=parser, transport_params=transport_params)
main()