Realtime encoding for livestream

[www2000]

Hi i working on a application for a livestream that need handle the encoding in real time.

the concept is that i generate the animation in Cairo and than pass thrue to ffmpeg using pillow but the that i have is that i can not generate this in real time in my test code this is to fast.

I know that ffmpeg have a realtime switch but i don't know how is done using pyav

my test code that i wand to stream

import av.container
import numpy as np
import cairo
import av
import PIL.Image as Image
import math
from time import time
from datetime import datetime, UTC

duration = 10
fps = 25
total_frames = duration * fps

ytURL = "rtmp://a.rtmp.youtube.com/live2/{youtube-streaming-api-key}"

# container = av.open(ytURL, mode="w", format='flv', options={"re"}) # livestreaming sink current disable
container = av.open("test3.mp4", "w", options={"re":""})  # file sink
width, height = 1920, 1080
options = {
    "preset": "superfast",
    "hwaccel": "cuda"
}
stream = container.add_stream("h264", rate=fps, options=options)
stream.width = width
stream.height = height
stream.pix_fmt = "yuv420p"

# confert cairo canvas to PILOW
def to_pil(surface: cairo.ImageSurface) -> Image:
    format = surface.get_format()
    size = (surface.get_width(), surface.get_height())
    stride = surface.get_stride()

    with surface.get_data() as memory:
        if format == cairo.Format.RGB24:
            return Image.frombuffer(
                "RGB", size, memory.tobytes(),
                'raw', "BGRX", stride)
        elif format == cairo.Format.ARGB32:
            return Image.frombuffer(
                "RGBA", size, memory.tobytes(),
                'raw', "BGRa", stride)
        else:
            raise NotImplementedError(repr(format))

# draw test scene to cairo canvas
def draw(ctx, frame, total_frames):
    fOverTf = frame / total_frames # location of frame
    
    # set background color
    r = (0.5 + 0.5 * np.sin(2 * np.pi * (0 / 3 + fOverTf)))
    g = (0.5 + 0.5 * np.sin(2 * np.pi * (1 / 3 + fOverTf)))
    b = (0.5 + 0.5 * np.sin(2 * np.pi * (2 / 3 + fOverTf)))
    ctx.set_source_rgba(r, g, b, 1)
    ctx.set_line_width(10)
    ctx.move_to(0, 0) 
    ctx.rectangle(0, 0, 1920, 1080)
    ctx.fill()
    
    # rotation line
    cx = width / 2
    cy = height / 2
    ctx.set_source_rgba(0, 0, 0, 1)
    ctx.set_line_width(12)
    ctx.set_line_cap(cairo.LINE_CAP_BUTT)
    ctx.move_to(cx, cy)
    
    fr = (frame / 25) * np.pi * 2
    xcos = np.cos(fr) * 200
    xsin = np.sin(fr) * 200
    ctx.line_to(cx + xsin, cy - xcos)
    ctx.stroke()
    
    # print frame number and gmt time on canvas
    ctx.set_font_size(40)
    ctx.select_font_face( 
        "Arial", cairo.FONT_SLANT_NORMAL, cairo.FONT_WEIGHT_NORMAL) 
    ctx.move_to(50, 50)
    ctx.show_text(str(frame))
    ctx.move_to(50,100)
    ctx.show_text(datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S'))
    ctx.stroke()

# start frame loop
start = time()
for frame_i in range(total_frames):
    

    surface = cairo.ImageSurface (cairo.FORMAT_RGB24, width, height)
    ctx = cairo.Context (surface)
    draw(ctx,frame_i , total_frames)
    
    im = to_pil(surface)    

    frame = av.VideoFrame.from_image(im)
    for packet in stream.encode(frame):
        container.mux(packet)

# end frame loop and calulate working time
end = time()
print(end-start)
        

# Flush stream
for packet in stream.encode():
    container.mux(packet)

# Close the file
container.close()

ffmpeg commandline that i current use for testing livestream to youtube

ffmpeg -re -loop 1 -i staticimage.png -i music.mp3 -shortest -c:v libx264 -preset slow -crf 18 -vf scale=out_color_matrix=bt709 -color_primaries bt709 -color_trc bt709 -colorspace bt709 -c:a aac -ar 48000 -ac 2 -b:a 320k -profile:v high -level 4.0 -bf 2 -coder 1 -pix_fmt yuv420p -b:v 1M -threads 4 -cpu-used 0 -r 25 -g 15 -movflags +faststart -f flv rtmp://a.rtmp.youtube.com/live2/{youtube-streaming-api-key}

[kenwaytis]

The FFmpeg -re flag emulates a real-time device read. PyAV doesn't have a direct equivalent because it expects the code generating the frames to manage its own timing (Or at least idk).

I can offer a solution: a token bucket. The principle is all over the web; it's basically using time calculation and a time-driven approach to achieve precise rate control. Simple and effective.

Here's an implementation.

import time
import threading

class TokenBucket:
    """
    An implementation of the token bucket algorithm to maintain a nearly
    constant consumption rate (FPS).
    """

    def __init__(self, fps: float, capacity: int = 3):
        """
        Initializes the TokenBucket.

        :param fps: The rate of token generation (frames per second).
        :param capacity: The maximum number of tokens the bucket can hold.
                         This allows for short bursts.
        """
        self.period = 1.0 / fps
        self.capacity = float(capacity)
        self.tokens = float(capacity)  # Start with a full bucket.
        self.last_check = time.perf_counter()
        self._lock = threading.Lock()

    def _add_new_tokens(self) -> None:
        now = time.perf_counter()
        elapsed = now - self.last_check
      
        new_tokens = elapsed / self.period
        if new_tokens > 0:
            self.tokens = min(self.capacity, self.tokens + new_tokens)
            self.last_check = now

    def consume(self) -> None:
        """
        Consume one token. If no tokens are available, this method will
        block until one is.
        """
        while True:
            with self._lock:
                self._add_new_tokens()
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
          
            # Sleep for a fraction of the period to avoid busy-waiting.
            time.sleep(self.period * 0.5)

Integrate it into your main loop.

# ... (setup code)

bucket = TokenBucket(fps=fps)

# start frame loop
start = time()
for frame_i in range(total_frames):

    bucket.consume()
  
    surface = cairo.ImageSurface (cairo.FORMAT_RGB24, width, height)
    ctx = cairo.Context (surface)
    draw(ctx,frame_i , total_frames)

    im = to_pil(surface)

    frame = av.VideoFrame.from_image(im)
    for packet in stream.encode(frame):
        container.mux(packet)

# ... (cleanup code)

Happy streaming by the way