diff --git a/Fluid.xcodeproj/project.pbxproj b/Fluid.xcodeproj/project.pbxproj index a2cb0a9f..fa32e09b 100644 --- a/Fluid.xcodeproj/project.pbxproj +++ b/Fluid.xcodeproj/project.pbxproj @@ -16,6 +16,7 @@ 7CDB0A2E2F3C4D5600FB7CAD /* AudioFixtureLoader.swift in Sources */ = {isa = PBXBuildFile; fileRef = 7CDB0A2A2F3C4D5600FB7CAD /* AudioFixtureLoader.swift */; }; 86CAA2D4EF18433096185602 /* LLMClientRequestBodyTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 343B29013F4441D6A797D12D /* LLMClientRequestBodyTests.swift */; }; 272BFB5CB271489892CAE50C /* TemperatureSupportTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 980330F3CE464336ADCE3E23 /* TemperatureSupportTests.swift */; }; + 5853C8F49B0743C69E9B8C77 /* StreamingResamplerTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 900080B1FF6B46458AE79921 /* StreamingResamplerTests.swift */; }; 7CDB0A2F2F3C4D5600FB7CAD /* dictation_fixture.wav in Resources */ = {isa = PBXBuildFile; fileRef = 7CDB0A2B2F3C4D5600FB7CAD /* dictation_fixture.wav */; }; 7CDB0A302F3C4D5600FB7CAD /* XCTest.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 7CDB0A2C2F3C4D5600FB7CAD /* XCTest.framework */; }; 7CE006BD2E80EBE600DDCCD6 /* AppUpdater in Frameworks */ = {isa = PBXBuildFile; productRef = 7CE006BC2E80EBE600DDCCD6 /* AppUpdater */; }; @@ -38,6 +39,7 @@ 7CDB0A292F3C4D5600FB7CAD /* DictationE2ETests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DictationE2ETests.swift; sourceTree = ""; }; 343B29013F4441D6A797D12D /* LLMClientRequestBodyTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = LLMClientRequestBodyTests.swift; sourceTree = ""; }; 980330F3CE464336ADCE3E23 /* TemperatureSupportTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TemperatureSupportTests.swift; sourceTree = ""; }; + 900080B1FF6B46458AE79921 /* StreamingResamplerTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = StreamingResamplerTests.swift; sourceTree = ""; }; 7CDB0A2A2F3C4D5600FB7CAD /* AudioFixtureLoader.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AudioFixtureLoader.swift; sourceTree = ""; }; 7CDB0A2B2F3C4D5600FB7CAD /* dictation_fixture.wav */ = {isa = PBXFileReference; lastKnownFileType = audio.wav; path = dictation_fixture.wav; sourceTree = ""; }; 7CDB0A2C2F3C4D5600FB7CAD /* XCTest.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = XCTest.framework; path = Platforms/MacOSX.platform/Developer/Library/Frameworks/XCTest.framework; sourceTree = DEVELOPER_DIR; }; @@ -110,6 +112,7 @@ 7C91B0022F42AA0100C0DEF0 /* HotkeyShortcutTests.swift */, 343B29013F4441D6A797D12D /* LLMClientRequestBodyTests.swift */, 980330F3CE464336ADCE3E23 /* TemperatureSupportTests.swift */, + 900080B1FF6B46458AE79921 /* StreamingResamplerTests.swift */, ); path = FluidDictationIntegrationTests; sourceTree = ""; @@ -266,6 +269,7 @@ 7C91B0012F42AA0100C0DEF0 /* HotkeyShortcutTests.swift in Sources */, 86CAA2D4EF18433096185602 /* LLMClientRequestBodyTests.swift in Sources */, 272BFB5CB271489892CAE50C /* TemperatureSupportTests.swift in Sources */, + 5853C8F49B0743C69E9B8C77 /* StreamingResamplerTests.swift in Sources */, ); runOnlyForDeploymentPostprocessing = 0; }; diff --git a/Sources/CoreAudioCaptureSupport/CoreAudioCaptureSupport.c b/Sources/CoreAudioCaptureSupport/CoreAudioCaptureSupport.c index 07f631fd..d8f4e90f 100644 --- a/Sources/CoreAudioCaptureSupport/CoreAudioCaptureSupport.c +++ b/Sources/CoreAudioCaptureSupport/CoreAudioCaptureSupport.c @@ -1,5 +1,6 @@ #include "include/CoreAudioCaptureSupport.h" +#include #include #include #include @@ -22,22 +23,33 @@ typedef struct { typedef struct { AudioObjectID deviceID; + AudioStreamID inputStreamID; AudioDeviceIOProcID ioProcID; AudioStreamBasicDescription format; uint32_t bufferFrameSize; uint32_t bytesPerSample; + bool virtualFormatListenerInstalled; + bool nominalSampleRateListenerInstalled; dispatch_semaphore_t packetSemaphore; + dispatch_queue_t listenerQueue; + AudioObjectPropertyListenerBlock formatChangedBlock; _Atomic uint64_t writeIndex; _Atomic uint64_t readIndex; _Atomic uint64_t droppedPackets; _Atomic bool running; + _Atomic bool formatChanged; FVPacketSlot slots[FV_RING_CAPACITY]; } FVCapture; static OSStatus fv_get_input_stream_format( AudioObjectID deviceID, - AudioStreamBasicDescription *format + AudioStreamBasicDescription *format, + AudioStreamID *outStreamID ) { + if (format == NULL) { + return kAudioHardwareBadObjectError; + } + AudioObjectPropertyAddress streamsAddress = { kAudioDevicePropertyStreams, kAudioObjectPropertyScopeInput, @@ -69,6 +81,9 @@ static OSStatus fv_get_input_stream_format( if (status != noErr || streamID == kAudioObjectUnknown) { return status != noErr ? status : kAudioHardwareBadObjectError; } + if (outStreamID != NULL) { + *outStreamID = streamID; + } AudioObjectPropertyAddress formatAddress = { kAudioStreamPropertyVirtualFormat, @@ -209,6 +224,10 @@ static OSStatus fv_io_proc( !atomic_load_explicit(&capture->running, memory_order_relaxed)) { return noErr; } + if (atomic_load_explicit(&capture->formatChanged, memory_order_acquire)) { + atomic_fetch_add_explicit(&capture->droppedPackets, 1, memory_order_relaxed); + return noErr; + } uint32_t frameCount = fv_frame_count(capture, inInputData); if (frameCount == 0) { @@ -282,6 +301,43 @@ static OSStatus fv_io_proc( return noErr; } +static bool fv_install_format_listener( + AudioObjectID objectID, + AudioObjectPropertySelector selector, + FVCapture *capture +) { + AudioObjectPropertyAddress address = { + selector, + kAudioObjectPropertyScopeGlobal, + kAudioObjectPropertyElementMain, + }; + OSStatus status = AudioObjectAddPropertyListenerBlock( + objectID, + &address, + capture->listenerQueue, + capture->formatChangedBlock + ); + return status == noErr; +} + +static void fv_remove_format_listener( + AudioObjectID objectID, + AudioObjectPropertySelector selector, + FVCapture *capture +) { + AudioObjectPropertyAddress address = { + selector, + kAudioObjectPropertyScopeGlobal, + kAudioObjectPropertyElementMain, + }; + (void) AudioObjectRemovePropertyListenerBlock( + objectID, + &address, + capture->listenerQueue, + capture->formatChangedBlock + ); +} + int32_t fv_core_audio_capture_create( AudioObjectID deviceID, FVCoreAudioCaptureRef *outCapture @@ -297,7 +353,11 @@ int32_t fv_core_audio_capture_create( } capture->deviceID = deviceID; - OSStatus status = fv_get_input_stream_format(deviceID, &capture->format); + OSStatus status = fv_get_input_stream_format( + deviceID, + &capture->format, + &capture->inputStreamID + ); if (status == noErr && !fv_format_is_supported(&capture->format, &capture->bytesPerSample)) { status = kAudioHardwareUnsupportedOperationError; @@ -324,6 +384,36 @@ int32_t fv_core_audio_capture_create( atomic_init(&capture->readIndex, 0); atomic_init(&capture->droppedPackets, 0); atomic_init(&capture->running, false); + atomic_init(&capture->formatChanged, false); + + capture->listenerQueue = dispatch_queue_create( + "com.fluidvoice.audio.format-listener", + DISPATCH_QUEUE_SERIAL + ); + if (capture->listenerQueue == NULL) { +#if !OS_OBJECT_USE_OBJC + dispatch_release(capture->packetSemaphore); +#endif + free(capture); + return kAudioHardwareUnspecifiedError; + } + AudioObjectPropertyListenerBlock formatChangedBlock = + ^(UInt32 inNumberAddresses, const AudioObjectPropertyAddress *inAddresses) { + (void) inNumberAddresses; + (void) inAddresses; + + atomic_store_explicit(&capture->formatChanged, true, memory_order_release); + dispatch_semaphore_signal(capture->packetSemaphore); + }; + capture->formatChangedBlock = Block_copy(formatChangedBlock); + if (capture->formatChangedBlock == NULL) { +#if !OS_OBJECT_USE_OBJC + dispatch_release(capture->listenerQueue); + dispatch_release(capture->packetSemaphore); +#endif + free(capture); + return kAudioHardwareUnspecifiedError; + } status = AudioDeviceCreateIOProcID( deviceID, @@ -332,13 +422,26 @@ int32_t fv_core_audio_capture_create( &capture->ioProcID ); if (status != noErr) { + Block_release(capture->formatChangedBlock); #if !OS_OBJECT_USE_OBJC + dispatch_release(capture->listenerQueue); dispatch_release(capture->packetSemaphore); #endif free(capture); return status; } + capture->virtualFormatListenerInstalled = fv_install_format_listener( + capture->inputStreamID, + kAudioStreamPropertyVirtualFormat, + capture + ); + capture->nominalSampleRateListenerInstalled = fv_install_format_listener( + capture->deviceID, + kAudioDevicePropertyNominalSampleRate, + capture + ); + *outCapture = (FVCoreAudioCaptureRef) capture; return noErr; } @@ -351,9 +454,79 @@ int32_t fv_core_audio_capture_start(FVCoreAudioCaptureRef captureRef) { if (atomic_load_explicit(&capture->running, memory_order_acquire)) { return noErr; } + dispatch_sync(capture->listenerQueue, ^{}); + atomic_store_explicit(&capture->formatChanged, false, memory_order_release); + + AudioStreamBasicDescription format; + uint32_t bytesPerSample = 0; + uint32_t bufferFrameSize = 0; + AudioStreamID newStreamID = kAudioObjectUnknown; + + OSStatus status = fv_get_input_stream_format( + capture->deviceID, + &format, + &newStreamID + ); + if (status == noErr && !fv_format_is_supported(&format, &bytesPerSample)) { + status = kAudioHardwareUnsupportedOperationError; + } + if (status == noErr) { + status = fv_get_buffer_frame_size(capture->deviceID, &bufferFrameSize); + } + if (status == noErr && + (bufferFrameSize == 0 || bufferFrameSize > FV_MAX_FRAMES_PER_PACKET)) { + status = kAudioHardwareUnsupportedOperationError; + } + if (status != noErr) { + return status; + } + + if (newStreamID != capture->inputStreamID) { + if (capture->virtualFormatListenerInstalled) { + fv_remove_format_listener( + capture->inputStreamID, + kAudioStreamPropertyVirtualFormat, + capture + ); + capture->virtualFormatListenerInstalled = false; + } + capture->inputStreamID = newStreamID; + capture->virtualFormatListenerInstalled = fv_install_format_listener( + capture->inputStreamID, + kAudioStreamPropertyVirtualFormat, + capture + ); + status = fv_get_input_stream_format( + capture->deviceID, + &format, + NULL + ); + if (status == noErr && !fv_format_is_supported(&format, &bytesPerSample)) { + status = kAudioHardwareUnsupportedOperationError; + } + if (status != noErr) { + return status; + } + } else if (!capture->virtualFormatListenerInstalled) { + capture->virtualFormatListenerInstalled = fv_install_format_listener( + capture->inputStreamID, + kAudioStreamPropertyVirtualFormat, + capture + ); + } + if (!capture->nominalSampleRateListenerInstalled) { + capture->nominalSampleRateListenerInstalled = fv_install_format_listener( + capture->deviceID, + kAudioDevicePropertyNominalSampleRate, + capture + ); + } + capture->format = format; + capture->bytesPerSample = bytesPerSample; + capture->bufferFrameSize = bufferFrameSize; atomic_store_explicit(&capture->running, true, memory_order_release); - OSStatus status = AudioDeviceStart(capture->deviceID, capture->ioProcID); + status = AudioDeviceStart(capture->deviceID, capture->ioProcID); if (status != noErr) { atomic_store_explicit(&capture->running, false, memory_order_release); dispatch_semaphore_signal(capture->packetSemaphore); @@ -388,11 +561,30 @@ void fv_core_audio_capture_destroy(FVCoreAudioCaptureRef captureRef) { if (atomic_load_explicit(&capture->running, memory_order_acquire)) { (void) fv_core_audio_capture_stop(captureRef); } + if (capture->virtualFormatListenerInstalled) { + fv_remove_format_listener( + capture->inputStreamID, + kAudioStreamPropertyVirtualFormat, + capture + ); + capture->virtualFormatListenerInstalled = false; + } + if (capture->nominalSampleRateListenerInstalled) { + fv_remove_format_listener( + capture->deviceID, + kAudioDevicePropertyNominalSampleRate, + capture + ); + capture->nominalSampleRateListenerInstalled = false; + } if (capture->ioProcID != NULL) { (void) AudioDeviceDestroyIOProcID(capture->deviceID, capture->ioProcID); capture->ioProcID = NULL; } + dispatch_sync(capture->listenerQueue, ^{}); + Block_release(capture->formatChangedBlock); #if !OS_OBJECT_USE_OBJC + dispatch_release(capture->listenerQueue); dispatch_release(capture->packetSemaphore); #endif free(capture); @@ -480,6 +672,12 @@ bool fv_core_audio_capture_is_running(FVCoreAudioCaptureRef captureRef) { atomic_load_explicit(&capture->running, memory_order_acquire); } +bool fv_core_audio_capture_format_changed(FVCoreAudioCaptureRef captureRef) { + FVCapture *capture = (FVCapture *) captureRef; + return capture != NULL && + atomic_load_explicit(&capture->formatChanged, memory_order_acquire); +} + double fv_core_audio_capture_sample_rate(FVCoreAudioCaptureRef captureRef) { const FVCapture *capture = (const FVCapture *) captureRef; return capture == NULL ? 0.0 : capture->format.mSampleRate; diff --git a/Sources/CoreAudioCaptureSupport/include/CoreAudioCaptureSupport.h b/Sources/CoreAudioCaptureSupport/include/CoreAudioCaptureSupport.h index f733297a..95e350c8 100644 --- a/Sources/CoreAudioCaptureSupport/include/CoreAudioCaptureSupport.h +++ b/Sources/CoreAudioCaptureSupport/include/CoreAudioCaptureSupport.h @@ -43,6 +43,7 @@ void fv_core_audio_capture_clear(FVCoreAudioCaptureRef capture); void fv_core_audio_capture_wake(FVCoreAudioCaptureRef capture); bool fv_core_audio_capture_is_running(FVCoreAudioCaptureRef capture); +bool fv_core_audio_capture_format_changed(FVCoreAudioCaptureRef capture); double fv_core_audio_capture_sample_rate(FVCoreAudioCaptureRef capture); uint32_t fv_core_audio_capture_buffer_frame_size(FVCoreAudioCaptureRef capture); uint64_t fv_core_audio_capture_dropped_packet_count(FVCoreAudioCaptureRef capture); diff --git a/Sources/Fluid/ContentView.swift b/Sources/Fluid/ContentView.swift index ba15c5b6..54a28de4 100644 --- a/Sources/Fluid/ContentView.swift +++ b/Sources/Fluid/ContentView.swift @@ -3147,9 +3147,9 @@ struct ContentView: View { if shouldPlayStartSound, !self.asr.isRunning { TranscriptionSoundPlayer.shared.playStartSound() } + self.captureRecordingContext() + self.prewarmPrivateAIDictationIfNeeded(for: .primary) await self.asr.start(onCaptureStarted: { - self.captureRecordingContext() - self.prewarmPrivateAIDictationIfNeeded(for: .primary) if shouldShowDictationOverlay { self.menuBarManager.showRecordingOverlayImmediately() } @@ -3758,14 +3758,14 @@ extension ContentView { if SettingsStore.shared.enableTranscriptionSounds, !self.asr.isRunning { TranscriptionSoundPlayer.shared.playStartSound() } + self.captureRecordingContext() + self.applyDictationPromptConfiguration(for: SettingsStore.shared.dictationPromptSelection(for: slot)) + self.prewarmPrivateAIDictationIfNeeded(for: slot) await self.asr.start(onCaptureStarted: { - self.captureRecordingContext() - self.applyDictationPromptConfiguration(for: SettingsStore.shared.dictationPromptSelection(for: slot)) self.appBench("overlay_mode_request mode=Dictation") self.menuBarManager.setOverlayMode(.dictation) self.menuBarManager.showRecordingOverlayImmediately() self.appBench("overlay_mode_requested mode=Dictation") - self.prewarmPrivateAIDictationIfNeeded(for: slot) }) if !self.asr.isRunning { self.menuBarManager.hideRecordingOverlayImmediately(reason: "asr_start_failed") diff --git a/Sources/Fluid/Services/ASRService.swift b/Sources/Fluid/Services/ASRService.swift index 6735e281..572df92e 100644 --- a/Sources/Fluid/Services/ASRService.swift +++ b/Sources/Fluid/Services/ASRService.swift @@ -735,7 +735,8 @@ final class ASRService: ObservableObject { } if let directAudioInput = self.directAudioInput, - directAudioInput.deviceID == device.id + directAudioInput.deviceID == device.id, + directAudioInput.formatChanged == false { return true } @@ -745,7 +746,14 @@ final class ASRService: ObservableObject { let pipeline = self.audioCapturePipeline do { - let directAudioInput = try DirectCoreAudioInput(deviceID: device.id) { samples, frameCount, sampleRate, inputHostTime, inputSampleTime in + let directAudioInput = try DirectCoreAudioInput( + deviceID: device.id, + onFormatChange: { [weak self] in + Task { @MainActor [weak self] in + self?.handleDirectInputFormatChanged() + } + } + ) { samples, frameCount, sampleRate, inputHostTime, inputSampleTime in pipeline.handle( samples: samples, frameCount: frameCount, @@ -815,6 +823,14 @@ final class ASRService: ObservableObject { self.activeAudioCaptureBackend = .audioEngine } + private func handleDirectInputFormatChanged() { + DebugLogger.shared.warning( + "Direct capture input format changed (Bluetooth route transition); rebuilding capture", + source: "ASRService" + ) + self.scheduleAudioRouteRecovery(reason: "input stream format changed") + } + private func stopActiveAudioCapture() { switch self.activeAudioCaptureBackend { case .directCoreAudio: @@ -899,6 +915,8 @@ final class ASRService: ObservableObject { private var engineConfigurationChangeObserver: NSObjectProtocol? private var audioRouteRecoveryTask: Task? private let audioRouteRecoveryDelayNanoseconds: UInt64 = 1_000_000_000 + private var pendingCaptureStartedCallback: (@MainActor () -> Void)? + private var captureStartedTimeoutTask: Task? private var audioEngineStandbyTask: Task? private let audioEngineStandbyNanoseconds: UInt64 = 8_000_000_000 private var isEngineTapInstalled = false @@ -931,14 +949,15 @@ final class ASRService: ObservableObject { /// from CoreAudio's realtime callback thread. private lazy var audioCapturePipeline: AudioCapturePipeline = .init( audioBuffer: self.audioBuffer, - onFirstAudio: { sessionID, sampleCount, frameLength, sampleRate, acquisitionMs, elapsedMs in - DispatchQueue.main.async { + onFirstAudio: { [weak self] sessionID, sampleCount, frameLength, sampleRate, acquisitionMs, elapsedMs in + DispatchQueue.main.async { [weak self] in let bufferMs = Int((Double(frameLength) / sampleRate * 1000).rounded()) DebugLogger.shared.benchmark( "ASR_BENCH", message: "session=\(sessionID) first_audio sampleCount=\(sampleCount) frameLength=\(frameLength) sampleRate=\(Int(sampleRate.rounded())) bufferMs=\(bufferMs) acquisitionMs=\(acquisitionMs) elapsedMs=\(elapsedMs)", source: "ASRBenchmark" ) + self?.fireCaptureStartedGate(timedOut: false) } }, onLevel: { [weak self] level in @@ -949,6 +968,44 @@ final class ASRService: ObservableObject { } ) + private func installCaptureStartedGate(_ callback: (@MainActor () -> Void)?) { + guard let callback else { return } + + self.pendingCaptureStartedCallback = callback + self.captureStartedTimeoutTask?.cancel() + self.captureStartedTimeoutTask = Task { [weak self] in + do { + try await Task.sleep(nanoseconds: 2_500_000_000) + } catch { + return + } + await MainActor.run { [weak self] in + self?.fireCaptureStartedGate(timedOut: true) + } + } + } + + private func fireCaptureStartedGate(timedOut: Bool) { + guard let callback = self.pendingCaptureStartedCallback else { return } + + self.pendingCaptureStartedCallback = nil + self.captureStartedTimeoutTask?.cancel() + self.captureStartedTimeoutTask = nil + if timedOut { + DebugLogger.shared.warning( + "Capture-start gate timed out waiting for first audio; proceeding", + source: "ASRService" + ) + } + callback() + } + + private func cancelCaptureStartedGate() { + self.pendingCaptureStartedCallback = nil + self.captureStartedTimeoutTask?.cancel() + self.captureStartedTimeoutTask = nil + } + init() { // CRITICAL FIX: Do NOT call any framework-triggering APIs here! // This includes: @@ -1228,11 +1285,11 @@ final class ASRService: ObservableObject { self.isDictionaryTrainingCaptureActive = false do { + self.installCaptureStartedGate(onCaptureStarted) try self.startPreferredAudioCapture() self.isRunning = true self.isDictionaryTrainingCaptureActive = forDictionaryTraining DebugLogger.shared.info("✅ Audio capture running", source: "ASRService") - onCaptureStarted?() // Pause only after capture is live so media control cannot delay the // first PCM packet. A quick stop while this await is in flight is @@ -1275,6 +1332,7 @@ final class ASRService: ObservableObject { self.isDictionaryTrainingCaptureActive = false self.audioCapturePipeline.setRecordingEnabled(false) self.isRunning = false + self.cancelCaptureStartedGate() self.stopActiveAudioCapture() self.retireAudioEngine(reason: "start_failed") DebugLogger.shared.error("Failed to start ASR session: \(error)", source: "ASRService") @@ -1383,6 +1441,7 @@ final class ASRService: ObservableObject { // Set isRunning to false before teardown so in-flight ASR chunks stop safely. DebugLogger.shared.debug("🚫 Setting isRunning = false...", source: "ASRService") self.isRunning = false + self.cancelCaptureStartedGate() DebugLogger.shared.debug("✅ isRunning disabled", source: "ASRService") // Stop monitoring device to prevent callbacks after stop @@ -1698,6 +1757,7 @@ final class ASRService: ObservableObject { // CRITICAL: Set isRunning to false FIRST to signal any in-flight chunks to abort early self.isRunning = false + self.cancelCaptureStartedGate() self.audioCapturePipeline.setRecordingEnabled(false) // Stop monitoring device @@ -3615,10 +3675,7 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { private var recordingSessionID: Int = 0 private var recordingStartHostTime: UInt64 = 0 private var recordingStopHostTime: UInt64? - private var resampleSourceRate: Double = 0 - private var resampleSourceFrameCursor: Int64 = 0 - private var resampleNextSourcePosition: Double = 0 - private var resamplePreviousSample: Float? + private let resampler = StreamingResampler() private var lastInputSampleEnd: Int64? // Smoothing state (kept off ASRService/@MainActor) @@ -3650,12 +3707,13 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { startHostTime: UInt64 = 0 ) { self.lock.lock() + let wasRecording = self.recordingEnabled if enabled { self.firstAudioReported = false self.recordingSessionID = sessionID self.recordingStartHostTime = startHostTime == 0 ? mach_absolute_time() : startHostTime self.recordingStopHostTime = nil - self.resetResamplerLocked() + self.resampler.reset() self.lastInputSampleEnd = nil self.recordingEnabled = true } @@ -3664,7 +3722,14 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { self.recordingSessionID = 0 self.recordingStartHostTime = 0 self.recordingStopHostTime = nil - self.resetResamplerLocked() + if wasRecording { + let tail = self.resampler.flush() + if tail.isEmpty == false { + self.audioBuffer.append(tail) + } + } else { + self.resampler.reset() + } self.lastInputSampleEnd = nil self.levelHistory.removeAll(keepingCapacity: true) self.smoothedLevel = 0.0 @@ -3775,14 +3840,11 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { { // Do not interpolate across a hardware discontinuity or a // packet dropped under extreme consumer backpressure. - self.resetResamplerLocked() + self.resampler.reset() } self.lastInputSampleEnd = inputSampleTime + Int64(acceptedRange.upperBound) } - let mono16k = self.resampleTo16kLocked( - acceptedSamples, - sourceSampleRate: sampleRate - ) + let mono16k = self.resampler.process(acceptedSamples, sourceRate: sampleRate) guard mono16k.isEmpty == false else { self.lock.unlock() return @@ -3791,9 +3853,9 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { if shouldReportFirstAudio { self.firstAudioReported = true } + self.audioBuffer.append(mono16k) self.lock.unlock() - self.audioBuffer.append(mono16k) if shouldReportFirstAudio { let acceptedHostTime = Self.hostTime( inputHostTime, @@ -3873,70 +3935,6 @@ private final nonisolated class AudioCapturePipeline: @unchecked Sendable { return Int((Double(end - start) / self.hostTicksPerSecond * 1000).rounded()) } - private func resetResamplerLocked() { - self.resampleSourceRate = 0 - self.resampleSourceFrameCursor = 0 - self.resampleNextSourcePosition = 0 - self.resamplePreviousSample = nil - } - - /// Stateful linear resampling keeps fractional phase across small hardware - /// callbacks. Stateless per-packet conversion silently shortens 44.1 kHz - /// recordings and introduces a discontinuity at every device cycle. - private func resampleTo16kLocked( - _ samples: [Float], - sourceSampleRate: Double - ) -> [Float] { - guard samples.isEmpty == false else { return [] } - if sourceSampleRate == 16_000.0 { - return samples - } - - if abs(self.resampleSourceRate - sourceSampleRate) > 0.5 { - self.resetResamplerLocked() - self.resampleSourceRate = sourceSampleRate - } - - let chunkStart = Double(self.resampleSourceFrameCursor) - let chunkEnd = chunkStart + Double(samples.count) - let step = sourceSampleRate / 16_000.0 - var output: [Float] = [] - output.reserveCapacity(Int(ceil(Double(samples.count) / step)) + 1) - - while self.resampleNextSourcePosition < chunkEnd { - let lowerFrame = Int64(floor(self.resampleNextSourcePosition)) - let fraction = Float(self.resampleNextSourcePosition - Double(lowerFrame)) - let localLower = lowerFrame - self.resampleSourceFrameCursor - - let lowerSample: Float - let upperSample: Float - if localLower < 0 { - guard localLower == -1, - let previousSample = self.resamplePreviousSample - else { break } - lowerSample = previousSample - upperSample = samples[0] - } else { - let index = Int(localLower) - guard index < samples.count else { break } - lowerSample = samples[index] - if fraction == 0 { - upperSample = lowerSample - } else { - guard index + 1 < samples.count else { break } - upperSample = samples[index + 1] - } - } - - output.append(lowerSample + (upperSample - lowerSample) * fraction) - self.resampleNextSourcePosition += step - } - - self.resampleSourceFrameCursor += Int64(samples.count) - self.resamplePreviousSample = samples.last - return output - } - private func calculateAudioLevel(_ samples: [Float]) -> CGFloat { guard samples.isEmpty == false else { return 0.0 } diff --git a/Sources/Fluid/Services/DirectCoreAudioInput.swift b/Sources/Fluid/Services/DirectCoreAudioInput.swift index 3e94e4ed..a1478b69 100644 --- a/Sources/Fluid/Services/DirectCoreAudioInput.swift +++ b/Sources/Fluid/Services/DirectCoreAudioInput.swift @@ -24,10 +24,18 @@ final class DirectCoreAudioInput { } let deviceID: AudioObjectID - let sampleRate: Double - let hardwareBufferFrameSize: UInt32 + var sampleRate: Double { + guard let capture else { return 0 } + return fv_core_audio_capture_sample_rate(capture) + } + + var hardwareBufferFrameSize: UInt32 { + guard let capture else { return 0 } + return fv_core_audio_capture_buffer_frame_size(capture) + } private var capture: FVCoreAudioCaptureRef? + private let onFormatChange: (@Sendable () -> Void)? private let packetHandler: PacketHandler private let workerQueue = DispatchQueue( label: "com.fluidvoice.audio.direct-input-consumer", @@ -35,7 +43,11 @@ final class DirectCoreAudioInput { ) private let workerGroup = DispatchGroup() - init(deviceID: AudioObjectID, packetHandler: @escaping PacketHandler) throws { + init( + deviceID: AudioObjectID, + onFormatChange: (@Sendable () -> Void)? = nil, + packetHandler: @escaping PacketHandler + ) throws { var capture: FVCoreAudioCaptureRef? let status = fv_core_audio_capture_create(deviceID, &capture) guard status == noErr, let capture else { @@ -44,8 +56,7 @@ final class DirectCoreAudioInput { self.deviceID = deviceID self.capture = capture - self.sampleRate = fv_core_audio_capture_sample_rate(capture) - self.hardwareBufferFrameSize = fv_core_audio_capture_buffer_frame_size(capture) + self.onFormatChange = onFormatChange self.packetHandler = packetHandler } @@ -63,6 +74,11 @@ final class DirectCoreAudioInput { return fv_core_audio_capture_dropped_packet_count(capture) } + var formatChanged: Bool { + guard let capture else { return false } + return fv_core_audio_capture_format_changed(capture) + } + func start() throws { guard let capture else { throw Self.error(status: kAudioHardwareBadObjectError, operation: "start direct Core Audio input") @@ -75,13 +91,18 @@ final class DirectCoreAudioInput { throw Self.error(status: status, operation: "start direct Core Audio input") } + let onFormatChange = self.onFormatChange let packetHandler = self.packetHandler let workerGroup = self.workerGroup let workerHandle = SendableCaptureHandle(rawValue: capture) workerGroup.enter() self.workerQueue.async { defer { workerGroup.leave() } - Self.consumePackets(capture: workerHandle.rawValue, packetHandler: packetHandler) + Self.consumePackets( + capture: workerHandle.rawValue, + onFormatChange: onFormatChange, + packetHandler: packetHandler + ) } } @@ -105,8 +126,11 @@ final class DirectCoreAudioInput { private nonisolated static func consumePackets( capture: FVCoreAudioCaptureRef, + onFormatChange: (@Sendable () -> Void)?, packetHandler: PacketHandler ) { + var didNotifyFormatChange = false + while true { var packet = FVCoreAudioPacket() while fv_core_audio_capture_peek(capture, &packet) { @@ -122,6 +146,11 @@ final class DirectCoreAudioInput { fv_core_audio_capture_consume(capture) } + if didNotifyFormatChange == false, fv_core_audio_capture_format_changed(capture) { + didNotifyFormatChange = true + onFormatChange?() + } + guard fv_core_audio_capture_is_running(capture) else { // AudioDeviceStop waits for the IOProc to leave. One final // acquire/drain above therefore captures the complete tail. diff --git a/Sources/Fluid/Services/StreamingResampler.swift b/Sources/Fluid/Services/StreamingResampler.swift new file mode 100644 index 00000000..b21356ea --- /dev/null +++ b/Sources/Fluid/Services/StreamingResampler.swift @@ -0,0 +1,197 @@ +import AVFoundation + +/// Streaming mono Float32 → 16 kHz mono Float32 conversion with proper +/// anti-alias filtering via AVAudioConverter. Stateful across calls so +/// fractional phase carries over between small hardware callbacks. +/// Not internally synchronized — the owner must serialize calls. +final class StreamingResampler { + private let targetRate = 16_000.0 + private var lastSourceRate: Double = 0 + + private struct ConverterState { + let sourceRate: Double + let converter: AVAudioConverter + let inputFormat: AVAudioFormat + let outputFormat: AVAudioFormat + var inputBuffer: AVAudioPCMBuffer + var outputBuffer: AVAudioPCMBuffer + } + + private var state: ConverterState? + + func process(_ samples: [Float], sourceRate: Double) -> [Float] { + defer { + self.lastSourceRate = sourceRate + } + + if sourceRate == self.targetRate { + if abs(self.lastSourceRate - sourceRate) > 0.5 { + self.state?.converter.reset() + } + return samples + } + guard samples.isEmpty == false, sourceRate > 0 else { return [] } + + let inputFrameCount = AVAudioFrameCount(samples.count) + let outputCapacity = AVAudioFrameCount( + Int(ceil(Double(samples.count) * self.targetRate / sourceRate)) + 64 + ) + guard var state = self.converterState( + for: sourceRate, + inputCapacity: inputFrameCount, + outputCapacity: outputCapacity + ) else { return [] } + + if inputFrameCount > state.inputBuffer.frameCapacity { + guard let inputBuffer = AVAudioPCMBuffer( + pcmFormat: state.inputFormat, + frameCapacity: inputFrameCount + ) else { return [] } + state.inputBuffer = inputBuffer + } + if outputCapacity > state.outputBuffer.frameCapacity { + guard let outputBuffer = AVAudioPCMBuffer( + pcmFormat: state.outputFormat, + frameCapacity: outputCapacity + ) else { return [] } + state.outputBuffer = outputBuffer + } + self.state = state + + let inputBuffer = state.inputBuffer + let outputBuffer = state.outputBuffer + outputBuffer.frameLength = 0 + inputBuffer.frameLength = inputFrameCount + + guard let inputChannel = inputBuffer.floatChannelData?[0] else { return [] } + samples.withUnsafeBufferPointer { buffer in + guard let baseAddress = buffer.baseAddress else { return } + inputChannel.update(from: baseAddress, count: samples.count) + } + + var servedInput = false + var conversionError: NSError? + let status = state.converter.convert(to: outputBuffer, error: &conversionError) { _, outStatus in + if servedInput { + outStatus.pointee = .noDataNow + return nil + } + + servedInput = true + outStatus.pointee = .haveData + return inputBuffer + } + + guard conversionError == nil, status != .error else { return [] } + guard let outputChannel = outputBuffer.floatChannelData?[0] else { return [] } + + return Array(UnsafeBufferPointer( + start: outputChannel, + count: Int(outputBuffer.frameLength) + )) + } + + func reset() { + self.state?.converter.reset() + self.lastSourceRate = 0 + } + + /// Drains samples buffered inside the converter (FIR group delay) and + /// resets it. Call at end of a recording segment so the tail of the + /// final word is not discarded. + func flush() -> [Float] { + defer { + self.lastSourceRate = 0 + } + + guard let state = self.state else { return [] } + + let outputBuffer: AVAudioPCMBuffer + if state.outputBuffer.frameCapacity >= 1024 { + outputBuffer = state.outputBuffer + } else if let buffer = AVAudioPCMBuffer( + pcmFormat: state.outputFormat, + frameCapacity: 1024 + ) { + outputBuffer = buffer + } else { + state.converter.reset() + return [] + } + outputBuffer.frameLength = 0 + + var conversionError: NSError? + let status = state.converter.convert(to: outputBuffer, error: &conversionError) { _, outStatus in + outStatus.pointee = .endOfStream + return nil + } + + defer { + state.converter.reset() + } + + guard conversionError == nil, status != .error else { return [] } + guard let outputChannel = outputBuffer.floatChannelData?[0] else { return [] } + + return Array(UnsafeBufferPointer( + start: outputChannel, + count: Int(outputBuffer.frameLength) + )) + } + + private func converterState( + for sourceRate: Double, + inputCapacity: AVAudioFrameCount, + outputCapacity: AVAudioFrameCount + ) -> ConverterState? { + if let state = self.state, + abs(state.sourceRate - sourceRate) <= 0.5 + { + if abs(self.lastSourceRate - sourceRate) > 0.5 { + state.converter.reset() + } + return state + } + + guard let inputFormat = AVAudioFormat( + standardFormatWithSampleRate: sourceRate, + channels: 1 + ), + let outputFormat = AVAudioFormat( + standardFormatWithSampleRate: self.targetRate, + channels: 1 + ), + let converter = AVAudioConverter(from: inputFormat, to: outputFormat) + else { + self.state = nil + return nil + } + + converter.primeMethod = .none + converter.sampleRateConverterQuality = AVAudioQuality.max.rawValue + + guard let inputBuffer = AVAudioPCMBuffer( + pcmFormat: inputFormat, + frameCapacity: inputCapacity + ), + let outputBuffer = AVAudioPCMBuffer( + pcmFormat: outputFormat, + frameCapacity: outputCapacity + ) + else { + self.state = nil + return nil + } + + let state = ConverterState( + sourceRate: sourceRate, + converter: converter, + inputFormat: inputFormat, + outputFormat: outputFormat, + inputBuffer: inputBuffer, + outputBuffer: outputBuffer + ) + self.state = state + return state + } +} diff --git a/Tests/FluidDictationIntegrationTests/StreamingResamplerTests.swift b/Tests/FluidDictationIntegrationTests/StreamingResamplerTests.swift new file mode 100644 index 00000000..652162e5 --- /dev/null +++ b/Tests/FluidDictationIntegrationTests/StreamingResamplerTests.swift @@ -0,0 +1,147 @@ +@testable import FluidVoice_Debug +import XCTest + +@MainActor +final class StreamingResamplerTests: XCTestCase { + func testPassthroughReturnsIdenticalArrayAt16kHz() { + let samples: [Float] = [0, 0.125, -0.25, 0.5, -0.75, 1, -1, 0.0625] + let resampler = StreamingResampler() + + let output = resampler.process(samples, sourceRate: 16_000) + + XCTAssertEqual(output, samples) + } + + func testLengthRatioFrom48kHzTo16kHz() { + let samples = Self.sineWave(frequency: 440, sampleRate: 48_000, duration: 1) + let resampler = StreamingResampler() + + let output = resampler.process(samples, sourceRate: 48_000) + + XCTAssertLessThanOrEqual(abs(output.count - 16_000), 32) + } + + func testAntiAliasingSuppresses10kHzFoldoverTo6kHz() { + let aliasedSource = Self.sineWave(frequency: 10_000, sampleRate: 24_000, duration: 1) + let inBandSource = Self.sineWave(frequency: 6000, sampleRate: 24_000, duration: 1) + + let aliasedOutput = StreamingResampler().process(aliasedSource, sourceRate: 24_000) + let inBandOutput = StreamingResampler().process(inBandSource, sourceRate: 24_000) + + let aliasedPower = Goertzel.power( + in: aliasedOutput, + targetFrequency: 6000, + sampleRate: 16_000 + ) + let inBandPower = Goertzel.power( + in: inBandOutput, + targetFrequency: 6000, + sampleRate: 16_000 + ) + + XCTAssertGreaterThan(inBandPower, 0) + XCTAssertLessThan(10 * log10(aliasedPower / inBandPower), -30) + } + + func testInBand4kHzTonePowerIsPreserved() { + let source = Self.sineWave(frequency: 4000, sampleRate: 48_000, duration: 1) + let groundTruth = Self.sineWave(frequency: 4000, sampleRate: 16_000, duration: 1) + + let output = StreamingResampler().process(source, sourceRate: 48_000) + + let convertedPower = Goertzel.power( + in: output, + targetFrequency: 4000, + sampleRate: 16_000 + ) + let groundTruthPower = Goertzel.power( + in: groundTruth, + targetFrequency: 4000, + sampleRate: 16_000 + ) + + XCTAssertGreaterThan(groundTruthPower, 0) + XCTAssertLessThan(abs(10 * log10(convertedPower / groundTruthPower)), 3) + } + + func testChunkedConversionMaintainsContinuity() { + let source = Self.sineWave(frequency: 440, sampleRate: 48_000, duration: 1) + let chunkedResampler = StreamingResampler() + var chunkedOutput: [Float] = [] + + var startIndex = 0 + while startIndex < source.count { + let endIndex = min(startIndex + 128, source.count) + chunkedOutput.append(contentsOf: chunkedResampler.process( + Array(source[startIndex.. [Float] { + let sampleCount = Int(sampleRate * duration) + return (0.. Double { + guard samples.isEmpty == false, sampleRate > 0 else { return 0 } + + let normalizedFrequency = targetFrequency / sampleRate + let coefficient = 2 * cos(2 * Double.pi * normalizedFrequency) + var q1 = 0.0 + var q2 = 0.0 + + for sample in samples { + let q0 = coefficient * q1 - q2 + Double(sample) + q2 = q1 + q1 = q0 + } + + return q1 * q1 + q2 * q2 - coefficient * q1 * q2 + } +}