LipSync Live

pip install decart

import asyncio
import fractions
import numpy as np
import os
from decart.lipsync import RealtimeLipsyncClient

async def main():

    frames_queue = asyncio.Queue()
    audio_queue = asyncio.Queue()

    client = RealtimeLipsyncClient(api_key=os.getenv("DECART_API_KEY"))
    frame_interval = fractions.Fraction(1, client._video_fps)
    time = asyncio.get_running_loop().time

    while True:
        frame_start_time = time()
        try:
            # Get audio clip of any duration when available, should be PCM 16000Hz Mono channel, could be any length
            audio_clip: bytes = audio_queue.get_nowait()
            await client.send_audio(audio_clip)
        except asyncio.QueueEmpty:
            pass

        # Get next frame - raw image data of shape (800, 450, 3), RGB format
        frame: np.ndarray = await frames_queue.get()
        await client.send_video_frame(frame)

        try:
            # Wait for resulting frame and corresponding audio clip if already available
            out_frame, out_audio_clip = await client.get_synced_output(
                timeout=frame_interval - (time() - frame_start_time)
            )

            # HANDLE OUT_FRAME AND OUT_AUDIO_CLIP HERE
        except asyncio.TimeoutError:
            pass

        # Wait before sending next frame
        await asyncio.sleep(frame_interval - (time() - frame_start_time))


if __name__ == "__main__":
    asyncio.run(main())