feat(video): fullscreen NDI output per screen with 50 Hz, reconnect and stats

- NDIReceiver: per-source receive thread, IOSurface pixel buffers, watchdog reconnect
- VideoOutput: borderless output window above menu bar, AVSampleBufferVideoRenderer,
  50 Hz display mode for app lifetime, display sleep assertion
- Screen cards with drag and drop, live fps and frame loss
- Per-minute stats CSV in ~/Library/Logs/Musubi for long-run tests
- --route launch argument for scripted tests

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
main
Samuel Zielke 2 days ago
parent 84d9a392db
commit 37f46791e4

@ -4,16 +4,26 @@ import SwiftUI
struct ContentView: View { struct ContentView: View {
let sourceList: SourceListModel let sourceList: SourceListModel
let outputs: OutputController
var body: some View { var body: some View {
NavigationSplitView { NavigationSplitView {
SourceListView(sourceList: sourceList) SourceListView(sourceList: sourceList)
.navigationSplitViewColumnWidth(min: 220, ideal: 260) .navigationSplitViewColumnWidth(min: 220, ideal: 260)
} detail: { } detail: {
EmptyStateView(text: "Noch keine Szene. Zieh eine Quelle auf einen Bildschirm.") ScreensView(outputs: outputs)
} }
.navigationTitle("Musubi") .navigationTitle("Musubi")
.navigationSubtitle("NDI rein. Bild, Ton und Stimme raus.") .navigationSubtitle("NDI rein. Bild, Ton und Stimme raus.")
.toolbar {
if outputs.isSleepBlocked {
ToolbarItem(placement: .primaryAction) {
Text("Ruhezustand blockiert")
.font(Theme.Font.caption)
.foregroundStyle(Theme.Color.nezumi)
}
}
}
} }
} }
@ -31,6 +41,7 @@ private struct SourceListView: View {
} else { } else {
List(sourceList.sources, id: \.ndiName) { source in List(sourceList.sources, id: \.ndiName) { source in
SourceRow(source: source) SourceRow(source: source)
.draggable(source.ndiName)
} }
.scrollContentBackground(.hidden) .scrollContentBackground(.hidden)
} }

@ -6,12 +6,16 @@ import SwiftUI
@main @main
struct MusubiApp: App { struct MusubiApp: App {
@State private var sourceList = SourceListModel() @State private var sourceList = SourceListModel()
@State private var outputs = OutputController()
var body: some SwiftUI.Scene { var body: some SwiftUI.Scene {
Window("Musubi", id: "main") { Window("Musubi", id: "main") {
ContentView(sourceList: sourceList) ContentView(sourceList: sourceList, outputs: outputs)
.preferredColorScheme(.dark) .preferredColorScheme(.dark)
.task { sourceList.start() } .task {
sourceList.start()
outputs.applyLaunchRoutes()
}
} }
.defaultSize(width: 960, height: 600) .defaultSize(width: 960, height: 600)
} }

@ -0,0 +1,156 @@
import AppKit
import MusubiCore
import NDIKit
import Observation
import VideoOutput
/// A connected screen as shown in the UI.
struct ScreenInfo: Identifiable, Equatable {
let id: CGDirectDisplayID
let name: String
let width: Int
let height: Int
var refreshRate: Double
}
/// One source shown fullscreen on one screen.
@MainActor
final class ActiveOutput {
let source: Source
let receiver: NDIReceiver
let window: OutputWindow
let displayID: CGDirectDisplayID
let originalMode: CGDisplayMode?
var snapshot: NDIReceiver.Snapshot
init(source: Source, receiver: NDIReceiver, window: OutputWindow, displayID: CGDirectDisplayID, originalMode: CGDisplayMode?) {
self.source = source
self.receiver = receiver
self.window = window
self.displayID = displayID
self.originalMode = originalMode
self.snapshot = receiver.snapshot
}
}
@MainActor @Observable
final class OutputController {
static let targetRefreshRate = 50.0
private(set) var screens: [ScreenInfo] = []
private(set) var outputs: [CGDirectDisplayID: ActiveOutput] = [:]
/// Bumped once per second so views re-read the receiver snapshots.
private(set) var tick = 0
private var sleepBlocker: SleepBlocker?
private var timer: Timer?
private let recorder = StatsRecorder()
var isSleepBlocked: Bool { sleepBlocker != nil }
init() {
refreshScreens()
NotificationCenter.default.addObserver(forName: NSApplication.didChangeScreenParametersNotification, object: nil, queue: .main) { [weak self] _ in
MainActor.assumeIsolated { self?.screensChanged() }
}
timer = Timer.scheduledTimer(withTimeInterval: 1, repeats: true) { [weak self] _ in
MainActor.assumeIsolated { self?.updateSnapshots() }
}
}
func show(_ source: Source, on displayID: CGDirectDisplayID) {
guard let screen = NSScreen.screens.first(where: { $0.displayID == displayID }) else { return }
stop(displayID)
let window = OutputWindow(screen: screen)
let sink = window.sink
let receiverName = "Musubi \(screen.localizedName)"
guard let receiver = NDIReceiver(source: source, receiverName: receiverName, onFrame: { sink.display($0) }) else {
Log.ndi.error("Could not create receiver for \(source.ndiName, privacy: .public)")
return
}
let mode = DisplayModeSwitcher.setRefreshRate(Self.targetRefreshRate, on: displayID)
window.onCancel = { [weak self] in self?.stop(displayID) }
window.setFrame(screen.frame, display: true)
window.orderFrontRegardless()
outputs[displayID] = ActiveOutput(source: source, receiver: receiver, window: window, displayID: displayID, originalMode: mode.original)
Log.ndi.info("Output \(source.ndiName, privacy: .public) → \(screen.localizedName, privacy: .public) @ \(mode.refreshRate) Hz")
updateSleepBlocker()
refreshScreens()
}
func stop(_ displayID: CGDirectDisplayID) {
guard let output = outputs.removeValue(forKey: displayID) else { return }
output.receiver.stop()
output.window.orderOut(nil)
if let original = output.originalMode {
DisplayModeSwitcher.restore(original, on: displayID)
}
Log.ndi.info("Output stopped on display \(displayID)")
updateSleepBlocker()
refreshScreens()
}
/// Applies `--route "<NDI name>=<screen index>"` launch arguments.
func applyLaunchRoutes(_ arguments: [String] = ProcessInfo.processInfo.arguments) {
for route in RouteArgument.parse(arguments) {
guard route.screenIndex < NSScreen.screens.count else {
Log.ndi.error("--route: no screen \(route.screenIndex)")
continue
}
show(Source(ndiName: route.ndiName), on: NSScreen.screens[route.screenIndex].displayID)
}
}
private func updateSnapshots() {
for output in outputs.values {
output.snapshot = output.receiver.snapshot
}
recorder.recordIfDue(outputs: Array(outputs.values), screens: screens)
tick += 1
}
private func updateSleepBlocker() {
if outputs.isEmpty {
sleepBlocker = nil
} else if sleepBlocker == nil {
sleepBlocker = SleepBlocker(reason: "Musubi: Ausgabe aktiv")
}
}
private func screensChanged() {
let connected = Set(NSScreen.screens.map(\.displayID))
for displayID in outputs.keys where !connected.contains(displayID) {
Log.ndi.notice("Display \(displayID) disconnected, stopping its output")
stop(displayID)
}
// Keep windows covering their screen after resolution changes.
for output in outputs.values {
if let screen = NSScreen.screens.first(where: { $0.displayID == output.displayID }) {
output.window.setFrame(screen.frame, display: true)
}
}
refreshScreens()
}
private func refreshScreens() {
screens = NSScreen.screens.map { screen in
let displayID = screen.displayID
return ScreenInfo(
id: displayID,
name: screen.localizedName,
width: Int(CGDisplayPixelsWide(displayID)),
height: Int(CGDisplayPixelsHigh(displayID)),
refreshRate: DisplayModeSwitcher.currentRefreshRate(of: displayID)
)
}
}
}
extension NSScreen {
var displayID: CGDirectDisplayID {
(deviceDescription[NSDeviceDescriptionKey("NSScreenNumber")] as? NSNumber)?.uint32Value ?? 0
}
}

@ -0,0 +1,111 @@
import DesignSystem
import MusubiCore
import SwiftUI
/// Cards for all connected screens; drop a source on a card to show it fullscreen there.
struct ScreensView: View {
let outputs: OutputController
var body: some View {
ScrollView {
VStack(alignment: .leading, spacing: Theme.Space.card) {
Text("Bildschirme")
.font(Theme.Font.title)
.foregroundStyle(Theme.Color.washi)
if outputs.outputs.isEmpty {
Text("Noch keine Szene. Zieh eine Quelle auf einen Bildschirm.")
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.nezumi)
}
LazyVGrid(columns: [GridItem(.adaptive(minimum: 260), spacing: Theme.Space.card)], spacing: Theme.Space.card) {
ForEach(outputs.screens) { screen in
ScreenCard(screen: screen, output: outputs.outputs[screen.id], tick: outputs.tick) {
outputs.stop(screen.id)
}
.dropDestination(for: String.self) { names, _ in
guard let name = names.first else { return false }
outputs.show(Source(ndiName: name), on: screen.id)
return true
}
}
}
}
.padding(Theme.Space.card)
}
.background(Theme.Color.sumi)
}
}
private struct ScreenCard: View {
let screen: ScreenInfo
let output: ActiveOutput?
let tick: Int
let onStop: () -> Void
var body: some View {
VStack(alignment: .leading, spacing: Theme.Space.unit) {
Text(verbatim: screen.name)
.font(Theme.Font.title)
.foregroundStyle(Theme.Color.washi)
Text(verbatim: "\(screen.width) × \(screen.height) · \(screen.refreshRate.formatted(.number.precision(.fractionLength(0...2)))) Hz")
.font(Theme.Font.number)
.foregroundStyle(Theme.Color.nezumi)
Divider().overlay(Theme.Color.usuzumi)
if let output {
OutputStatus(output: output, tick: tick)
Button("Ausgabe beenden", action: onStop)
.buttonStyle(.bordered)
} else {
Text("Quelle hierher ziehen")
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.nezumi)
}
}
.frame(maxWidth: .infinity, minHeight: 150, alignment: .topLeading)
.padding(Theme.Space.card)
.background(Theme.Color.kasumi, in: RoundedRectangle(cornerRadius: Theme.Radius.card))
.overlay(
RoundedRectangle(cornerRadius: Theme.Radius.card)
.strokeBorder(output == nil ? Theme.Color.usuzumi : Theme.Color.ai, lineWidth: output == nil ? Theme.Stroke.line : Theme.Stroke.cord)
)
}
}
private struct OutputStatus: View {
let output: ActiveOutput
let tick: Int
var body: some View {
let snapshot = output.snapshot
VStack(alignment: .leading, spacing: Theme.Space.unit / 2) {
Text(verbatim: output.source.label)
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.washi)
switch snapshot.state {
case .connecting:
Text("Verbinde …")
.font(Theme.Font.caption)
.foregroundStyle(Theme.Color.nezumi)
case .live:
HStack(spacing: Theme.Space.unit / 2) {
Circle().fill(Theme.Color.moegi).frame(width: Theme.Stroke.knot, height: Theme.Stroke.knot)
Text(verbatim: "\(snapshot.width) × \(snapshot.height) · \(snapshot.frameRate.formatted(.number.precision(.fractionLength(0...2)))) fps")
.font(Theme.Font.number)
.foregroundStyle(Theme.Color.nezumi)
}
case .stalled(let retryIn):
Text("\(output.source.label) antwortet nicht mehr. Neuer Versuch in \(Int(retryIn.rounded(.up))) s.")
.font(Theme.Font.caption)
.foregroundStyle(Theme.Color.kohaku)
}
let lost = snapshot.stats.missing + snapshot.stats.droppedByReceiver
Text("\(lost) Frames verloren (\(snapshot.stats.lossRate.formatted(.percent.precision(.fractionLength(2)))))")
.font(Theme.Font.number)
.foregroundStyle(snapshot.stats.lossRate >= 0.001 ? Theme.Color.kohaku : Theme.Color.nezumi)
}
}
}

@ -0,0 +1,74 @@
import CoreGraphics
import Foundation
import MusubiCore
import NDIKit
/// Appends one CSV line per output and minute to ~/Library/Logs/Musubi/stats-<date>.csv (for long-run tests).
@MainActor
final class StatsRecorder {
private let interval: TimeInterval = 60
private var lastWrite = Date()
private var lastCPU = StatsRecorder.cpuTime()
private var lastReceived: [CGDirectDisplayID: Int64] = [:]
private let directory = FileManager.default.homeDirectoryForCurrentUser.appending(path: "Library/Logs/Musubi")
private static let header = "time,display,refresh_hz,source,state,width,height,source_fps,shown_fps,received,missing,dropped_by_ndi,loss_rate,cpu_percent\n"
func recordIfDue(outputs: [ActiveOutput], screens: [ScreenInfo]) {
let now = Date()
let elapsed = now.timeIntervalSince(lastWrite)
guard elapsed >= interval else { return }
let cpu = Self.cpuTime()
let cpuPercent = (cpu - lastCPU) / elapsed * 100
lastCPU = cpu
lastWrite = now
guard !outputs.isEmpty else { return }
let time = now.ISO8601Format()
var lines = ""
for output in outputs {
let s = output.snapshot
let previous = lastReceived[output.displayID] ?? s.stats.received
lastReceived[output.displayID] = s.stats.received
let shownFPS = Double(s.stats.received - previous) / elapsed
let refresh = screens.first { $0.id == output.displayID }?.refreshRate ?? 0
let state: String = switch s.state {
case .connecting: "connecting"
case .live: "live"
case .stalled: "stalled"
}
let name = output.source.ndiName.replacingOccurrences(of: "\"", with: "'")
lines += "\(time),\(output.displayID),\(refresh),\"\(name)\",\(state),\(s.width),\(s.height),"
lines += String(format: "%.2f,%.2f,%lld,%lld,%lld,%.6f,%.1f\n",
s.frameRate, shownFPS, s.stats.received, s.stats.missing, s.stats.droppedByReceiver,
s.stats.lossRate, cpuPercent)
}
append(lines, date: now)
}
private func append(_ text: String, date: Date) {
let day = date.formatted(.iso8601.year().month().day())
let file = directory.appending(path: "stats-\(day).csv")
do {
try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
if !FileManager.default.fileExists(atPath: file.path) {
try Self.header.write(to: file, atomically: true, encoding: .utf8)
}
let handle = try FileHandle(forWritingTo: file)
defer { try? handle.close() }
try handle.seekToEnd()
try handle.write(contentsOf: Data(text.utf8))
} catch {
Log.ndi.error("Stats CSV: \(error.localizedDescription, privacy: .public)")
}
}
/// User + system CPU seconds of this process.
private static func cpuTime() -> Double {
var usage = rusage()
getrusage(RUSAGE_SELF, &usage)
func seconds(_ t: timeval) -> Double { Double(t.tv_sec) + Double(t.tv_usec) / 1_000_000 }
return seconds(usage.ru_utime) + seconds(usage.ru_stime)
}
}

@ -0,0 +1,41 @@
/// Decides when a receiver counts as stalled and when to try reconnecting.
public struct ConnectionWatchdog: Sendable, Equatable {
public enum State: Equatable, Sendable {
case connecting
case live
/// No frames for a while. `retryIn` is the time until the next reconnect attempt.
case stalled(retryIn: Double)
}
public var stallAfter: Double
public var retryInterval: Double
private var lastFrame: Double?
private var lastRetry: Double?
private let startedAt: Double
public init(now: Double, stallAfter: Double = 2, retryInterval: Double = 3) {
self.startedAt = now
self.stallAfter = stallAfter
self.retryInterval = retryInterval
}
public mutating func frameArrived(at now: Double) {
lastFrame = now
lastRetry = nil
}
public func state(at now: Double) -> State {
let reference = lastFrame ?? startedAt
guard now - reference >= stallAfter else { return lastFrame == nil ? .connecting : .live }
let since = lastRetry ?? reference + stallAfter
return .stalled(retryIn: max(0, retryInterval - (now - since)))
}
/// Returns true when a reconnect should be attempted now, and records the attempt.
public mutating func shouldRetry(at now: Double) -> Bool {
guard case .stalled(let retryIn) = state(at: now), retryIn == 0 else { return false }
lastRetry = now
return true
}
}

@ -0,0 +1,43 @@
/// A display mode reduced to what matters for picking one. Mirrors CGDisplayMode.
public struct DisplayModeInfo: Hashable, Sendable {
public var width: Int // points
public var height: Int
public var pixelWidth: Int
public var pixelHeight: Int
public var refreshRate: Double // Hz, 0 for some built-in panels
public init(width: Int, height: Int, pixelWidth: Int, pixelHeight: Int, refreshRate: Double) {
self.width = width
self.height = height
self.pixelWidth = pixelWidth
self.pixelHeight = pixelHeight
self.refreshRate = refreshRate
}
public var isHiDPI: Bool { pixelWidth > width }
}
public enum DisplayModeSelection {
/// Picks the mode with the same size as `current` and a refresh rate within
/// `tolerance` of `targetHz`. Returns nil if `current` already matches or no such mode exists.
public static func mode(
matching targetHz: Double,
current: DisplayModeInfo,
available: [DisplayModeInfo],
tolerance: Double = 0.1
) -> DisplayModeInfo? {
if abs(current.refreshRate - targetHz) <= tolerance { return nil }
return available
.filter {
$0.width == current.width && $0.height == current.height
&& abs($0.refreshRate - targetHz) <= tolerance
}
// Prefer the same backing resolution, then the closest rate.
.min { lhs, rhs in
let lhsSame = lhs.pixelWidth == current.pixelWidth && lhs.pixelHeight == current.pixelHeight
let rhsSame = rhs.pixelWidth == current.pixelWidth && rhs.pixelHeight == current.pixelHeight
if lhsSame != rhsSame { return lhsSame }
return abs(lhs.refreshRate - targetHz) < abs(rhs.refreshRate - targetHz)
}
}
}

@ -0,0 +1,44 @@
/// Counts received and missing video frames from NDI timestamps.
///
/// A gap larger than 1.5 frame durations counts the frames that should have arrived in between.
/// Gaps longer than `stallThreshold` are treated as an outage (reconnect), not as lost frames.
public struct FrameStats: Sendable, Equatable {
public private(set) var received: Int64 = 0
public private(set) var missing: Int64 = 0
public private(set) var droppedByReceiver: Int64 = 0
private var lastTimestamp: Int64?
/// NDI timestamps are in 100 ns units.
public static let ticksPerSecond: Double = 10_000_000
public var stallThreshold: Double = 2.0 // seconds
public init() {}
/// Lost frames relative to all frames that should have been shown.
public var lossRate: Double {
let lost = missing + droppedByReceiver
let expected = received + lost
return expected == 0 ? 0 : Double(lost) / Double(expected)
}
public mutating func record(timestamp: Int64, frameRateN: Int32, frameRateD: Int32) {
received += 1
defer { lastTimestamp = timestamp }
guard let last = lastTimestamp, frameRateN > 0, frameRateD > 0 else { return }
let delta = Double(timestamp - last) / Self.ticksPerSecond
let frameDuration = Double(frameRateD) / Double(frameRateN)
guard delta > frameDuration * 1.5, delta < stallThreshold else { return }
missing += Int64((delta / frameDuration).rounded()) - 1
}
/// Total frames NDI itself dropped because we did not capture fast enough.
public mutating func updateDroppedByReceiver(_ total: Int64) {
droppedByReceiver = total
}
/// Call after a reconnect so the outage is not counted as lost frames.
public mutating func resetTimeline() {
lastTimestamp = nil
}
}

@ -0,0 +1,16 @@
/// Parses `--route "<NDI name>=<screen index>"` launch arguments (for scripted tests until scenes exist).
public enum RouteArgument {
public struct Parsed: Equatable, Sendable {
public var ndiName: String
public var screenIndex: Int
}
public static func parse(_ arguments: [String]) -> [Parsed] {
zip(arguments, arguments.dropFirst()).compactMap { flag, value in
guard flag == "--route", let separator = value.lastIndex(of: "=") else { return nil }
let name = String(value[..<separator])
guard !name.isEmpty, let index = Int(value[value.index(after: separator)...]), index >= 0 else { return nil }
return Parsed(ndiName: name, screenIndex: index)
}
}
}

@ -0,0 +1,195 @@
import CNDI
import CoreVideo
import Foundation
import MusubiCore
import OSLog
/// Receives video from one NDI source on a dedicated thread and hands out IOSurface-backed pixel buffers.
///
/// Reconnects on its own when frames stop arriving (see `ConnectionWatchdog`).
public final class NDIReceiver: @unchecked Sendable {
public struct Snapshot: Sendable, Equatable {
public var state: ConnectionWatchdog.State
public var stats: FrameStats
public var width: Int
public var height: Int
public var frameRate: Double
}
public typealias FrameHandler = @Sendable (CVPixelBuffer) -> Void
public let source: Source
private let instance: NDIlib_recv_instance_t
private let onFrame: FrameHandler
private let log = Logger(subsystem: "studio.zielke.musubi", category: "ndi-recv")
// Shared with the UI, guarded by `lock`.
private let lock = NSLock()
private var running = true
private var watchdog: ConnectionWatchdog
private var current: Snapshot
// Only touched on the receive thread.
private var pool: CVPixelBufferPool?
private var poolKey: (width: Int, height: Int, format: OSType)?
public init?(source: Source, receiverName: String, onFrame: @escaping FrameHandler) {
let created: NDIlib_recv_instance_t? = source.ndiName.withCString { name in
receiverName.withCString { recvName in
var ndiSource = NDIlib_source_t()
ndiSource.p_ndi_name = name
var settings = NDIlib_recv_create_v3_t(
source_to_connect_to: ndiSource,
color_format: NDIlib_recv_color_format_UYVY_BGRA,
bandwidth: NDIlib_recv_bandwidth_highest,
allow_video_fields: false,
p_ndi_recv_name: recvName
)
return NDIlib_recv_create_v3(&settings)
}
}
guard let created else { return nil }
let now = ProcessInfo.processInfo.systemUptime
self.source = source
self.instance = created
self.onFrame = onFrame
self.watchdog = ConnectionWatchdog(now: now)
self.current = Snapshot(state: .connecting, stats: FrameStats(), width: 0, height: 0, frameRate: 0)
let thread = Thread { [self] in run() }
thread.name = "studio.zielke.musubi.ndi-recv"
thread.qualityOfService = .userInteractive
thread.start()
}
deinit {
NDIlib_recv_destroy(instance)
}
/// Stops the receive thread. The instance is destroyed once the last reference goes away.
public func stop() {
lock.withLock { running = false }
}
public var snapshot: Snapshot {
lock.withLock {
var snapshot = current
snapshot.state = watchdog.state(at: ProcessInfo.processInfo.systemUptime)
return snapshot
}
}
// MARK: - Receive thread
private func run() {
var lastPerformanceCheck = 0.0
while lock.withLock({ running }) {
var frame = NDIlib_video_frame_v2_t()
let type = NDIlib_recv_capture_v3(instance, &frame, nil, nil, 250)
let now = ProcessInfo.processInfo.systemUptime
if type == NDIlib_frame_type_video {
if let buffer = makePixelBuffer(from: frame) {
onFrame(buffer)
}
let rate = frame.frame_rate_D > 0 ? Double(frame.frame_rate_N) / Double(frame.frame_rate_D) : 0
lock.withLock {
watchdog.frameArrived(at: now)
current.stats.record(timestamp: frame.timestamp, frameRateN: frame.frame_rate_N, frameRateD: frame.frame_rate_D)
current.width = Int(frame.xres)
current.height = Int(frame.yres)
current.frameRate = rate
}
NDIlib_recv_free_video_v2(instance, &frame)
}
if now - lastPerformanceCheck >= 1 {
lastPerformanceCheck = now
var dropped = NDIlib_recv_performance_t()
NDIlib_recv_get_performance(instance, nil, &dropped)
lock.withLock { current.stats.updateDroppedByReceiver(dropped.video_frames) }
}
if lock.withLock({ watchdog.shouldRetry(at: now) }) {
log.notice("No frames from \(self.source.ndiName, privacy: .public), reconnecting")
reconnect()
lock.withLock { current.stats.resetTimeline() }
}
}
}
private func reconnect() {
source.ndiName.withCString { name in
var ndiSource = NDIlib_source_t()
ndiSource.p_ndi_name = name
NDIlib_recv_connect(instance, nil)
NDIlib_recv_connect(instance, &ndiSource)
}
}
private func makePixelBuffer(from frame: NDIlib_video_frame_v2_t) -> CVPixelBuffer? {
let format: OSType
let bytesPerPixel: Int
switch frame.FourCC {
case NDIlib_FourCC_video_type_UYVY:
format = kCVPixelFormatType_422YpCbCr8 // '2vuy' has the same byte order as UYVY
bytesPerPixel = 2
case NDIlib_FourCC_video_type_BGRA, NDIlib_FourCC_video_type_BGRX:
format = kCVPixelFormatType_32BGRA
bytesPerPixel = 4
default:
return nil
}
let width = Int(frame.xres), height = Int(frame.yres)
guard width > 0, height > 0, let source = frame.p_data,
let pool = pixelBufferPool(width: width, height: height, format: format)
else { return nil }
var buffer: CVPixelBuffer?
guard CVPixelBufferPoolCreatePixelBuffer(nil, pool, &buffer) == kCVReturnSuccess, let buffer else { return nil }
CVPixelBufferLockBaseAddress(buffer, [])
defer { CVPixelBufferUnlockBaseAddress(buffer, []) }
guard let destination = CVPixelBufferGetBaseAddress(buffer) else { return nil }
let sourceStride = Int(frame.line_stride_in_bytes)
let destinationStride = CVPixelBufferGetBytesPerRow(buffer)
if sourceStride == destinationStride {
memcpy(destination, source, sourceStride * height)
} else {
let rowBytes = min(width * bytesPerPixel, sourceStride, destinationStride)
for row in 0..<height {
memcpy(destination + row * destinationStride, source + row * sourceStride, rowBytes)
}
}
if format == kCVPixelFormatType_422YpCbCr8 {
// NDI uses BT.709 for HD and BT.601 for SD.
let hd = height >= 720
CVBufferSetAttachment(buffer, kCVImageBufferYCbCrMatrixKey,
hd ? kCVImageBufferYCbCrMatrix_ITU_R_709_2 : kCVImageBufferYCbCrMatrix_ITU_R_601_4, .shouldPropagate)
CVBufferSetAttachment(buffer, kCVImageBufferColorPrimariesKey,
hd ? kCVImageBufferColorPrimaries_ITU_R_709_2 : kCVImageBufferColorPrimaries_SMPTE_C, .shouldPropagate)
CVBufferSetAttachment(buffer, kCVImageBufferTransferFunctionKey, kCVImageBufferTransferFunction_ITU_R_709_2, .shouldPropagate)
}
return buffer
}
private func pixelBufferPool(width: Int, height: Int, format: OSType) -> CVPixelBufferPool? {
if let key = poolKey, key.width == width, key.height == height, key.format == format, let pool { return pool }
let attributes: [CFString: Any] = [
kCVPixelBufferWidthKey: width,
kCVPixelBufferHeightKey: height,
kCVPixelBufferPixelFormatTypeKey: format,
kCVPixelBufferIOSurfacePropertiesKey: [CFString: Any](),
kCVPixelBufferMetalCompatibilityKey: true,
]
var newPool: CVPixelBufferPool?
CVPixelBufferPoolCreate(nil, [kCVPixelBufferPoolMinimumBufferCountKey: 4] as CFDictionary, attributes as CFDictionary, &newPool)
pool = newPool
poolKey = (width, height, format)
log.info("\(self.source.ndiName, privacy: .public): \(width)x\(height) \(format == kCVPixelFormatType_32BGRA ? "BGRA" : "UYVY", privacy: .public)")
return newPool
}
}

@ -0,0 +1,55 @@
import CoreGraphics
import MusubiCore
import OSLog
/// Switches a display to a target refresh rate for the lifetime of the app and restores it on request.
public enum DisplayModeSwitcher {
private static let log = Logger(subsystem: "studio.zielke.musubi", category: "display")
/// Returns the refresh rate in effect afterwards, and the original mode to restore later (nil if unchanged).
@discardableResult
public static func setRefreshRate(_ hz: Double, on display: CGDirectDisplayID) -> (refreshRate: Double, original: CGDisplayMode?) {
guard let current = CGDisplayCopyDisplayMode(display) else { return (0, nil) }
let options = [kCGDisplayShowDuplicateLowResolutionModes: true] as CFDictionary
let modes = (CGDisplayCopyAllDisplayModes(display, options) as? [CGDisplayMode]) ?? []
guard let target = DisplayModeSelection.mode(matching: hz, current: info(current), available: modes.map(info)),
let mode = modes.first(where: { info($0) == target })
else {
if abs(current.refreshRate - hz) > 0.1 {
log.notice("Display \(display): no \(hz) Hz mode, staying at \(current.refreshRate) Hz")
}
return (current.refreshRate, nil)
}
guard apply(mode, to: display) else { return (current.refreshRate, nil) }
log.info("Display \(display): \(current.refreshRate) Hz → \(mode.refreshRate) Hz")
return (mode.refreshRate, current)
}
public static func restore(_ mode: CGDisplayMode, on display: CGDirectDisplayID) {
if apply(mode, to: display) {
log.info("Display \(display): restored \(mode.refreshRate) Hz")
}
}
public static func currentRefreshRate(of display: CGDirectDisplayID) -> Double {
CGDisplayCopyDisplayMode(display)?.refreshRate ?? 0
}
private static func apply(_ mode: CGDisplayMode, to display: CGDirectDisplayID) -> Bool {
var config: CGDisplayConfigRef?
guard CGBeginDisplayConfiguration(&config) == .success else { return false }
CGConfigureDisplayWithDisplayMode(config, display, mode, nil)
// .forAppOnly: macOS reverts the mode when Musubi quits or crashes.
let result = CGCompleteDisplayConfiguration(config, .forAppOnly)
if result != .success { log.error("Display \(display): mode change failed (\(result.rawValue))") }
return result == .success
}
private static func info(_ mode: CGDisplayMode) -> DisplayModeInfo {
DisplayModeInfo(width: mode.width, height: mode.height,
pixelWidth: mode.pixelWidth, pixelHeight: mode.pixelHeight,
refreshRate: mode.refreshRate)
}
}

@ -0,0 +1,58 @@
import AppKit
import AVFoundation
/// A borderless window covering one screen, above the menu bar, that shows a video layer.
@MainActor
public final class OutputWindow: NSWindow {
public let sink: VideoSink
/// Called when the operator presses Esc on the output.
public var onCancel: (() -> Void)?
public init(screen: NSScreen) {
let view = VideoView()
sink = VideoSink(renderer: view.displayLayer.sampleBufferRenderer)
super.init(contentRect: screen.frame, styleMask: .borderless, backing: .buffered, defer: false)
contentView = view
setFrame(screen.frame, display: false)
level = .screenSaver
collectionBehavior = [.canJoinAllSpaces, .stationary, .ignoresCycle, .fullScreenNone]
backgroundColor = .black
isOpaque = true
hasShadow = false
isReleasedWhenClosed = false
animationBehavior = .none
}
public override var canBecomeKey: Bool { true }
public override func cancelOperation(_ sender: Any?) {
onCancel?()
}
}
/// Layer-hosting view backed by `AVSampleBufferDisplayLayer`; hides the cursor while it is inside.
@MainActor
final class VideoView: NSView {
let displayLayer = AVSampleBufferDisplayLayer()
init() {
super.init(frame: .zero)
displayLayer.videoGravity = .resizeAspect
displayLayer.backgroundColor = NSColor.black.cgColor
layer = displayLayer
wantsLayer = true
}
@available(*, unavailable)
required init?(coder: NSCoder) { fatalError() }
override func updateTrackingAreas() {
super.updateTrackingAreas()
trackingAreas.forEach(removeTrackingArea)
addTrackingArea(NSTrackingArea(rect: .zero, options: [.mouseEnteredAndExited, .mouseMoved, .activeAlways, .inVisibleRect], owner: self))
}
override func mouseEntered(with event: NSEvent) { NSCursor.setHiddenUntilMouseMoves(true) }
override func mouseMoved(with event: NSEvent) { NSCursor.setHiddenUntilMouseMoves(true) }
}

@ -0,0 +1,20 @@
import IOKit.pwr_mgt
/// Keeps the display (and therefore the system) awake while held.
public final class SleepBlocker {
private var assertion: IOPMAssertionID = 0
public init?(reason: String) {
let result = IOPMAssertionCreateWithName(
kIOPMAssertionTypePreventUserIdleDisplaySleep as CFString,
IOPMAssertionLevel(kIOPMAssertionLevelOn),
reason as CFString,
&assertion
)
guard result == kIOReturnSuccess else { return nil }
}
deinit {
IOPMAssertionRelease(assertion)
}
}

@ -1,3 +0,0 @@
// Video presentation (Metal / AVSampleBufferDisplayLayer), one fullscreen window per NSScreen.
// Implemented in milestone 1.
import MusubiCore

@ -0,0 +1,43 @@
import AVFoundation
import CoreMedia
import CoreVideo
/// Hands pixel buffers to an `AVSampleBufferVideoRenderer`. Safe to call from the receive thread.
public final class VideoSink: @unchecked Sendable {
private let renderer: AVSampleBufferVideoRenderer
// Only touched from the single thread that calls `display(_:)`.
private var formatDescription: CMVideoFormatDescription?
public init(renderer: AVSampleBufferVideoRenderer) {
self.renderer = renderer
}
public func display(_ pixelBuffer: CVPixelBuffer) {
if renderer.status == .failed || renderer.requiresFlushToResumeDecoding {
renderer.flush()
}
if formatDescription.map({ !CMVideoFormatDescriptionMatchesImageBuffer($0, imageBuffer: pixelBuffer) }) ?? true {
formatDescription = nil
CMVideoFormatDescriptionCreateForImageBuffer(allocator: nil, imageBuffer: pixelBuffer, formatDescriptionOut: &formatDescription)
}
guard let formatDescription else { return }
var timing = CMSampleTimingInfo(duration: .invalid, presentationTimeStamp: CMClockGetTime(CMClockGetHostTimeClock()), decodeTimeStamp: .invalid)
var sampleBuffer: CMSampleBuffer?
guard CMSampleBufferCreateReadyWithImageBuffer(
allocator: nil, imageBuffer: pixelBuffer, formatDescription: formatDescription,
sampleTiming: &timing, sampleBufferOut: &sampleBuffer
) == noErr, let sampleBuffer else { return }
if let attachments = CMSampleBufferGetSampleAttachmentsArray(sampleBuffer, createIfNecessary: true),
CFArrayGetCount(attachments) > 0 {
let dictionary = unsafeBitCast(CFArrayGetValueAtIndex(attachments, 0), to: CFMutableDictionary.self)
CFDictionarySetValue(dictionary,
Unmanaged.passUnretained(kCMSampleAttachmentKey_DisplayImmediately).toOpaque(),
Unmanaged.passUnretained(kCFBooleanTrue).toOpaque())
}
renderer.enqueue(sampleBuffer)
}
}

@ -0,0 +1,127 @@
import Testing
@testable import MusubiCore
@Suite struct DisplayModeSelectionTests {
let current = DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 1920, pixelHeight: 1080, refreshRate: 60)
@Test func picksSameSizeAt50Hz() {
let modes = [
current,
DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 1920, pixelHeight: 1080, refreshRate: 50),
DisplayModeInfo(width: 1280, height: 720, pixelWidth: 1280, pixelHeight: 720, refreshRate: 50),
]
#expect(DisplayModeSelection.mode(matching: 50, current: current, available: modes)?.refreshRate == 50)
#expect(DisplayModeSelection.mode(matching: 50, current: current, available: modes)?.width == 1920)
}
@Test func prefersSameBackingResolution() {
let hidpi = DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 3840, pixelHeight: 2160, refreshRate: 60)
let modes = [
DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 1920, pixelHeight: 1080, refreshRate: 50),
DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 3840, pixelHeight: 2160, refreshRate: 50),
]
#expect(DisplayModeSelection.mode(matching: 50, current: hidpi, available: modes)?.pixelWidth == 3840)
}
@Test func nilWhenAlreadyAt50HzOrUnavailable() {
let at50 = DisplayModeInfo(width: 1920, height: 1080, pixelWidth: 1920, pixelHeight: 1080, refreshRate: 50)
#expect(DisplayModeSelection.mode(matching: 50, current: at50, available: [at50]) == nil)
#expect(DisplayModeSelection.mode(matching: 50, current: current, available: [current]) == nil)
}
}
@Suite struct FrameStatsTests {
// 50 fps → 200_000 ticks per frame.
let frame: Int64 = 200_000
@Test func countsGapsAsMissingFrames() {
var stats = FrameStats()
stats.record(timestamp: 0, frameRateN: 50, frameRateD: 1)
stats.record(timestamp: frame, frameRateN: 50, frameRateD: 1)
stats.record(timestamp: frame * 4, frameRateN: 50, frameRateD: 1) // 2 frames missing
#expect(stats.received == 3)
#expect(stats.missing == 2)
#expect(stats.lossRate == 0.4)
}
@Test func toleratesJitter() {
var stats = FrameStats()
stats.record(timestamp: 0, frameRateN: 50, frameRateD: 1)
stats.record(timestamp: frame + frame / 4, frameRateN: 50, frameRateD: 1)
#expect(stats.missing == 0)
}
@Test func outageIsNotCountedAsLoss() {
var stats = FrameStats()
stats.record(timestamp: 0, frameRateN: 50, frameRateD: 1)
stats.record(timestamp: 50_000_000, frameRateN: 50, frameRateD: 1) // 5 s later
#expect(stats.missing == 0)
}
@Test func includesFramesDroppedByReceiver() {
var stats = FrameStats()
for i in 0..<998 { stats.record(timestamp: Int64(i) * frame, frameRateN: 50, frameRateD: 1) }
stats.updateDroppedByReceiver(2)
#expect(stats.lossRate == 0.002)
}
@Test func handlesNTSCRates() {
var stats = FrameStats()
let ticks = Int64((1001.0 / 30000.0 * FrameStats.ticksPerSecond).rounded())
stats.record(timestamp: 0, frameRateN: 30000, frameRateD: 1001)
stats.record(timestamp: ticks * 3, frameRateN: 30000, frameRateD: 1001)
#expect(stats.missing == 2)
}
}
@Suite struct ConnectionWatchdogTests {
@Test func connectingThenLive() {
var dog = ConnectionWatchdog(now: 0)
#expect(dog.state(at: 1) == .connecting)
dog.frameArrived(at: 1)
#expect(dog.state(at: 1.5) == .live)
}
@Test func stallsAndRetriesEveryInterval() {
var dog = ConnectionWatchdog(now: 0)
dog.frameArrived(at: 10)
#expect(dog.state(at: 11.9) == .live)
#expect(dog.state(at: 12) == .stalled(retryIn: 3))
let retry1 = dog.shouldRetry(at: 13)
#expect(!retry1)
let retry2 = dog.shouldRetry(at: 15)
#expect(retry2)
#expect(dog.state(at: 16) == .stalled(retryIn: 2))
let retry3 = dog.shouldRetry(at: 18)
#expect(retry3)
}
@Test func recoversWhenFramesReturn() {
var dog = ConnectionWatchdog(now: 0)
dog.frameArrived(at: 0)
let retry4 = dog.shouldRetry(at: 5)
#expect(retry4)
dog.frameArrived(at: 6)
#expect(dog.state(at: 6.5) == .live)
}
@Test func neverConnectedSourceAlsoRetries() {
var dog = ConnectionWatchdog(now: 0)
let retry5 = dog.shouldRetry(at: 5)
#expect(retry5)
}
}
@Suite struct RouteArgumentTests {
@Test func parsesRoutes() {
let args = ["Musubi", "--route", "ZOWIEBOX (PGM)=1", "-NSFoo", "YES", "--route", "MV=0"]
#expect(RouteArgument.parse(args) == [
.init(ndiName: "ZOWIEBOX (PGM)", screenIndex: 1),
.init(ndiName: "MV", screenIndex: 0),
])
}
@Test func ignoresMalformed() {
#expect(RouteArgument.parse(["--route", "noindex", "--route", "=1", "--route", "A=-1", "--route"]).isEmpty)
}
}
Loading…
Cancel
Save

Powered by TurnKey Linux.