diff --git a/node/src/analyzer.ts b/node/src/analyzer.ts index 8b40be1..5a470e5 100644 --- a/node/src/analyzer.ts +++ b/node/src/analyzer.ts @@ -117,8 +117,11 @@ export class Collector extends FrameProcessor { this.collectingEnabled = enabled; } + /** True while the collector has collected some audio the analyzer can act on. */ get initialized(): boolean { - return this.hasBufferedAudio && this.nativeCollector !== null; + return ( + this.collectingEnabled && this.hasBufferedAudio && this.nativeCollector !== null + ); } get currentStreamInfo(): FrameProcessorStreamInfo | null { diff --git a/node/test/analyzer.test.ts b/node/test/analyzer.test.ts index 03cde46..42e0d08 100644 --- a/node/test/analyzer.test.ts +++ b/node/test/analyzer.test.ts @@ -223,6 +223,42 @@ describe("Analyzer", () => { analyzer.close(); }); + it("pauses analysis while the collector is disabled", () => { + const analyzer = new Analyzer({ + model: { getId: () => "analysis-test-model" } as never, + licenseKey: "test-license", + analysisInterval: 0.01, + }); + const events: AnalysisEvent[] = []; + analyzer.on("analysisResult", (event) => events.push(event)); + analyzer.collector.process(makeFrame()); + + vi.advanceTimersByTime(10); + expect(sdk.analyzers[0]!.analyzeCalls).toBe(1); + + analyzer.collector.setEnabled(false); + const frame = makeFrame(); + expect(analyzer.collector.process(frame)).toBe(frame); + vi.advanceTimersByTime(50); + + expect(sdk.collectors[0]!.blocks).toHaveLength(1); + expect(sdk.analyzers[0]!.analyzeCalls).toBe(1); + expect(events).toHaveLength(1); + + // Re-enabling drops the stale buffer and waits for fresh audio. + analyzer.collector.setEnabled(true); + expect(sdk.analyzers[0]!.resetCalls).toBe(1); + vi.advanceTimersByTime(50); + expect(sdk.analyzers[0]!.analyzeCalls).toBe(1); + + analyzer.collector.process(makeFrame()); + vi.advanceTimersByTime(10); + expect(sdk.analyzers[0]!.analyzeCalls).toBe(2); + expect(events).toHaveLength(2); + + analyzer.close(); + }); + it("stops the analyzer when RoomIO closes its collector", () => { const analyzer = new Analyzer({ model: { getId: () => "analysis-test-model" } as never, diff --git a/python/src/livekit/plugins/ai_coustics/analyzer.py b/python/src/livekit/plugins/ai_coustics/analyzer.py index d225130..094bebd 100644 --- a/python/src/livekit/plugins/ai_coustics/analyzer.py +++ b/python/src/livekit/plugins/ai_coustics/analyzer.py @@ -94,7 +94,9 @@ def enabled(self, value: bool) -> None: @property def initialized(self) -> bool: - return self._has_buffered_audio and self._collector is not None + """True while the collector is feeding fresh audio the analyzer can act on.""" + + return self._enabled and self._has_buffered_audio and self._collector is not None @property def stream_info(self) -> dict[str, str]: diff --git a/python/tests/test_analyzer.py b/python/tests/test_analyzer.py index 94e3f33..f759f8d 100644 --- a/python/tests/test_analyzer.py +++ b/python/tests/test_analyzer.py @@ -267,6 +267,56 @@ async def test_records_failed_analysis_metrics( assert not metric_instruments["score"].measurements +@pytest.mark.asyncio +async def test_disabled_collector_pauses_analysis( + fake_sdk: tuple[FakeCollector, FakeNativeAnalyzer, list[int]], +) -> None: + native_collector, native_analyzer, _ = fake_sdk + analyzer = Analyzer( + model=cast(aic_sdk.Model, FakeModel()), + license_key="test-license", + analysis_interval=0.01, + ) + events: list[AnalysisEvent] = [] + analyzer.on("analysis_result", events.append) + analyzer.collector._process(make_frame()) + + for _ in range(20): + if native_analyzer.analyze_calls: + break + await asyncio.sleep(0.005) + assert native_analyzer.analyze_calls >= 1 + + analyzer.collector.enabled = False + frame = make_frame() + assert analyzer.collector._process(frame) is frame + assert len(native_collector.blocks) == 1 + + # Let an analysis that was already in flight drain, then confirm no new ones start. + await asyncio.sleep(0.03) + calls_while_disabled = native_analyzer.analyze_calls + events_while_disabled = len(events) + await asyncio.sleep(0.06) + assert native_analyzer.analyze_calls == calls_while_disabled + assert len(events) == events_while_disabled + + # Re-enabling drops the stale buffer and waits for fresh audio. + analyzer.collector.enabled = True + assert native_analyzer.reset_calls == 1 + await asyncio.sleep(0.06) + assert native_analyzer.analyze_calls == calls_while_disabled + + analyzer.collector._process(make_frame()) + for _ in range(20): + if native_analyzer.analyze_calls > calls_while_disabled: + break + await asyncio.sleep(0.005) + + await asyncio.wait_for(analyzer.aclose(), timeout=1.0) + assert native_analyzer.analyze_calls > calls_while_disabled + assert len(events) > events_while_disabled + + @pytest.mark.asyncio async def test_room_closing_collector_stops_analyzer( fake_sdk: tuple[FakeCollector, FakeNativeAnalyzer, list[int]],