PyTrickle

from pytrickle import FrameProcessor, StreamServer
from pytrickle.frames import VideoFrame, AudioFrame
from typing import Optional, List

class MyProcessor(FrameProcessor):

    async def initialize(self):
        # Load AI models or initialise any resources here.
        # Called once before the first frame arrives.
        pass

    async def process_video_async(self, frame: VideoFrame) -> Optional[VideoFrame]:
        # frame.tensor is a PyTorch tensor: [H, W, C], float32, 0-1 range
        tensor = frame.tensor.clone()
        # Apply your model, filter, or transform here.
        return frame.replace_tensor(tensor)

    async def process_audio_async(self, frame: AudioFrame) -> Optional[List[AudioFrame]]:
        # Return the frame unchanged or process it.
        return [frame]

    def update_params(self, params: dict):
        # Called when the orchestrator sends a parameter update.
        # Update instance variables to change behaviour at runtime.
        pass

async def main():
    processor = MyProcessor()
    await processor.start()  # calls initialize()

    app = StreamServer(
        frame_processor=processor,
        port=8000,
        capability_name="my-processor",
    )
    await app.run_forever()

import asyncio
asyncio.run(main())