Video Application Examples

from daily import *
from PIL import Image
import time

class SendImageApp:
    def __init__(self, image_file, framerate):
        self.__image = Image.open(image_file)
        self.__framerate = framerate
        
        # Create virtual camera matching image dimensions
        self.__camera = Daily.create_camera_device(
            "my-camera",
            width=self.__image.width,
            height=self.__image.height,
            color_format="RGB"
        )
        
        self.__client = CallClient()
    
    def send_image(self):
        sleep_time = 1.0 / self.__framerate
        image_bytes = self.__image.tobytes()
        
        while not self.__app_quit:
            self.__camera.write_frame(image_bytes)
            time.sleep(sleep_time)

python3 send_image.py -m MEETING_URL -i image.png -f 30

import gi
gi.require_version("Gst", "1.0")
gi.require_version("GstApp", "1.0")
from gi.repository import Gst, GstApp, GLib
from daily import *

VIDEO_WIDTH = 1280
VIDEO_HEIGHT = 720
AUDIO_SAMPLE_RATE = 48000
AUDIO_CHANNELS = 2

class GstApp:
    def __init__(self, filename):
        # Create virtual camera for video
        self.__camera = Daily.create_camera_device(
            "my-camera",
            width=VIDEO_WIDTH,
            height=VIDEO_HEIGHT,
            color_format="I420"
        )
        
        # Create virtual microphone for audio
        self.__microphone = Daily.create_microphone_device(
            "my-mic",
            sample_rate=AUDIO_SAMPLE_RATE,
            channels=AUDIO_CHANNELS,
            non_blocking=True
        )
        
        # Setup GStreamer pipeline
        self.__player = Gst.Pipeline.new("player")
        
        source = Gst.ElementFactory.make("filesrc", None)
        source.set_property("location", filename)
        
        decodebin = Gst.ElementFactory.make("decodebin", None)
        decodebin.connect("pad-added", self.decodebin_callback)
        
        self.__player.add(source)
        self.__player.add(decodebin)
        source.link(decodebin)
    
    def appsink_video_new_sample(self, appsink):
        buffer = appsink.pull_sample().get_buffer()
        (_, info) = buffer.map(Gst.MapFlags.READ)
        self.__camera.write_frame(info.data)
        buffer.unmap(info)
        return Gst.FlowReturn.OK
    
    def appsink_audio_new_sample(self, appsink):
        buffer = appsink.pull_sample().get_buffer()
        (_, info) = buffer.map(Gst.MapFlags.READ)
        self.__microphone.write_frames(info.data)
        buffer.unmap(info)
        return Gst.FlowReturn.OK

python3 media_player.py -m MEETING_URL -i video.mp4

import queue
import threading
from PIL import Image
from ultralytics import YOLO
from daily import *

class DailyYOLO(EventHandler):
    def __init__(self):
        self.__client = CallClient(event_handler=self)
        
        # Load YOLO model
        self.__model = YOLO("yolov8n.pt")
        self.__camera = None
        
        self.__queue = queue.Queue()
        
        # Process frames in background thread
        self.__thread = threading.Thread(target=self.process_frames)
        self.__thread.start()
    
    def on_participant_joined(self, participant):
        print(f"Participant {participant['id']} joined, analyzing frames...")
        self.__client.set_video_renderer(
            participant["id"], 
            self.on_video_frame
        )
    
    def setup_camera(self, video_frame):
        if not self.__camera:
            # Create camera matching participant video size
            self.__camera = Daily.create_camera_device(
                "camera",
                width=video_frame.width,
                height=video_frame.height,
                color_format="RGB"
            )
            self.__client.update_inputs({
                "camera": {
                    "isEnabled": True,
                    "settings": {"deviceId": "camera"}
                }
            })
    
    def process_frames(self):
        while not self.__app_quit:
            video_frame = self.__queue.get()
            
            # Convert frame to PIL Image
            image = Image.frombytes(
                "RGBA",
                (video_frame.width, video_frame.height),
                video_frame.buffer
            )
            
            # Run YOLO detection
            results = self.__model.track(image)
            
            # Draw bounding boxes and send back
            pil = Image.fromarray(results[0].plot(), mode="RGB").tobytes()
            self.__camera.write_frame(pil)
    
    def on_video_frame(self, participant_id, video_frame, video_source):
        # Process ~15 frames per second
        if time.time() - self.__time > 0.05:
            self.__time = time.time()
            self.setup_camera(video_frame)
            self.__queue.put(video_frame)

python3 yolo.py -m MEETING_URL

import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib, Gtk
import cairo
from daily import *

class DailyGtkApp(Gtk.Application):
    def __init__(self, meeting_url, participant_id, save_audio, screen_share):
        super().__init__(application_id="co.daily.DailyGtkApp")
        
        self.__client = CallClient()
        
        # Configure subscription profiles
        self.__client.update_subscription_profiles({
            "base": {
                "microphone": "subscribed",
                "camera": "unsubscribed" if screen_share else "subscribed",
                "screenVideo": "subscribed" if screen_share else "unsubscribed",
            }
        })
        
        self.__video_source = "screenVideo" if screen_share else "camera"
    
    def join(self, meeting_url, participant_id):
        if self.__save_audio:
            self.__client.set_audio_renderer(
                participant_id,
                self.on_audio_data
            )
        
        self.__client.set_video_renderer(
            participant_id,
            self.on_video_frame,
            video_source=self.__video_source,
            color_format="BGRA"
        )
        
        self.__client.join(meeting_url, completion=self.on_joined)
    
    def on_video_frame(self, participant_id, video_frame, video_source):
        self.__frame_width = video_frame.width
        self.__frame_height = video_frame.height
        self.__frame = video_frame
        self.__drawing_area.queue_draw()
    
    def drawing_area_draw(self, area, context, w, h, data):
        if self.__joined and self.__frame is not None:
            image = bytearray(self.__frame.buffer)
            width = self.__frame_width
            height = self.__frame_height
            
            stride = cairo.ImageSurface.format_stride_for_width(
                cairo.FORMAT_ARGB32, width
            )
            cairo_surface = cairo.ImageSurface.create_for_data(
                image, cairo.FORMAT_ARGB32, width, height, stride
            )
            
            # Scale to fit window
            width_ratio = float(self.__width) / float(width)
            height_ratio = float(self.__height) / float(height)
            scale_xy = min(height_ratio, width_ratio)
            
            context.scale(scale_xy, scale_xy)
            context.set_source_surface(cairo_surface)
            context.paint()

# View participant camera
python3 gtk_app.py -m MEETING_URL -p PARTICIPANT_ID

# View screen share
python3 gtk_app.py -m MEETING_URL -p PARTICIPANT_ID -s

# Save audio to file
python3 gtk_app.py -m MEETING_URL -p PARTICIPANT_ID -a

camera = Daily.create_camera_device(
    "device-name",
    width=1280,
    height=720,
    color_format="RGB"  # or "RGBA", "I420", "BGRA"
)

client.join(
    meeting_url,
    client_settings={
        "inputs": {
            "camera": {
                "isEnabled": True,
                "settings": {"deviceId": "my-camera"}
            }
        },
        "publishing": {
            "camera": {
                "isPublishing": True,
                "sendSettings": {
                    "encodings": {
                        "low": {
                            "maxBitrate": 1000000,
                            "maxFramerate": 30.0,
                            "scaleResolutionDownBy": 1.0,
                        }
                    }
                }
            }
        }
    }
)

def on_video_frame(participant_id, video_frame, video_source):
    # video_frame.width, video_frame.height
    # video_frame.buffer - raw frame data
    # video_source - "camera" or "screenVideo"
    pass

client.set_video_renderer(
    participant_id,
    on_video_frame,
    video_source="camera",  # or "screenVideo"
    color_format="RGBA"
)

# From PIL Image
image = Image.open("frame.png")
camera.write_frame(image.tobytes())

# From numpy array
import numpy as np
frame = np.zeros((720, 1280, 3), dtype=np.uint8)
camera.write_frame(frame.tobytes())

# From raw bytes
frame_data = b"..."  # Must match width * height * bytes_per_pixel
camera.write_frame(frame_data)