Custom video frame processing (#530)

Sources/LiveKit/Protocols/VideoProcessor.swift

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2025 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+
+@objc
+public protocol VideoProcessor {
+    func process(frame: VideoFrame) -> VideoFrame?
+}

Sources/LiveKit/Track/Capturers/CameraCapturer.swift

@@ -96,9 +96,12 @@ public class CameraCapturer: VideoCapturer {
     // RTCCameraVideoCapturer used internally for now
     private lazy var capturer: LKRTCCameraVideoCapturer = .init(delegate: adapter)
 
-    init(delegate: LKRTCVideoCapturerDelegate, options: CameraCaptureOptions) {
+    init(delegate: LKRTCVideoCapturerDelegate,
+         options: CameraCaptureOptions,
+         processor: VideoProcessor? = nil)
+    {
         _cameraCapturerState = StateSync(State(options: options))
-        super.init(delegate: delegate)
+        super.init(delegate: delegate, processor: processor)
 
         log("isMultitaskingAccessSupported: \(isMultitaskingAccessSupported)", .info)
     }
@@ -293,10 +296,13 @@ public extension LocalVideoTrack {
     @objc
     static func createCameraTrack(name: String? = nil,
                                   options: CameraCaptureOptions? = nil,
-                                  reportStatistics: Bool = false) -> LocalVideoTrack
+                                  reportStatistics: Bool = false,
+                                  processor: VideoProcessor? = nil) -> LocalVideoTrack
     {
         let videoSource = RTC.createVideoSource(forScreenShare: false)
-        let capturer = CameraCapturer(delegate: videoSource, options: options ?? CameraCaptureOptions())
+        let capturer = CameraCapturer(delegate: videoSource,
+                                      options: options ?? CameraCaptureOptions(),
+                                      processor: processor)
         return LocalVideoTrack(name: name ?? Track.cameraName,
                                source: .camera,
                                capturer: capturer,

Sources/LiveKit/Track/Capturers/VideoCapturer.swift

@@ -46,6 +46,8 @@ public class VideoCapturer: NSObject, Loggable, VideoCapturerProtocol {
     public let delegates = MulticastDelegate<VideoCapturerDelegate>(label: "VideoCapturerDelegate")
     public let rendererDelegates = MulticastDelegate<VideoRenderer>(label: "VideoCapturerRendererDelegate")
 
+    private let processingQueue = DispatchQueue(label: "io.livekit.videocapturer.processing")
+
     /// Array of supported pixel formats that can be used to capture a frame.
     ///
     /// Usually the following formats are supported but it is recommended to confirm at run-time:
@@ -70,16 +72,23 @@ public class VideoCapturer: NSObject, Loggable, VideoCapturerProtocol {
 
     let dimensionsCompleter = AsyncCompleter<Dimensions>(label: "Dimensions", defaultTimeout: .defaultCaptureStart)
 
-    struct State: Equatable {
+    struct State {
         // Counts calls to start/stopCapturer so multiple Tracks can use the same VideoCapturer.
         var startStopCounter: Int = 0
         var dimensions: Dimensions? = nil
+        weak var processor: VideoProcessor? = nil
+        var isFrameProcessingBusy: Bool = false
     }
 
-    var _state = StateSync(State())
+    let _state: StateSync<State>
 
     public var dimensions: Dimensions? { _state.dimensions }
 
+    public weak var processor: VideoProcessor? {
+        get { _state.processor }
+        set { _state.mutate { $0.processor = newValue } }
+    }
+
     func set(dimensions newValue: Dimensions?) {
         let didUpdate = _state.mutate {
             let oldDimensions = $0.dimensions
@@ -103,8 +112,9 @@ public class VideoCapturer: NSObject, Loggable, VideoCapturerProtocol {
         _state.startStopCounter == 0 ? .stopped : .started
     }
 
-    init(delegate: LKRTCVideoCapturerDelegate) {
+    init(delegate: LKRTCVideoCapturerDelegate, processor: VideoProcessor? = nil) {
         self.delegate = delegate
+        _state = StateSync(State(processor: processor))
         super.init()
 
         _state.onDidMutate = { [weak self] newState, oldState in
@@ -223,16 +233,45 @@ extension VideoCapturer {
                                device: AVCaptureDevice?,
                                options: VideoCaptureOptions)
     {
-        // Resolve real dimensions (apply frame rotation)
-        set(dimensions: Dimensions(width: frame.width, height: frame.height).apply(rotation: frame.rotation))
+        if _state.isFrameProcessingBusy {
+            log("Frame processing hasn't completed yet, skipping frame...", .warning)
+            return
+        }
+
+        processingQueue.async { [weak self] in
+            guard let self else { return }
+
+            // Mark as frame processing busy.
+            self._state.mutate { $0.isFrameProcessingBusy = true }
+            defer {
+                self._state.mutate { $0.isFrameProcessingBusy = false }
+            }
+
+            var rtcFrame: LKRTCVideoFrame = frame
+            guard var lkFrame: VideoFrame = frame.toLKType() else {
+                self.log("Failed to convert a RTCVideoFrame to VideoFrame.", .error)
+                return
+            }
+
+            // Apply processing if we have a processor attached.
+            if let processor = self._state.processor {
+                guard let processedFrame = processor.process(frame: lkFrame) else {
+                    self.log("VideoProcessor didn't return a frame, skipping frame.", .warning)
+                    return
+                }
+                lkFrame = processedFrame
+                rtcFrame = processedFrame.toRTCType()
+            }
+
+            // Resolve real dimensions (apply frame rotation)
+            self.set(dimensions: Dimensions(width: rtcFrame.width, height: rtcFrame.height).apply(rotation: rtcFrame.rotation))
 
-        delegate?.capturer(capturer, didCapture: frame)
+            self.delegate?.capturer(capturer, didCapture: rtcFrame)
 
-        if rendererDelegates.isDelegatesNotEmpty {
-            if let lkVideoFrame = frame.toLKType() {
-                rendererDelegates.notify { renderer in
-                    renderer.render?(frame: lkVideoFrame)
-                    renderer.render?(frame: lkVideoFrame, captureDevice: device, captureOptions: options)
+            if self.rendererDelegates.isDelegatesNotEmpty {
+                self.rendererDelegates.notify { renderer in
+                    renderer.render?(frame: lkFrame)
+                    renderer.render?(frame: lkFrame, captureDevice: device, captureOptions: options)
                 }
             }
         }

Sources/LiveKit/Track/Local/LocalVideoTrack.swift

@@ -82,6 +82,12 @@ extension LocalVideoTrack: VideoTrack {
 public extension LocalVideoTrack {
     var publishOptions: TrackPublishOptions? { super._state.lastPublishOptions }
     var publishState: Track.PublishState { super._state.publishState }
+
+    /// Convenience access to ``VideoCapturer/processor``.
+    var processor: VideoProcessor? {
+        get { capturer._state.processor }
+        set { capturer._state.mutate { $0.processor = newValue } }
+    }
 }
 
 public extension LocalVideoTrack {

Sources/LiveKit/Types/VideoFrame.swift

@@ -37,6 +37,11 @@ public class CVPixelVideoBuffer: VideoBuffer, RTCCompatibleVideoBuffer {
         _rtcType.pixelBuffer
     }
 
+    public init(pixelBuffer: CVPixelBuffer) {
+        _rtcType = LKRTCCVPixelBuffer(pixelBuffer: pixelBuffer)
+    }
+
+    // Internal only.
     init(rtcCVPixelBuffer: LKRTCCVPixelBuffer) {
         _rtcType = rtcCVPixelBuffer
     }