feat(audio): audio matrix with NDI audio, talk key and channel strips

- NDIReceiver captures audio (audio-only bandwidth for sound routes)
- AUHAL output per device, mic input via AUHAL, fan-out to routes
- Drift-compensated, allocation-free render path; backlog trimmed on start
- Global talk key (Ctrl-Option-Space, hold or toggle) gates mic routes
- Sound tab with channel strips, meters, gain, mute, L/R/L+R
- --audio-route launch argument, per-minute audio CSV, 10 s diagnostics log

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
main
Samuel Zielke 3 days ago
parent 3d19a2252a
commit 2d6d0236d5

@ -0,0 +1,189 @@
import AudioMatrix
import AVFoundation
import Foundation
import MusubiCore
import NDIKit
import Observation
/// App-side glue for the audio matrix: NDI audio receivers, meters, talk key and stats.
@MainActor @Observable
final class AudioController {
let matrix = AudioMatrix()
let talkKey = TalkKey()
/// Smoothed peak meters in dBFS (route ID → level), updated 20× per second.
private(set) var meters: [UUID: Float] = [:]
private(set) var channelMeters: [String: Float] = [:] // "<deviceUID>#<channel>"
private(set) var dropouts: [UUID: Int] = [:]
/// Highest peak per route since the last stats line.
private(set) var minutePeaks: [UUID: Float] = [:]
var selectedDeviceUID: String?
private var ndiReceivers: [String: NDIReceiver] = [:]
private var meterTimer: Timer?
private var diagnosticsTimer: Timer?
private let recorder = AudioStatsRecorder()
init() {
selectedDeviceUID = Self.preferredOutput(in: matrix.outputDevices)?.uid
talkKey.onChange = { [weak self] active in self?.matrix.talkActive = active }
talkKey.register()
meterTimer = Timer.scheduledTimer(withTimeInterval: 0.05, repeats: true) { [weak self] _ in
MainActor.assumeIsolated { self?.updateMeters() }
}
diagnosticsTimer = Timer.scheduledTimer(withTimeInterval: 10, repeats: true) { [weak self] _ in
MainActor.assumeIsolated { self?.logDiagnostics() }
}
}
var selectedDevice: AudioDevice? {
matrix.outputDevices.first { $0.uid == selectedDeviceUID } ?? Self.preferredOutput(in: matrix.outputDevices)
}
func add(_ source: AudioMatrix.SourceRef, channel: Int) {
guard let device = selectedDevice else { return }
if case .input = source {
AVCaptureDevice.requestAccess(for: .audio) { _ in }
}
matrix.addRoute(source, deviceUID: device.uid, channel: channel)
syncNDIReceivers()
}
func remove(_ id: UUID) {
matrix.removeRoute(id)
syncNDIReceivers()
}
func label(for source: AudioMatrix.SourceRef) -> String {
switch source {
case .ndi(let name): Source.defaultLabel(for: name)
case .input(let uid): matrix.devices.first { $0.uid == uid }?.name ?? uid
}
}
/// One audio-only NDI receiver per NDI source that has routes.
private func syncNDIReceivers() {
let wanted = Set(matrix.ndiFanouts.keys)
for name in ndiReceivers.keys where !wanted.contains(name) {
ndiReceivers.removeValue(forKey: name)?.stop()
}
for name in wanted where ndiReceivers[name] == nil {
guard let fanout = matrix.ndiFanouts[name] else { continue }
ndiReceivers[name] = NDIReceiver(
source: Source(ndiName: name), receiverName: "Musubi Ton", bandwidth: .audioOnly,
onAudio: { chunk in
fanout.push(planar: chunk.data, channelStride: chunk.channelStride, channels: chunk.channels,
frames: chunk.frames, sampleRate: chunk.sampleRate)
}
)
Log.audio.info("NDI audio receiver for \(name, privacy: .public)")
}
}
private func updateMeters() {
let fall: Float = 1.2 // dB per tick
for route in matrix.routes {
let peak = matrix.takePeak(route.id)
meters[route.id] = max(peak, (meters[route.id] ?? Level.floorDB) - fall)
minutePeaks[route.id] = max(peak, minutePeaks[route.id] ?? Level.floorDB)
dropouts[route.id] = matrix.dropouts(route.id)
}
meters = meters.filter { id, _ in matrix.routes.contains { $0.id == id } }
if let device = selectedDevice {
for channel in 0..<device.outputChannels {
let key = "\(device.uid)#\(channel)"
let peak = matrix.takeChannelPeak(deviceUID: device.uid, channel: channel)
channelMeters[key] = max(peak, (channelMeters[key] ?? Level.floorDB) - fall)
}
}
recorder.recordIfDue(controller: self)
}
private func logDiagnostics() {
for route in matrix.routes {
guard let d = matrix.diagnostics(route.id) else { continue }
Log.audio.info("\(self.label(for: route.source), privacy: .public): buffered \(d.bufferedFrames) underruns \(d.underruns) overruns \(d.overruns) correction \(String(format: "%+.0f ppm", d.correction * 1_000_000), privacy: .public)")
}
}
func resetMinutePeaks() {
minutePeaks.removeAll()
}
/// Applies `--audio-route` launch arguments.
func applyLaunchRoutes(_ arguments: [String] = ProcessInfo.processInfo.arguments) {
for route in AudioRouteArgument.parse(arguments) {
guard let device = matrix.outputDevices.first(where: { $0.name.localizedCaseInsensitiveContains(route.device) }),
route.channel <= device.outputChannels
else {
Log.audio.error("--audio-route: no output \(route.device, privacy: .public) with channel \(route.channel)")
continue
}
let source: AudioMatrix.SourceRef
switch route.kind {
case .ndi:
source = .ndi(name: route.source)
case .input:
guard let input = matrix.inputDevices.first(where: { $0.name.localizedCaseInsensitiveContains(route.source) }) else {
Log.audio.error("--audio-route: no input \(route.source, privacy: .public)")
continue
}
source = .input(deviceUID: input.uid)
}
selectedDeviceUID = device.uid
add(source, channel: route.channel - 1)
}
}
func ndiAudioInfo(_ name: String) -> NDIReceiver.Snapshot? {
ndiReceivers[name]?.snapshot
}
/// MOTU first, then any multi-channel interface, then the system default.
private static func preferredOutput(in devices: [AudioDevice]) -> AudioDevice? {
devices.first { $0.name.localizedCaseInsensitiveContains("MOTU") || $0.name.localizedCaseInsensitiveContains("M2") }
?? devices.first { $0.name.localizedCaseInsensitiveContains("USB") }
?? devices.first
}
}
/// Appends one CSV line per audio route and minute to ~/Library/Logs/Musubi/audio-<date>.csv.
@MainActor
final class AudioStatsRecorder {
private var lastWrite = Date()
private let directory = FileManager.default.homeDirectoryForCurrentUser.appending(path: "Library/Logs/Musubi")
func recordIfDue(controller: AudioController) {
let now = Date()
guard now.timeIntervalSince(lastWrite) >= 60 else { return }
lastWrite = now
let routes = controller.matrix.routes
guard !routes.isEmpty else { return }
var lines = ""
for route in routes {
let source: String = switch route.source {
case .ndi(let name): "ndi:\(name)"
case .input(let uid): "input:\(uid)"
}
var rate = 0
if case .ndi(let name) = route.source { rate = controller.ndiAudioInfo(name)?.audioSampleRate ?? 0 }
let peak = controller.minutePeaks[route.id] ?? Level.floorDB
lines += "\(now.ISO8601Format()),\"\(source)\",\"\(route.deviceUID)\",\(route.channel + 1),\(rate),\(controller.dropouts[route.id] ?? 0),\(String(format: "%.1f", peak)),\(route.gainDB),\(route.muted)\n"
}
controller.resetMinutePeaks()
let file = directory.appending(path: "audio-\(now.formatted(.iso8601.year().month().day())).csv")
do {
try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
if !FileManager.default.fileExists(atPath: file.path) {
try "time,source,device,channel,source_rate,dropouts,peak_dbfs,gain_db,muted\n".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(lines.utf8))
} catch {
Log.audio.error("Audio CSV: \(error.localizedDescription, privacy: .public)")
}
}
}

@ -5,17 +5,32 @@ import SwiftUI
struct ContentView: View { struct ContentView: View {
let sourceList: SourceListModel let sourceList: SourceListModel
let outputs: OutputController let outputs: OutputController
let audio: AudioController
@AppStorage("detailTab") private var tab = DetailTab.screens
enum DetailTab: String { case screens, sound }
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: {
ScreensView(outputs: outputs) switch tab {
case .screens: ScreensView(outputs: outputs)
case .sound: SoundView(audio: audio)
}
} }
.navigationTitle("Musubi") .navigationTitle("Musubi")
.navigationSubtitle("NDI rein. Bild, Ton und Stimme raus.") .navigationSubtitle("NDI rein. Bild, Ton und Stimme raus.")
.toolbar { .toolbar {
ToolbarItem(placement: .principal) {
Picker("Ansicht", selection: $tab) {
Text("Bildschirme").tag(DetailTab.screens)
Text("Ton").tag(DetailTab.sound)
}
.pickerStyle(.segmented)
.fixedSize()
}
if outputs.isSleepBlocked { if outputs.isSleepBlocked {
ToolbarItem(placement: .primaryAction) { ToolbarItem(placement: .primaryAction) {
Text("Ruhezustand blockiert") Text("Ruhezustand blockiert")

@ -7,14 +7,16 @@ import SwiftUI
struct MusubiApp: App { struct MusubiApp: App {
@State private var sourceList = SourceListModel() @State private var sourceList = SourceListModel()
@State private var outputs = OutputController() @State private var outputs = OutputController()
@State private var audio = AudioController()
var body: some SwiftUI.Scene { var body: some SwiftUI.Scene {
Window("Musubi", id: "main") { Window("Musubi", id: "main") {
ContentView(sourceList: sourceList, outputs: outputs) ContentView(sourceList: sourceList, outputs: outputs, audio: audio)
.preferredColorScheme(.dark) .preferredColorScheme(.dark)
.task { .task {
sourceList.start() sourceList.start()
outputs.applyLaunchRoutes() outputs.applyLaunchRoutes()
audio.applyLaunchRoutes()
} }
} }
.defaultSize(width: 960, height: 600) .defaultSize(width: 960, height: 600)
@ -23,4 +25,5 @@ struct MusubiApp: App {
enum Log { enum Log {
static let ndi = Logger(subsystem: "studio.zielke.musubi", category: "ndi") static let ndi = Logger(subsystem: "studio.zielke.musubi", category: "ndi")
static let audio = Logger(subsystem: "studio.zielke.musubi", category: "audio")
} }

@ -0,0 +1,261 @@
import AudioMatrix
import DesignSystem
import SwiftUI
/// Channel strips of the selected output device; drop an NDI source or a microphone onto a strip.
struct SoundView: View {
let audio: AudioController
var body: some View {
ScrollView {
VStack(alignment: .leading, spacing: Theme.Space.card) {
header
MicrophoneChips(audio: audio)
if let device = audio.selectedDevice {
if let error = audio.matrix.errors[device.uid] {
Text(error).font(Theme.Font.body).foregroundStyle(Theme.Color.kohaku)
}
if audio.matrix.routes.isEmpty {
Text("Noch kein Ton geroutet. Zieh eine Quelle oder ein Mikrofon auf einen Kanal.")
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.nezumi)
}
LazyVGrid(columns: [GridItem(.adaptive(minimum: 300), spacing: Theme.Space.card)], spacing: Theme.Space.card) {
ForEach(0..<device.outputChannels, id: \.self) { channel in
ChannelStrip(audio: audio, device: device, channel: channel)
}
}
} else {
Text("Kein Ausgabegerät gefunden. Schließ das Audio-Interface an.")
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.nezumi)
}
}
.padding(Theme.Space.card)
.padding(.bottom, 120)
}
.background(Theme.Color.sumi)
.overlay(alignment: .bottomTrailing) {
TalkButton(talkKey: audio.talkKey).padding(Theme.Space.card)
}
}
private var header: some View {
HStack {
Text("Ton").font(Theme.Font.title).foregroundStyle(Theme.Color.washi)
Spacer()
Picker("Ausgabegerät", selection: Binding(
get: { audio.selectedDevice?.uid ?? "" },
set: { audio.selectedDeviceUID = $0 }
)) {
ForEach(audio.matrix.outputDevices) { device in
Text(verbatim: "\(device.name) · \(device.outputChannels) Kanäle").tag(device.uid)
}
}
.fixedSize()
}
}
}
private struct MicrophoneChips: View {
let audio: AudioController
var body: some View {
VStack(alignment: .leading, spacing: Theme.Space.unit) {
Text("Mikrofone – über die Sprechtaste")
.font(Theme.Font.caption)
.foregroundStyle(Theme.Color.nezumi)
ScrollView(.horizontal, showsIndicators: false) {
HStack(spacing: Theme.Space.unit) {
ForEach(audio.matrix.inputDevices) { device in
Text(verbatim: device.name)
.font(Theme.Font.caption)
.foregroundStyle(Theme.Color.washi)
.padding(.horizontal, Theme.Space.unit)
.padding(.vertical, Theme.Space.unit / 2)
.background(Theme.Color.kasumi, in: RoundedRectangle(cornerRadius: Theme.Radius.chip))
.overlay(RoundedRectangle(cornerRadius: Theme.Radius.chip).strokeBorder(Theme.Color.usuzumi))
.draggable("input:\(device.uid)")
}
}
}
}
}
}
private struct ChannelStrip: View {
let audio: AudioController
let device: AudioDevice
let channel: Int
@State private var targeted = false
private var routes: [AudioMatrix.Route] {
audio.matrix.routes.filter { $0.deviceUID == device.uid && $0.channel == channel }
}
var body: some View {
VStack(alignment: .leading, spacing: Theme.Space.unit) {
HStack {
Text(verbatim: "Out \(channel + 1)").font(Theme.Font.title).foregroundStyle(Theme.Color.washi)
Spacer()
LevelText(db: audio.channelMeters["\(device.uid)#\(channel)"] ?? Level.floorDB)
}
LevelMeter(db: audio.channelMeters["\(device.uid)#\(channel)"] ?? Level.floorDB)
Divider().overlay(Theme.Color.usuzumi)
if routes.isEmpty {
Text("Quelle hierher ziehen")
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.nezumi)
}
ForEach(routes) { route in
RouteRow(audio: audio, route: route)
}
}
.frame(maxWidth: .infinity, minHeight: 140, alignment: .topLeading)
.padding(Theme.Space.card)
.background(Theme.Color.kasumi, in: RoundedRectangle(cornerRadius: Theme.Radius.card))
.overlay(
RoundedRectangle(cornerRadius: Theme.Radius.card)
.strokeBorder(targeted || !routes.isEmpty ? Theme.Color.ai : Theme.Color.usuzumi,
lineWidth: routes.isEmpty && !targeted ? Theme.Stroke.line : Theme.Stroke.cord)
)
.dropDestination(for: String.self) { items, _ in
guard let item = items.first else { return false }
if item.hasPrefix("input:") {
audio.add(.input(deviceUID: String(item.dropFirst("input:".count))), channel: channel)
} else {
audio.add(.ndi(name: item), channel: channel)
}
return true
} isTargeted: { targeted = $0 }
}
}
private struct RouteRow: View {
let audio: AudioController
let route: AudioMatrix.Route
var body: some View {
VStack(alignment: .leading, spacing: Theme.Space.unit / 2) {
HStack(spacing: Theme.Space.unit) {
if case .input = route.source {
Image(systemName: "mic").foregroundStyle(Theme.Color.nezumi)
}
Text(verbatim: audio.label(for: route.source))
.font(Theme.Font.body)
.foregroundStyle(Theme.Color.washi)
.lineLimit(1)
Spacer()
Picker("Kanal", selection: Binding(get: { route.selection }, set: { audio.matrix.setSelection(route.id, $0) })) {
Text("L").tag(ChannelSelection.left)
Text("R").tag(ChannelSelection.right)
Text("L+R").tag(ChannelSelection.mono)
}
.labelsHidden()
.pickerStyle(.segmented)
.fixedSize()
Button {
audio.matrix.setMuted(route.id, !route.muted)
} label: {
Text("M").font(Theme.Font.caption.bold())
}
.buttonStyle(.bordered)
.tint(route.muted ? Theme.Color.kohaku : nil)
.help(route.muted ? "Stummschaltung aufheben" : "Stummschalten")
Button {
audio.remove(route.id)
} label: {
Image(systemName: "xmark")
}
.buttonStyle(.borderless)
.help("Route entfernen")
}
HStack(spacing: Theme.Space.unit) {
Slider(value: Binding(get: { Double(route.gainDB) }, set: { audio.matrix.setGain(route.id, db: Float($0)) }),
in: -60...12)
.controlSize(.small)
Text(verbatim: route.gainDB <= Level.floorDB ? "−∞ dB" : String(format: "%+.1f dB", route.gainDB))
.font(Theme.Font.number)
.foregroundStyle(Theme.Color.nezumi)
.frame(width: 70, alignment: .trailing)
}
LevelMeter(db: audio.meters[route.id] ?? Level.floorDB)
if let count = audio.dropouts[route.id], count > 0 {
Text("\(count) Aussetzer").font(Theme.Font.caption).foregroundStyle(Theme.Color.kohaku)
}
}
.padding(.vertical, Theme.Space.unit / 2)
}
}
/// Horizontal peak meter; amber close to 0 dBFS.
private struct LevelMeter: View {
let db: Float
var body: some View {
GeometryReader { geometry in
let fraction = CGFloat((max(db, Level.floorDB) - Level.floorDB) / -Level.floorDB)
ZStack(alignment: .leading) {
Capsule().fill(Theme.Color.usuzumi)
Capsule()
.fill(db > -3 ? Theme.Color.kohaku : Theme.Color.washi.opacity(0.7))
.frame(width: geometry.size.width * fraction)
}
}
.frame(height: 4)
}
}
private struct LevelText: View {
let db: Float
var body: some View {
Text(verbatim: db <= Level.floorDB ? "−∞" : String(format: "%.0f dBFS", db))
.font(Theme.Font.number)
.foregroundStyle(db > -3 ? Theme.Color.kohaku : Theme.Color.nezumi)
}
}
/// Large push-to-talk button; red only while talking.
private struct TalkButton: View {
let talkKey: TalkKey
@State private var pressing = false
var body: some View {
VStack(alignment: .trailing, spacing: Theme.Space.unit) {
VStack(spacing: Theme.Space.unit / 2) {
Image(systemName: talkKey.isActive ? "mic.fill" : "mic")
.font(.system(size: 28, weight: .semibold))
Text("Sprechen").font(Theme.Font.title)
Text(verbatim: talkKey.displayName).font(Theme.Font.caption)
}
.foregroundStyle(Theme.Color.washi)
.frame(width: 160, height: 120)
.background(talkKey.isActive ? Theme.Color.shu : Theme.Color.kasumi, in: RoundedRectangle(cornerRadius: Theme.Radius.card))
.overlay(RoundedRectangle(cornerRadius: Theme.Radius.card).strokeBorder(Theme.Color.usuzumi, lineWidth: Theme.Stroke.line))
.contentShape(RoundedRectangle(cornerRadius: Theme.Radius.card))
.gesture(
DragGesture(minimumDistance: 0)
.onChanged { _ in
guard !pressing else { return }
pressing = true
talkKey.press()
}
.onEnded { _ in
pressing = false
talkKey.release()
}
)
Picker("Sprechtaste", selection: Binding(get: { talkKey.mode }, set: { talkKey.mode = $0 })) {
Text("Gedrückt halten").tag(TalkKey.Mode.hold)
Text("Umschalten").tag(TalkKey.Mode.toggle)
}
.labelsHidden()
.pickerStyle(.segmented)
.fixedSize()
}
}
}

@ -0,0 +1,63 @@
import Carbon.HIToolbox
import Foundation
import Observation
/// Global talk key (works while Musubi is in the background). Default: ⌃⌥ Space.
///
/// Uses Carbon hot keys, which need no Accessibility permission and report press and release.
@MainActor @Observable
final class TalkKey {
enum Mode: String, CaseIterable { case hold, toggle }
var mode: Mode = Mode(rawValue: UserDefaults.standard.string(forKey: "talkMode") ?? "") ?? .hold {
didSet { UserDefaults.standard.set(mode.rawValue, forKey: "talkMode") }
}
private(set) var isActive = false
var onChange: ((Bool) -> Void)?
let displayName = "⌃⌥ Leertaste"
private var hotKey: EventHotKeyRef?
private var handler: EventHandlerRef?
func register() {
var types = [
EventTypeSpec(eventClass: OSType(kEventClassKeyboard), eventKind: UInt32(kEventHotKeyPressed)),
EventTypeSpec(eventClass: OSType(kEventClassKeyboard), eventKind: UInt32(kEventHotKeyReleased)),
]
InstallEventHandler(GetApplicationEventTarget(), talkKeyHandler, types.count, &types,
Unmanaged.passUnretained(self).toOpaque(), &handler)
let id = EventHotKeyID(signature: OSType(0x4D534249), id: 1) // 'MSBI'
RegisterEventHotKey(UInt32(kVK_Space), UInt32(controlKey | optionKey), id, GetApplicationEventTarget(), 0, &hotKey)
}
/// Button or hot key pressed.
func press() {
switch mode {
case .hold: set(true)
case .toggle: set(!isActive)
}
}
/// Button or hot key released.
func release() {
if mode == .hold { set(false) }
}
func set(_ active: Bool) {
guard active != isActive else { return }
isActive = active
onChange?(active)
}
fileprivate func handle(kind: UInt32) {
if kind == UInt32(kEventHotKeyPressed) { press() } else { release() }
}
}
private func talkKeyHandler(_: EventHandlerCallRef?, event: EventRef?, userData: UnsafeMutableRawPointer?) -> OSStatus {
guard let event, let userData else { return OSStatus(eventNotHandledErr) }
let kind = GetEventKind(event)
let key = Unmanaged<TalkKey>.fromOpaque(userData).takeUnretainedValue()
MainActor.assumeIsolated { key.handle(kind: kind) }
return noErr
}

@ -0,0 +1,94 @@
import CoreAudio
import Foundation
/// A Core Audio device as shown in the UI.
public struct AudioDevice: Identifiable, Hashable, Sendable {
public var id: String { uid }
public let deviceID: AudioDeviceID
public let uid: String
public let name: String
public let inputChannels: Int
public let outputChannels: Int
public let sampleRate: Double
}
public enum AudioDevices {
public static func all() -> [AudioDevice] {
var address = AudioObjectPropertyAddress(
mSelector: kAudioHardwarePropertyDevices,
mScope: kAudioObjectPropertyScopeGlobal,
mElement: kAudioObjectPropertyElementMain
)
var size: UInt32 = 0
guard AudioObjectGetPropertyDataSize(AudioObjectID(kAudioObjectSystemObject), &address, 0, nil, &size) == noErr else { return [] }
var ids = [AudioDeviceID](repeating: 0, count: Int(size) / MemoryLayout<AudioDeviceID>.size)
guard AudioObjectGetPropertyData(AudioObjectID(kAudioObjectSystemObject), &address, 0, nil, &size, &ids) == noErr else { return [] }
return ids.compactMap(device(for:))
}
public static func device(uid: String) -> AudioDevice? {
all().first { $0.uid == uid }
}
static func device(for id: AudioDeviceID) -> AudioDevice? {
guard let uid = string(id, kAudioDevicePropertyDeviceUID),
let name = string(id, kAudioObjectPropertyName)
else { return nil }
return AudioDevice(
deviceID: id, uid: uid, name: name,
inputChannels: channelCount(id, scope: kAudioObjectPropertyScopeInput),
outputChannels: channelCount(id, scope: kAudioObjectPropertyScopeOutput),
sampleRate: nominalSampleRate(id)
)
}
static func nominalSampleRate(_ id: AudioDeviceID) -> Double {
var address = AudioObjectPropertyAddress(mSelector: kAudioDevicePropertyNominalSampleRate,
mScope: kAudioObjectPropertyScopeGlobal, mElement: kAudioObjectPropertyElementMain)
var rate: Float64 = 0
var size = UInt32(MemoryLayout<Float64>.size)
return AudioObjectGetPropertyData(id, &address, 0, nil, &size, &rate) == noErr ? rate : 0
}
/// Calls `handler` on the main queue whenever devices are added or removed.
/// Keep the returned token alive as long as you want notifications.
public static func observeChanges(_ handler: @escaping @MainActor @Sendable () -> Void) -> AnyObject {
DeviceListObserver(handler: handler)
}
private static func string(_ id: AudioDeviceID, _ selector: AudioObjectPropertySelector) -> String? {
var address = AudioObjectPropertyAddress(mSelector: selector, mScope: kAudioObjectPropertyScopeGlobal,
mElement: kAudioObjectPropertyElementMain)
var value: Unmanaged<CFString>?
var size = UInt32(MemoryLayout<Unmanaged<CFString>?>.size)
guard AudioObjectGetPropertyData(id, &address, 0, nil, &size, &value) == noErr, let value else { return nil }
return value.takeRetainedValue() as String
}
private static func channelCount(_ id: AudioDeviceID, scope: AudioObjectPropertyScope) -> Int {
var address = AudioObjectPropertyAddress(mSelector: kAudioDevicePropertyStreamConfiguration,
mScope: scope, mElement: kAudioObjectPropertyElementMain)
var size: UInt32 = 0
guard AudioObjectGetPropertyDataSize(id, &address, 0, nil, &size) == noErr, size > 0 else { return 0 }
let raw = UnsafeMutableRawPointer.allocate(byteCount: Int(size), alignment: MemoryLayout<AudioBufferList>.alignment)
defer { raw.deallocate() }
guard AudioObjectGetPropertyData(id, &address, 0, nil, &size, raw) == noErr else { return 0 }
let list = UnsafeMutableAudioBufferListPointer(raw.assumingMemoryBound(to: AudioBufferList.self))
return list.reduce(0) { $0 + Int($1.mNumberChannels) }
}
}
private final class DeviceListObserver {
private var address = AudioObjectPropertyAddress(mSelector: kAudioHardwarePropertyDevices,
mScope: kAudioObjectPropertyScopeGlobal, mElement: kAudioObjectPropertyElementMain)
private let block: AudioObjectPropertyListenerBlock
init(handler: @escaping @MainActor @Sendable () -> Void) {
block = { _, _ in MainActor.assumeIsolated { handler() } }
AudioObjectAddPropertyListenerBlock(AudioObjectID(kAudioObjectSystemObject), &address, .main, block)
}
deinit {
AudioObjectRemovePropertyListenerBlock(AudioObjectID(kAudioObjectSystemObject), &address, .main, block)
}
}

@ -0,0 +1,28 @@
import os
/// Distributes one producer's audio to several feeds (one per route).
///
/// `push` may run on a real-time thread: it only try-locks and otherwise uses the last known list.
public final class AudioFanout: @unchecked Sendable {
private let feeds = OSAllocatedUnfairLock<[AudioFeed]>(initialState: [])
private var cached: [AudioFeed] = [] // producer thread only
public init() {}
public func add(_ feed: AudioFeed) {
feeds.withLock { $0.append(feed) }
}
public func remove(_ feed: AudioFeed) {
feeds.withLock { $0.removeAll { $0 === feed } }
}
public var isEmpty: Bool { feeds.withLock { $0.isEmpty } }
public func push(planar data: UnsafePointer<Float>, channelStride: Int, channels: Int, frames: Int, sampleRate: Double) {
if let current = feeds.withLockIfAvailable({ $0 }) { cached = current }
for feed in cached {
feed.push(planar: data, channelStride: channelStride, channels: channels, frames: frames, sampleRate: sampleRate)
}
}
}

@ -0,0 +1,46 @@
import CAudioRing
import Foundation
/// Producer side of one route: takes audio from any thread (NDI, input device) into a stereo ring buffer.
///
/// One feed has exactly one producer and one consumer (the route on the render thread).
public final class AudioFeed: @unchecked Sendable {
let ring: OpaquePointer
/// Sample rate of the audio in the ring. Written by the producer, read by the render thread.
private(set) var sampleRate: Double = 0
// Producer-only scratch for interleaving.
private var scratch: UnsafeMutablePointer<Float>
private var scratchFrames = 0
public init(capacitySeconds: Double = 1, maxSampleRate: Double = 96_000) {
ring = musubi_ring_create(UInt32(capacitySeconds * maxSampleRate), 2)!
scratchFrames = 4096
scratch = .allocate(capacity: scratchFrames * 2)
}
deinit {
musubi_ring_destroy(ring)
scratch.deallocate()
}
/// Pushes planar audio. Mono is duplicated to both channels; channels beyond two are ignored.
public func push(planar data: UnsafePointer<Float>, channelStride: Int, channels: Int, frames: Int, sampleRate: Double) {
guard channels > 0, frames > 0 else { return }
self.sampleRate = sampleRate
ensureScratch(frames)
let left = data
let right = channels > 1 ? data + channelStride : data
for i in 0..<frames {
scratch[i * 2] = left[i]
scratch[i * 2 + 1] = right[i]
}
musubi_ring_write(ring, scratch, UInt32(frames))
}
private func ensureScratch(_ frames: Int) {
guard frames > scratchFrames else { return }
scratch.deallocate()
scratchFrames = frames
scratch = .allocate(capacity: frames * 2)
}
}

@ -0,0 +1,221 @@
import CAudioRing
import Foundation
import Observation
import OSLog
/// Routes audio sources (NDI, input devices) to single channels of output devices.
///
/// NDI audio is pushed in by the caller through the `AudioFanout` returned from `fanout(forNDI:)`;
/// input devices are captured here. Input-device routes are gated by the talk key.
@MainActor @Observable
public final class AudioMatrix {
public enum SourceRef: Hashable, Sendable {
case ndi(name: String)
case input(deviceUID: String)
}
public struct Route: Identifiable, Equatable, Sendable {
public let id: UUID
public let source: SourceRef
public let deviceUID: String
public let channel: Int
public var selection: ChannelSelection
public var gainDB: Float
public var muted: Bool
}
public private(set) var devices: [AudioDevice] = []
public private(set) var routes: [Route] = []
public private(set) var errors: [String: String] = [:] // device UID → message
public var talkActive = false { didSet { applyGates() } }
/// Fanouts for NDI sources; the app feeds them from NDI receivers.
public private(set) var ndiFanouts: [String: AudioFanout] = [:]
private var renderers: [UUID: RouteRenderer] = [:]
/// Removed routes stay alive briefly so the render thread never releases (frees) them.
private var retired: [RouteRenderer] = []
private var outputEngines: [String: OutputEngine] = [:]
private var inputEngines: [String: InputEngine] = [:]
private var deviceObserver: AnyObject?
private let log = Logger(subsystem: "studio.zielke.musubi", category: "audio")
public init() {
devices = AudioDevices.all()
deviceObserver = AudioDevices.observeChanges { [weak self] in self?.devicesChanged() }
}
public var outputDevices: [AudioDevice] { devices.filter { $0.outputChannels > 0 } }
public var inputDevices: [AudioDevice] { devices.filter { $0.inputChannels > 0 } }
@discardableResult
public func addRoute(_ source: SourceRef, deviceUID: String, channel: Int, selection: ChannelSelection = .mono) -> UUID {
let id = UUID()
let latency: Double = if case .input = source { 0.012 } else { 0.08 }
let feed = AudioFeed()
let renderer = RouteRenderer(id: id, feed: feed, outputChannel: channel, selection: selection, targetLatency: latency)
renderers[id] = renderer
routes.append(Route(id: id, source: source, deviceUID: deviceUID, channel: channel, selection: selection, gainDB: 0, muted: false))
switch source {
case .ndi(let name):
let fanout = ndiFanouts[name] ?? AudioFanout()
ndiFanouts[name] = fanout
fanout.add(feed)
case .input(let uid):
inputEngine(for: uid)?.fanout.add(feed)
}
applyGates()
rebuildOutputs()
log.info("Route \(String(describing: source), privacy: .public) → \(deviceUID, privacy: .public) ch \(channel + 1)")
return id
}
public func removeRoute(_ id: UUID) {
guard let index = routes.firstIndex(where: { $0.id == id }), let renderer = renderers.removeValue(forKey: id) else { return }
let route = routes.remove(at: index)
retire(renderer)
switch route.source {
case .ndi(let name):
ndiFanouts[name]?.remove(renderer.feed)
if ndiFanouts[name]?.isEmpty == true { ndiFanouts[name] = nil }
case .input(let uid):
inputEngines[uid]?.fanout.remove(renderer.feed)
if !routes.contains(where: { $0.source == route.source }) {
inputEngines.removeValue(forKey: uid)?.stop()
}
}
rebuildOutputs()
}
public func setGain(_ id: UUID, db: Float) {
update(id) { $0.gainDB = db }
renderers[id]?.gain = Level.gain(fromDB: db)
}
public func setMuted(_ id: UUID, _ muted: Bool) {
update(id) { $0.muted = muted }
renderers[id]?.muted = muted
}
public func setSelection(_ id: UUID, _ selection: ChannelSelection) {
update(id) { $0.selection = selection }
renderers[id]?.selection = selection
}
/// Peak level of a route since the last call, in dBFS.
public func takePeak(_ id: UUID) -> Float {
guard let renderer = renderers[id] else { return Level.floorDB }
let peak = renderer.peak
renderer.peak = 0
return Level.db(fromGain: peak)
}
/// Peak level of an output channel since the last call, in dBFS.
public func takeChannelPeak(deviceUID: String, channel: Int) -> Float {
guard let engine = outputEngines[deviceUID], channel < engine.channels else { return Level.floorDB }
let peak = engine.channelPeaks[channel]
engine.channelPeaks[channel] = 0
return Level.db(fromGain: peak)
}
/// Under- and overruns of a route since it was created (dropouts = audible clicks).
public func dropouts(_ id: UUID) -> Int {
guard let renderer = renderers[id] else { return 0 }
return renderer.totalUnderruns + renderer.totalOverruns
}
public struct Diagnostics: Sendable {
public let underruns: Int
public let overruns: Int
public let bufferedFrames: Int
public let correction: Double
}
public func diagnostics(_ id: UUID) -> Diagnostics? {
guard let renderer = renderers[id] else { return nil }
return Diagnostics(underruns: renderer.totalUnderruns, overruns: renderer.totalOverruns,
bufferedFrames: Int(musubi_ring_available(renderer.feed.ring)), correction: renderer.correction)
}
// MARK: - Engines
private func retire(_ renderer: RouteRenderer) {
retired.append(renderer)
Task { [weak self] in
try? await Task.sleep(for: .seconds(2))
self?.retired.removeAll { $0 === renderer }
}
}
private func update(_ id: UUID, _ change: (inout Route) -> Void) {
guard let index = routes.firstIndex(where: { $0.id == id }) else { return }
change(&routes[index])
}
private func applyGates() {
for route in routes {
guard case .input = route.source else { continue }
renderers[route.id]?.gateOpen = talkActive
}
}
private func inputEngine(for uid: String) -> InputEngine? {
if let engine = inputEngines[uid] { return engine }
guard let device = devices.first(where: { $0.uid == uid }) else { return nil }
do {
let engine = try InputEngine(device: device)
inputEngines[uid] = engine
errors[uid] = nil
return engine
} catch {
errors[uid] = "\(device.name) lässt sich nicht öffnen."
log.error("Input \(device.name, privacy: .public): \(String(describing: error), privacy: .public)")
return nil
}
}
/// Starts engines for devices with routes, stops the others, and hands each engine its routes.
private func rebuildOutputs() {
let wanted = Set(routes.map(\.deviceUID))
for uid in outputEngines.keys where !wanted.contains(uid) {
outputEngines.removeValue(forKey: uid)?.stop()
}
for uid in wanted {
if outputEngines[uid] == nil, let device = devices.first(where: { $0.uid == uid }) {
do {
outputEngines[uid] = try OutputEngine(device: device)
errors[uid] = nil
log.info("Output \(device.name, privacy: .public) started at \(device.sampleRate) Hz")
} catch {
errors[uid] = "\(device.name) lässt sich nicht öffnen."
log.error("Output \(device.name, privacy: .public): \(String(describing: error), privacy: .public)")
}
}
outputEngines[uid]?.setRoutes(routes.filter { $0.deviceUID == uid }.compactMap { renderers[$0.id] })
}
}
/// A device was plugged in or out: drop engines of vanished devices, restart those that came back.
private func devicesChanged() {
devices = AudioDevices.all()
let present = Set(devices.map(\.uid))
for uid in outputEngines.keys where !present.contains(uid) {
outputEngines.removeValue(forKey: uid)?.stop()
log.notice("Output device \(uid, privacy: .public) disappeared")
}
for uid in inputEngines.keys where !present.contains(uid) {
inputEngines.removeValue(forKey: uid)?.stop()
log.notice("Input device \(uid, privacy: .public) disappeared")
}
// Reattach input routes of devices that are back (add is idempotent via remove first).
for route in routes {
guard case .input(let uid) = route.source, present.contains(uid),
let renderer = renderers[route.id], let engine = inputEngine(for: uid) else { continue }
engine.fanout.remove(renderer.feed)
engine.fanout.add(renderer.feed)
}
rebuildOutputs()
}
}

@ -38,6 +38,10 @@ public struct DriftController: Sendable {
if priming { if priming {
guard Double(available) >= targetFill, available >= needed else { return .fill } guard Double(available) >= targetFill, available >= needed else { return .fill }
priming = false priming = false
// NDI often delivers a backlog on connect: start at the target latency instead of draining slowly.
if Double(available) > targetFill * 1.5 {
return .skip(frames: available - Int(targetFill), step: nominalStep)
}
} else if available < needed { } else if available < needed {
underruns += 1 underruns += 1
priming = true priming = true
@ -52,9 +56,11 @@ public struct DriftController: Sendable {
} }
let error = (Double(available) - targetFill) / targetFill let error = (Double(available) - targetFill) / targetFill
smoothedError += 0.02 * (error - smoothedError) // Slow loop (seconds): NDI delivers audio in ~20 ms chunks, so the fill level is a sawtooth
integral = clamp(integral + smoothedError * 0.00002, maxCorrection) // that must not drive the pitch. Clock drift is tens to hundreds of ppm and changes slowly.
correction = clamp(smoothedError * 0.0005 + integral, maxCorrection) smoothedError += 0.005 * (error - smoothedError)
integral = clamp(integral + smoothedError * 0.000002, maxCorrection)
correction = clamp(smoothedError * 0.0003 + integral, maxCorrection)
return .play(step: nominalStep * (1 + correction)) return .play(step: nominalStep * (1 + correction))
} }

@ -0,0 +1,91 @@
import AudioToolbox
import CoreAudio
/// Captures up to two channels of an input device (e.g. headset mic) into an `AudioFanout`.
final class InputEngine: @unchecked Sendable {
let device: AudioDevice
let fanout = AudioFanout()
private let sampleRate: Double
private let channels: Int
private var unit: AudioUnit?
private let maxFrames = 4096
private let samples: UnsafeMutablePointer<Float>
private let bufferList: UnsafeMutableAudioBufferListPointer
init(device: AudioDevice, bufferFrames: UInt32 = 128) throws {
self.device = device
self.sampleRate = device.sampleRate
self.channels = min(max(device.inputChannels, 1), 2)
samples = .allocate(capacity: maxFrames * 2)
samples.initialize(repeating: 0, count: maxFrames * 2)
bufferList = AudioBufferList.allocate(maximumBuffers: channels)
var description = AudioComponentDescription(
componentType: kAudioUnitType_Output, componentSubType: kAudioUnitSubType_HALOutput,
componentManufacturer: kAudioUnitManufacturer_Apple, componentFlags: 0, componentFlagsMask: 0)
guard let component = AudioComponentFindNext(nil, &description) else { throw AudioError.noHAL }
var newUnit: AudioUnit?
try check(AudioComponentInstanceNew(component, &newUnit), "instance")
guard let newUnit else { throw AudioError.noHAL }
unit = newUnit
var enable: UInt32 = 1, disable: UInt32 = 0
let size = UInt32(MemoryLayout<UInt32>.size)
try check(AudioUnitSetProperty(newUnit, kAudioOutputUnitProperty_EnableIO, kAudioUnitScope_Input, 1, &enable, size), "enable input")
try check(AudioUnitSetProperty(newUnit, kAudioOutputUnitProperty_EnableIO, kAudioUnitScope_Output, 0, &disable, size), "disable output")
var deviceID = device.deviceID
try check(AudioUnitSetProperty(newUnit, kAudioOutputUnitProperty_CurrentDevice, kAudioUnitScope_Global, 0,
&deviceID, UInt32(MemoryLayout<AudioDeviceID>.size)), "device")
var format = AudioStreamBasicDescription.floatNonInterleaved(sampleRate: sampleRate, channels: channels)
try check(AudioUnitSetProperty(newUnit, kAudioUnitProperty_StreamFormat, kAudioUnitScope_Output, 1,
&format, UInt32(MemoryLayout<AudioStreamBasicDescription>.size)), "format")
var frames = bufferFrames
AudioUnitSetProperty(newUnit, kAudioDevicePropertyBufferFrameSize, kAudioUnitScope_Global, 0, &frames, size)
var callback = AURenderCallbackStruct(inputProc: inputCaptured, inputProcRefCon: Unmanaged.passUnretained(self).toOpaque())
try check(AudioUnitSetProperty(newUnit, kAudioOutputUnitProperty_SetInputCallback, kAudioUnitScope_Global, 0,
&callback, UInt32(MemoryLayout<AURenderCallbackStruct>.size)), "callback")
try check(AudioUnitInitialize(newUnit), "initialize")
try check(AudioOutputUnitStart(newUnit), "start")
}
deinit {
stop()
samples.deallocate()
free(bufferList.unsafeMutablePointer)
}
func stop() {
guard let unit else { return }
AudioOutputUnitStop(unit)
AudioUnitUninitialize(unit)
AudioComponentInstanceDispose(unit)
self.unit = nil
}
fileprivate func capture(flags: UnsafeMutablePointer<AudioUnitRenderActionFlags>, timestamp: UnsafePointer<AudioTimeStamp>, frames: UInt32) -> OSStatus {
guard let unit, Int(frames) <= maxFrames else { return noErr }
for channel in 0..<channels {
bufferList[channel] = AudioBuffer(mNumberChannels: 1, mDataByteSize: frames * 4,
mData: UnsafeMutableRawPointer(samples + channel * maxFrames))
}
let status = AudioUnitRender(unit, flags, timestamp, 1, frames, bufferList.unsafeMutablePointer)
guard status == noErr else { return status }
fanout.push(planar: samples, channelStride: maxFrames, channels: channels, frames: Int(frames), sampleRate: sampleRate)
return noErr
}
}
private func inputCaptured(
refCon: UnsafeMutableRawPointer,
flags: UnsafeMutablePointer<AudioUnitRenderActionFlags>,
timestamp: UnsafePointer<AudioTimeStamp>,
bus: UInt32,
frames: UInt32,
data: UnsafeMutablePointer<AudioBufferList>?
) -> OSStatus {
Unmanaged<InputEngine>.fromOpaque(refCon).takeUnretainedValue().capture(flags: flags, timestamp: timestamp, frames: frames)
}

@ -0,0 +1,119 @@
import AudioToolbox
import CoreAudio
import os
/// Plays the routes of one output device through an AUHAL unit.
final class OutputEngine: @unchecked Sendable {
let device: AudioDevice
let sampleRate: Double
let channels: Int
private var unit: AudioUnit?
private let routes = OSAllocatedUnfairLock<[RouteRenderer]>(initialState: [])
private var renderRoutes: [RouteRenderer] = [] // render thread only
/// Peak per output channel, written on the render thread, read and reset by the UI.
let channelPeaks: UnsafeMutablePointer<Float>
init(device: AudioDevice, bufferFrames: UInt32 = 256) throws {
self.device = device
self.sampleRate = device.sampleRate
self.channels = device.outputChannels
channelPeaks = .allocate(capacity: max(device.outputChannels, 1))
channelPeaks.initialize(repeating: 0, count: max(device.outputChannels, 1))
var description = AudioComponentDescription(
componentType: kAudioUnitType_Output, componentSubType: kAudioUnitSubType_HALOutput,
componentManufacturer: kAudioUnitManufacturer_Apple, componentFlags: 0, componentFlagsMask: 0)
guard let component = AudioComponentFindNext(nil, &description) else { throw AudioError.noHAL }
var newUnit: AudioUnit?
try check(AudioComponentInstanceNew(component, &newUnit), "instance")
guard let newUnit else { throw AudioError.noHAL }
unit = newUnit
var deviceID = device.deviceID
try check(AudioUnitSetProperty(newUnit, kAudioOutputUnitProperty_CurrentDevice, kAudioUnitScope_Global, 0,
&deviceID, UInt32(MemoryLayout<AudioDeviceID>.size)), "device")
var format = AudioStreamBasicDescription.floatNonInterleaved(sampleRate: sampleRate, channels: channels)
try check(AudioUnitSetProperty(newUnit, kAudioUnitProperty_StreamFormat, kAudioUnitScope_Input, 0,
&format, UInt32(MemoryLayout<AudioStreamBasicDescription>.size)), "format")
var frames = bufferFrames
AudioUnitSetProperty(newUnit, kAudioDevicePropertyBufferFrameSize, kAudioUnitScope_Global, 0,
&frames, UInt32(MemoryLayout<UInt32>.size))
var callback = AURenderCallbackStruct(inputProc: outputRender, inputProcRefCon: Unmanaged.passUnretained(self).toOpaque())
try check(AudioUnitSetProperty(newUnit, kAudioUnitProperty_SetRenderCallback, kAudioUnitScope_Input, 0,
&callback, UInt32(MemoryLayout<AURenderCallbackStruct>.size)), "callback")
try check(AudioUnitInitialize(newUnit), "initialize")
try check(AudioOutputUnitStart(newUnit), "start")
}
deinit {
stop()
channelPeaks.deallocate()
}
func stop() {
guard let unit else { return }
AudioOutputUnitStop(unit)
AudioUnitUninitialize(unit)
AudioComponentInstanceDispose(unit)
self.unit = nil
}
func setRoutes(_ newRoutes: [RouteRenderer]) {
routes.withLock { $0 = newRoutes }
}
fileprivate func render(frames: Int, buffers: UnsafeMutableAudioBufferListPointer) {
if let current = routes.withLockIfAvailable({ $0 }) { renderRoutes = current }
for buffer in buffers {
if let data = buffer.mData { memset(data, 0, Int(buffer.mDataByteSize)) }
}
for route in renderRoutes where route.outputChannel < buffers.count {
guard let data = buffers[route.outputChannel].mData else { continue }
route.render(into: data.assumingMemoryBound(to: Float.self), frames: frames, outputRate: sampleRate)
}
for (index, buffer) in buffers.enumerated() where index < channels {
guard let data = buffer.mData?.assumingMemoryBound(to: Float.self) else { continue }
var peak: Float = 0
for i in 0..<frames { peak = max(peak, abs(data[i])) }
channelPeaks[index] = max(channelPeaks[index], peak)
}
}
}
private func outputRender(
refCon: UnsafeMutableRawPointer,
flags: UnsafeMutablePointer<AudioUnitRenderActionFlags>,
timestamp: UnsafePointer<AudioTimeStamp>,
bus: UInt32,
frames: UInt32,
data: UnsafeMutablePointer<AudioBufferList>?
) -> OSStatus {
guard let data else { return noErr }
let engine = Unmanaged<OutputEngine>.fromOpaque(refCon).takeUnretainedValue()
engine.render(frames: Int(frames), buffers: UnsafeMutableAudioBufferListPointer(data))
return noErr
}
enum AudioError: Error {
case noHAL
case coreAudio(String, OSStatus)
}
func check(_ status: OSStatus, _ step: String) throws {
if status != noErr { throw AudioError.coreAudio(step, status) }
}
extension AudioStreamBasicDescription {
static func floatNonInterleaved(sampleRate: Double, channels: Int) -> AudioStreamBasicDescription {
AudioStreamBasicDescription(
mSampleRate: sampleRate, mFormatID: kAudioFormatLinearPCM,
mFormatFlags: kAudioFormatFlagIsFloat | kAudioFormatFlagIsPacked | kAudioFormatFlagIsNonInterleaved,
mBytesPerPacket: 4, mFramesPerPacket: 1, mBytesPerFrame: 4,
mChannelsPerFrame: UInt32(channels), mBitsPerChannel: 32, mReserved: 0)
}
}

@ -0,0 +1,106 @@
import CAudioRing
import Foundation
/// Real-time state of one route: feed → resampler → gain → one output channel.
///
/// `render` runs on the Core Audio render thread and never allocates or locks.
/// Settings are plain stored values written from the main thread (benign races on aligned scalars).
final class RouteRenderer: @unchecked Sendable {
let id: UUID
let feed: AudioFeed
let outputChannel: Int
let targetLatency: Double
var selection: ChannelSelection
var gain: Float = 1
var muted = false
/// Closed gate = silent (used for talkback).
var gateOpen = true
// Meter, read and reset by the UI.
var peak: Float = 0
private(set) var underruns = 0
private(set) var overruns = 0
private var drift: DriftController?
private var driftRates: (source: Double, output: Double) = (0, 0)
private var resampler = Resampler()
private var currentGain: Float = 0
private let input: UnsafeMutablePointer<Float>
private let output: UnsafeMutablePointer<Float>
private static let maxFrames = 8192
init(id: UUID, feed: AudioFeed, outputChannel: Int, selection: ChannelSelection, targetLatency: Double) {
self.id = id
self.feed = feed
self.outputChannel = outputChannel
self.selection = selection
self.targetLatency = targetLatency
input = .allocate(capacity: Self.maxFrames * 2 * 2 + 16)
output = .allocate(capacity: Self.maxFrames * 2)
}
deinit {
input.deallocate()
output.deallocate()
}
/// Mixes this route into `channel` (non-interleaved float buffer of the output device).
func render(into channel: UnsafeMutablePointer<Float>, frames requested: Int, outputRate: Double) {
let frames = min(requested, Self.maxFrames)
let sourceRate = feed.sampleRate
guard sourceRate > 0, outputRate > 0 else { return }
if drift == nil || driftRates != (sourceRate, outputRate) {
if let old = drift { underruns += old.underruns; overruns += old.overruns }
drift = DriftController(sourceRate: sourceRate, outputRate: outputRate, targetLatency: targetLatency)
driftRates = (sourceRate, outputRate)
resampler.reset()
}
let available = Int(musubi_ring_available(feed.ring))
let estimate = Int(Double(frames) * drift!.nominalStep * 1.002) + 2
let step: Double
switch drift!.update(available: available, needed: estimate) {
case .fill:
currentGain = 0
decayPeak()
return
case .play(let s):
step = s
case .skip(let skipFrames, let s):
musubi_ring_skip(feed.ring, UInt32(skipFrames))
step = s
}
let needed = resampler.inputFramesNeeded(outputFrames: frames, step: step)
let read = Int(musubi_ring_read(feed.ring, input, UInt32(needed)))
if read < needed {
// Cannot happen after the drift check, but never read garbage.
(input + read * 2).update(repeating: 0, count: (needed - read) * 2)
}
resampler.process(input: input, output: output, outputFrames: frames, step: step)
// Linear gain ramp over the block avoids zipper noise and clicks on mute/talk.
let target: Float = (muted || !gateOpen) ? 0 : gain
let start = currentGain
let delta = (target - start) / Float(frames)
var blockPeak: Float = 0
for i in 0..<frames {
let g = start + delta * Float(i + 1)
let sample = selection.sample(left: output[i * 2], right: output[i * 2 + 1]) * g
channel[i] += sample
blockPeak = max(blockPeak, abs(sample))
}
currentGain = target
peak = max(peak, blockPeak)
}
var correction: Double { drift?.correction ?? 0 }
var totalUnderruns: Int { underruns + (drift?.underruns ?? 0) }
var totalOverruns: Int { overruns + (drift?.overruns ?? 0) }
private func decayPeak() {
peak *= 0.9
}
}

@ -0,0 +1,34 @@
/// Parses `--audio-route "<kind>:<source>=<device>:<channel>"` launch arguments (scripted tests until scenes exist).
///
/// `kind` is `ndi` (NDI source name) or `input` (input device name, substring match).
/// `device` is an output device name (substring match), `channel` is 1-based.
/// Example: `--audio-route "ndi:ZOWIEBOX (PGM)=MOTU:2"`
public enum AudioRouteArgument {
public enum Kind: String, Sendable { case ndi, input }
public struct Parsed: Equatable, Sendable {
public var kind: Kind
public var source: String
public var device: String
public var channel: Int // 1-based
}
public static func parse(_ arguments: [String]) -> [Parsed] {
zip(arguments, arguments.dropFirst()).compactMap { flag, value in
guard flag == "--audio-route",
let colon = value.firstIndex(of: ":"),
let kind = Kind(rawValue: String(value[..<colon])),
let equals = value.lastIndex(of: "="),
equals > colon
else { return nil }
let source = String(value[value.index(after: colon)..<equals])
let target = value[value.index(after: equals)...]
guard let lastColon = target.lastIndex(of: ":"),
let channel = Int(target[target.index(after: lastColon)...]), channel >= 1
else { return nil }
let device = String(target[..<lastColon])
guard !source.isEmpty, !device.isEmpty else { return nil }
return Parsed(kind: kind, source: source, device: device, channel: channel)
}
}
}

@ -4,23 +4,45 @@ import Foundation
import MusubiCore import MusubiCore
import OSLog import OSLog
/// Receives video from one NDI source on a dedicated thread and hands out IOSurface-backed pixel buffers. /// Receives video and/or audio from one NDI source on dedicated threads.
/// ///
/// Video is handed out as IOSurface-backed pixel buffers, audio as planar float chunks.
/// Reconnects on its own when frames stop arriving (see `ConnectionWatchdog`). /// Reconnects on its own when frames stop arriving (see `ConnectionWatchdog`).
public final class NDIReceiver: @unchecked Sendable { public final class NDIReceiver: @unchecked Sendable {
public enum Bandwidth: Sendable {
case highest
/// Audio and metadata only; the sender transmits no video to this receiver.
case audioOnly
}
public struct Snapshot: Sendable, Equatable { public struct Snapshot: Sendable, Equatable {
public var state: ConnectionWatchdog.State public var state: ConnectionWatchdog.State
public var stats: FrameStats public var stats: FrameStats
public var width: Int public var width: Int
public var height: Int public var height: Int
public var frameRate: Double public var frameRate: Double
public var audioSampleRate: Int
public var audioChannels: Int
public var audioChunks: Int64
}
/// One block of planar 32-bit float audio. Only valid during the handler call.
public struct AudioChunk: @unchecked Sendable {
public let data: UnsafePointer<Float>
/// Distance between channel planes, in samples.
public let channelStride: Int
public let channels: Int
public let frames: Int
public let sampleRate: Double
} }
public typealias FrameHandler = @Sendable (CVPixelBuffer) -> Void public typealias FrameHandler = @Sendable (CVPixelBuffer) -> Void
public typealias AudioHandler = @Sendable (AudioChunk) -> Void
public let source: Source public let source: Source
private let instance: NDIlib_recv_instance_t private let instance: NDIlib_recv_instance_t
private let onFrame: FrameHandler private let onFrame: FrameHandler?
private let onAudio: AudioHandler?
private let log = Logger(subsystem: "studio.zielke.musubi", category: "ndi-recv") private let log = Logger(subsystem: "studio.zielke.musubi", category: "ndi-recv")
// Shared with the UI, guarded by `lock`. // Shared with the UI, guarded by `lock`.
@ -33,7 +55,13 @@ public final class NDIReceiver: @unchecked Sendable {
private var pool: CVPixelBufferPool? private var pool: CVPixelBufferPool?
private var poolKey: (width: Int, height: Int, format: OSType)? private var poolKey: (width: Int, height: Int, format: OSType)?
public init?(source: Source, receiverName: String, onFrame: @escaping FrameHandler) { public init?(
source: Source,
receiverName: String,
bandwidth: Bandwidth = .highest,
onFrame: FrameHandler? = nil,
onAudio: AudioHandler? = nil
) {
let created: NDIlib_recv_instance_t? = source.ndiName.withCString { name in let created: NDIlib_recv_instance_t? = source.ndiName.withCString { name in
receiverName.withCString { recvName in receiverName.withCString { recvName in
var ndiSource = NDIlib_source_t() var ndiSource = NDIlib_source_t()
@ -41,7 +69,7 @@ public final class NDIReceiver: @unchecked Sendable {
var settings = NDIlib_recv_create_v3_t( var settings = NDIlib_recv_create_v3_t(
source_to_connect_to: ndiSource, source_to_connect_to: ndiSource,
color_format: NDIlib_recv_color_format_UYVY_BGRA, color_format: NDIlib_recv_color_format_UYVY_BGRA,
bandwidth: NDIlib_recv_bandwidth_highest, bandwidth: bandwidth == .audioOnly ? NDIlib_recv_bandwidth_audio_only : NDIlib_recv_bandwidth_highest,
allow_video_fields: false, allow_video_fields: false,
p_ndi_recv_name: recvName p_ndi_recv_name: recvName
) )
@ -54,11 +82,24 @@ public final class NDIReceiver: @unchecked Sendable {
self.source = source self.source = source
self.instance = created self.instance = created
self.onFrame = onFrame self.onFrame = onFrame
self.onAudio = onAudio
self.watchdog = ConnectionWatchdog(now: now) self.watchdog = ConnectionWatchdog(now: now)
self.current = Snapshot(state: .connecting, stats: FrameStats(), width: 0, height: 0, frameRate: 0) self.current = Snapshot(state: .connecting, stats: FrameStats(), width: 0, height: 0, frameRate: 0,
audioSampleRate: 0, audioChannels: 0, audioChunks: 0)
// The first started loop also runs the watchdog. NDI allows capturing video and audio on separate threads.
let videoPrimary = onFrame != nil
if videoPrimary {
startThread(name: "studio.zielke.musubi.ndi-recv") { [self] in run(capturesVideo: true, isPrimary: true) }
}
if onAudio != nil {
startThread(name: "studio.zielke.musubi.ndi-audio") { [self] in run(capturesVideo: false, isPrimary: !videoPrimary) }
}
}
let thread = Thread { [self] in run() } private func startThread(name: String, _ body: @escaping @Sendable () -> Void) {
thread.name = "studio.zielke.musubi.ndi-recv" let thread = Thread(block: body)
thread.name = name
thread.qualityOfService = .userInteractive thread.qualityOfService = .userInteractive
thread.start() thread.start()
} }
@ -82,27 +123,16 @@ public final class NDIReceiver: @unchecked Sendable {
// MARK: - Receive thread // MARK: - Receive thread
private func run() { private func run(capturesVideo: Bool, isPrimary: Bool) {
var lastPerformanceCheck = 0.0 var lastPerformanceCheck = 0.0
while lock.withLock({ running }) { while lock.withLock({ running }) {
var frame = NDIlib_video_frame_v2_t() let now: Double
let type = NDIlib_recv_capture_v3(instance, &frame, nil, nil, 250) if capturesVideo {
let now = ProcessInfo.processInfo.systemUptime now = captureVideo()
} else {
if type == NDIlib_frame_type_video { now = captureAudio()
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)
} }
guard isPrimary else { continue }
if now - lastPerformanceCheck >= 1 { if now - lastPerformanceCheck >= 1 {
lastPerformanceCheck = now lastPerformanceCheck = now
@ -123,6 +153,56 @@ public final class NDIReceiver: @unchecked Sendable {
} }
} }
/// Captures one video frame (or times out). Returns the capture time.
private func captureVideo() -> Double {
var frame = NDIlib_video_frame_v2_t()
let type = NDIlib_recv_capture_v3(instance, &frame, nil, nil, 250)
let now = ProcessInfo.processInfo.systemUptime
guard type == NDIlib_frame_type_video else { return now }
if let onFrame, 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)
return now
}
/// Captures one audio frame (or times out). Returns the capture time.
private func captureAudio() -> Double {
var frame = NDIlib_audio_frame_v3_t()
let type = NDIlib_recv_capture_v3(instance, nil, &frame, nil, 250)
let now = ProcessInfo.processInfo.systemUptime
guard type == NDIlib_frame_type_audio else { return now }
if frame.FourCC == NDIlib_FourCC_audio_type_FLTP, let data = frame.p_data, frame.no_samples > 0, frame.no_channels > 0 {
data.withMemoryRebound(to: Float.self, capacity: Int(frame.channel_stride_in_bytes) / 4 * Int(frame.no_channels)) { samples in
onAudio?(AudioChunk(
data: samples,
channelStride: Int(frame.channel_stride_in_bytes) / MemoryLayout<Float>.size,
channels: Int(frame.no_channels),
frames: Int(frame.no_samples),
sampleRate: Double(frame.sample_rate)
))
}
}
lock.withLock {
if onFrame == nil { watchdog.frameArrived(at: now) }
current.audioSampleRate = Int(frame.sample_rate)
current.audioChannels = Int(frame.no_channels)
current.audioChunks += 1
}
NDIlib_recv_free_audio_v3(instance, &frame)
return now
}
private func reconnect() { private func reconnect() {
source.ndiName.withCString { name in source.ndiName.withCString { name in
var ndiSource = NDIlib_source_t() var ndiSource = NDIlib_source_t()

@ -80,6 +80,13 @@ import Testing
#expect(drift.update(available: 960, needed: 512) == .play(step: 1)) #expect(drift.update(available: 960, needed: 512) == .play(step: 1))
} }
@Test func trimsBacklogAfterPrimingWithoutCountingOverrun() {
var drift = DriftController(sourceRate: 48_000, outputRate: 48_000) // target 960
#expect(drift.update(available: 6700, needed: 512) == .skip(frames: 5740, step: 1))
#expect(drift.overruns == 0)
#expect(drift.update(available: 960, needed: 512) == .play(step: 1))
}
@Test func countsUnderrunAndRefills() { @Test func countsUnderrunAndRefills() {
var drift = DriftController(sourceRate: 48_000, outputRate: 48_000) var drift = DriftController(sourceRate: 48_000, outputRate: 48_000)
_ = drift.update(available: 1000, needed: 512) _ = drift.update(available: 1000, needed: 512)

@ -140,3 +140,23 @@ import Testing
#expect(RouteArgument.parse(["--route", "noindex", "--route", "=1", "--route", "A=-1", "--route"]).isEmpty) #expect(RouteArgument.parse(["--route", "noindex", "--route", "=1", "--route", "A=-1", "--route"]).isEmpty)
} }
} }
@Suite struct AudioRouteArgumentTests {
@Test func parsesNDIAndInputRoutes() {
let args = ["--audio-route", "ndi:ZOWIEBOX (PGM)=MOTU M2:2", "--audio-route", "input:Poly BT700=MOTU:1"]
#expect(AudioRouteArgument.parse(args) == [
.init(kind: .ndi, source: "ZOWIEBOX (PGM)", device: "MOTU M2", channel: 2),
.init(kind: .input, source: "Poly BT700", device: "MOTU", channel: 1),
])
}
@Test func sourceMayContainColonsAndEquals() {
let args = ["--audio-route", "ndi:HOST (a=b:c)=Out:1"]
#expect(AudioRouteArgument.parse(args).first?.source == "HOST (a=b:c)")
}
@Test func rejectsMalformed() {
let args = ["--audio-route", "foo:X=Dev:1", "--audio-route", "ndi:X=Dev:0", "--audio-route", "ndi:X=Dev", "--audio-route", "ndi:=Dev:1"]
#expect(AudioRouteArgument.parse(args).isEmpty)
}
}

@ -38,6 +38,7 @@ targets:
CFBundleVersion: "1" CFBundleVersion: "1"
LSApplicationCategoryType: public.app-category.video LSApplicationCategoryType: public.app-category.video
NSLocalNetworkUsageDescription: Musubi sucht NDI-Quellen im Technik-Netz. NSLocalNetworkUsageDescription: Musubi sucht NDI-Quellen im Technik-Netz.
NSMicrophoneUsageDescription: Musubi legt dein Headset-Mikro per Sprechtaste auf einen Ausgangskanal.
NSBonjourServices: [_ndi._tcp] NSBonjourServices: [_ndi._tcp]
settings: settings:
base: base:

Loading…
Cancel
Save

Powered by TurnKey Linux.