From 3043f1f83aed0009ed428d73785dcdbfb08a44dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oscar=20S=C3=B6derlund?= Date: Thu, 6 Aug 2026 09:18:31 +0200 Subject: [PATCH] feat(capture): dynamic source-level mic capture Replace single-mic tracking (default source only) with multi-source discovery mirroring how sink capture works. - Add ListSources to parec protocol client (filters out sink monitors) - Replace single mic/micName fields with mics map[string]*reader - reconcileMics discovers and captures all input sources dynamically - Fixes Bluetooth headset mic not being captured when it's exposed as a PipeWire Filter rather than the default source --- internal/capture/capture.go | 2 +- internal/capture/parec.go | 29 ++++---- internal/capture/parec_test.go | 95 +++++++++++++------------- internal/capture/reconcile.go | 95 ++++++++++++++------------ internal/protocol/parec/client.go | 37 ++++++++++ internal/protocol/parec/client_test.go | 56 +++++++++++++++ 6 files changed, 206 insertions(+), 108 deletions(-) diff --git a/internal/capture/capture.go b/internal/capture/capture.go index 46a0a9a..a26f550 100644 --- a/internal/capture/capture.go +++ b/internal/capture/capture.go @@ -17,6 +17,6 @@ type Source interface { // Narrowed for testability; *parec.Client satisfies it structurally. type sinkClient interface { ListSinks(ctx context.Context, req parec.ListSinksRequest) (parec.ListSinksResponse, error) - GetDefaultSource(ctx context.Context, req parec.GetDefaultSourceRequest) (parec.GetDefaultSourceResponse, error) + ListSources(ctx context.Context, req parec.ListSourcesRequest) (parec.ListSourcesResponse, error) StartCapture(ctx context.Context, req parec.StartCaptureRequest) (*parec.CaptureStream, error) } diff --git a/internal/capture/parec.go b/internal/capture/parec.go index ebfca4c..c655e20 100644 --- a/internal/capture/parec.go +++ b/internal/capture/parec.go @@ -15,14 +15,13 @@ import ( var tickInterval = time.Second // Parec implements Source by dynamically monitoring every PulseAudio sink -// plus the default microphone, mixing simultaneous sink audio together. +// plus all input sources (microphones), mixing simultaneous audio together. type Parec struct { client sinkClient - mu sync.Mutex - sinks map[string]*reader // keyed by sink name - mic *reader - micName string + mu sync.Mutex + sinks map[string]*reader // keyed by sink name + mics map[string]*reader // keyed by source name sinkBackoff backoff sinkLastTry time.Time @@ -36,7 +35,7 @@ type Parec struct { // NewParec creates a Parec source using the given parec protocol client. func NewParec(client *parec.Client) *Parec { - return &Parec{client: client, sinks: make(map[string]*reader)} + return &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} } // Start begins dynamic sink and microphone capture, returning a channel of @@ -83,17 +82,15 @@ func (c *Parec) mixTick() frame.Dual { sinkFrames = append(sinkFrames, data) } } - micReader := c.mic - c.mu.Unlock() - - mic := frame.Silent(pcm.FrameBytes) - if micReader != nil { - if data, ok := micReader.take(); ok { - mic = data + micFrames := make([][]byte, 0, len(c.mics)) + for _, r := range c.mics { + if data, ok := r.take(); ok { + micFrames = append(micFrames, data) } } + c.mu.Unlock() - return frame.Dual{Sys: pcm.Mix(sinkFrames...), Mic: mic} + return frame.Dual{Sys: pcm.Mix(sinkFrames...), Mic: pcm.Mix(micFrames...)} } // Stop terminates all capture streams. Idempotent. @@ -108,8 +105,8 @@ func (c *Parec) Stop() error { for _, r := range c.sinks { _ = r.stop() } - if c.mic != nil { - _ = c.mic.stop() + for _, r := range c.mics { + _ = r.stop() } c.mu.Unlock() }) diff --git a/internal/capture/parec_test.go b/internal/capture/parec_test.go index c6da817..ac42826 100644 --- a/internal/capture/parec_test.go +++ b/internal/capture/parec_test.go @@ -96,8 +96,8 @@ type fakeSinkClient struct { client *parec.Client sinks []parec.Sink sinksErr error - source string - sourceErr error + sources []parec.Source + sourcesErr error startErrFn func(device string) error } @@ -115,16 +115,16 @@ func (f *fakeSinkClient) ListSinks(ctx context.Context, _ parec.ListSinksRequest return parec.ListSinksResponse{Sinks: f.sinks}, nil } -func (f *fakeSinkClient) GetDefaultSource( +func (f *fakeSinkClient) ListSources( ctx context.Context, - _ parec.GetDefaultSourceRequest, -) (parec.GetDefaultSourceResponse, error) { + _ parec.ListSourcesRequest, +) (parec.ListSourcesResponse, error) { f.mu.Lock() defer f.mu.Unlock() - if f.sourceErr != nil { - return parec.GetDefaultSourceResponse{}, f.sourceErr + if f.sourcesErr != nil { + return parec.ListSourcesResponse{}, f.sourcesErr } - return parec.GetDefaultSourceResponse{Source: f.source}, nil + return parec.ListSourcesResponse{Sources: f.sources}, nil } func (f *fakeSinkClient) StartCapture( @@ -154,16 +154,16 @@ func (f *fakeSinkClient) setSinksErr(err error) { f.sinksErr = err } -func (f *fakeSinkClient) setSource(name string) { +func (f *fakeSinkClient) setSources(sources ...parec.Source) { f.mu.Lock() defer f.mu.Unlock() - f.source = name + f.sources = sources } -func (f *fakeSinkClient) setSourceErr(err error) { +func (f *fakeSinkClient) setSourcesErr(err error) { f.mu.Lock() defer f.mu.Unlock() - f.sourceErr = err + f.sourcesErr = err } // waitFor polls cond until it returns true or the timeout elapses. @@ -206,7 +206,7 @@ func TestParec_MultiSinkMixing(t *testing.T) { parec.Sink{Name: "b", MonitorSource: "b.monitor"}, ) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} frames, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -227,7 +227,7 @@ func TestParec_SinkAppears(t *testing.T) { client := newFakeSinkClient() client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x05, 0x00}, pcm.FrameBytes/2)) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} frames, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -259,7 +259,7 @@ func TestParec_SinkDisappears(t *testing.T) { parec.Sink{Name: "b", MonitorSource: "b.monitor"}, ) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} if _, err := src.Start(context.Background()); err != nil { t.Fatal(err) } @@ -288,7 +288,7 @@ func TestParec_SinkReappearsNewIndex(t *testing.T) { client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x05, 0x00}, pcm.FrameBytes/2)) client.setSinks(parec.Sink{Name: "a", MonitorSource: "a.monitor"}) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} _, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -319,14 +319,14 @@ func TestParec_SinkReappearsNewIndex(t *testing.T) { }) } -func TestParec_MicHotSwapSuccess(t *testing.T) { +func TestParec_MicSourceAppears(t *testing.T) { withFastPolling(t) client := newFakeSinkClient() client.runner.setStream("mic1", bytes.Repeat([]byte{0x07, 0x00}, pcm.FrameBytes/2)) client.runner.setStream("mic2", bytes.Repeat([]byte{0x08, 0x00}, pcm.FrameBytes/2)) - client.setSource("mic1") + client.setSources(parec.Source{Name: "mic1"}) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} frames, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -336,36 +336,38 @@ func TestParec_MicHotSwapSuccess(t *testing.T) { waitFor(t, time.Second, func() bool { src.mu.Lock() defer src.mu.Unlock() - return src.micName == "mic1" + _, ok := src.mics["mic1"] + return ok }) - client.setSource("mic2") + client.setSources(parec.Source{Name: "mic1"}, parec.Source{Name: "mic2"}) waitFor(t, time.Second, func() bool { src.mu.Lock() defer src.mu.Unlock() - return src.micName == "mic2" + _, ok := src.mics["mic2"] + return ok }) found := false for range 200 { f := <-frames - if f.Mic[0] == 0x08 { + if f.Mic[0] == 0x08 || f.Mic[0] == 0x0f { found = true break } } if !found { - t.Error("expected mic audio from new source after hot-swap") + t.Error("expected mic audio from new source after discovery") } } -func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) { +func TestParec_MicCaptureFailureDoesNotAffectOthers(t *testing.T) { withFastPolling(t) client := newFakeSinkClient() client.runner.setStream("mic1", bytes.Repeat([]byte{0x07, 0x00}, pcm.FrameBytes/2)) - client.setSource("mic1") + client.setSources(parec.Source{Name: "mic1"}) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} _, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -375,7 +377,8 @@ func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) { waitFor(t, time.Second, func() bool { src.mu.Lock() defer src.mu.Unlock() - return src.micName == "mic1" + _, ok := src.mics["mic1"] + return ok }) client.mu.Lock() @@ -386,24 +389,19 @@ func TestParec_MicHotSwapFailureRetainsOld(t *testing.T) { return nil } client.mu.Unlock() - client.setSource("mic2") + client.setSources(parec.Source{Name: "mic1"}, parec.Source{Name: "mic2"}) time.Sleep(100 * time.Millisecond) src.mu.Lock() - name := src.micName - mic := src.mic + _, hasMic1 := src.mics["mic1"] + _, hasMic2 := src.mics["mic2"] src.mu.Unlock() - if name != "mic1" { - t.Errorf("expected mic to remain mic1 after failed swap, got %q", name) - } - if mic == nil { - t.Fatal("expected mic reader to still be present") + if !hasMic1 { + t.Error("expected mic1 to still be active") } - select { - case <-mic.dead(): - t.Error("old mic reader should still be alive after a failed swap") - default: + if hasMic2 { + t.Error("expected mic2 to not be active after failed start") } } @@ -417,7 +415,7 @@ func TestParec_ReaderExitTriggersRestartWithoutKillingOthers(t *testing.T) { parec.Sink{Name: "b", MonitorSource: "b.monitor"}, ) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} _, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -465,9 +463,9 @@ func TestParec_PulseAudioUnreachableThenRecovers(t *testing.T) { withFastPolling(t) client := newFakeSinkClient() client.setSinksErr(io.ErrClosedPipe) - client.setSourceErr(io.ErrClosedPipe) + client.setSourcesErr(io.ErrClosedPipe) - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} frames, err := src.Start(context.Background()) if err != nil { t.Fatal(err) @@ -484,15 +482,16 @@ func TestParec_PulseAudioUnreachableThenRecovers(t *testing.T) { client.runner.setStream("a.monitor", bytes.Repeat([]byte{0x09, 0x00}, pcm.FrameBytes/2)) client.setSinksErr(nil) client.setSinks(parec.Sink{Name: "a", MonitorSource: "a.monitor"}) - client.setSourceErr(nil) - client.setSource("mic1") + client.setSourcesErr(nil) + client.setSources(parec.Source{Name: "mic1"}) client.runner.setStream("mic1", bytes.Repeat([]byte{0x0a, 0x00}, pcm.FrameBytes/2)) waitFor(t, 2*time.Second, func() bool { src.mu.Lock() defer src.mu.Unlock() _, hasSink := src.sinks["a"] - return hasSink && src.micName == "mic1" + _, hasMic := src.mics["mic1"] + return hasSink && hasMic }) } @@ -508,7 +507,7 @@ func isSilent(data []byte) bool { func TestParec_Stop_Idempotent(t *testing.T) { withFastPolling(t) client := newFakeSinkClient() - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} if _, err := src.Start(context.Background()); err != nil { t.Fatal(err) } @@ -523,7 +522,7 @@ func TestParec_Stop_Idempotent(t *testing.T) { func TestParec_Backpressure_SlowConsumerDoesNotDropFrames(t *testing.T) { withFastPolling(t) client := newFakeSinkClient() - src := &Parec{client: client, sinks: make(map[string]*reader)} + src := &Parec{client: client, sinks: make(map[string]*reader), mics: make(map[string]*reader)} frames, err := src.Start(context.Background()) if err != nil { t.Fatal(err) diff --git a/internal/capture/reconcile.go b/internal/capture/reconcile.go index f337adf..f68e392 100644 --- a/internal/capture/reconcile.go +++ b/internal/capture/reconcile.go @@ -51,7 +51,7 @@ func (c *Parec) reconcileLoop(ctx context.Context) { defer micTicker.Stop() c.reconcileSinks(ctx) - c.reconcileMic(ctx) + c.reconcileMics(ctx) for { select { @@ -60,7 +60,7 @@ func (c *Parec) reconcileLoop(ctx context.Context) { case <-sinkTicker.C: c.reconcileSinks(ctx) case <-micTicker.C: - c.reconcileMic(ctx) + c.reconcileMics(ctx) } } } @@ -148,18 +148,17 @@ func (c *Parec) stopSink(name string, r *reader, reason string) { ) } -// reconcileMic resolves the default microphone source and swaps to it only -// after the replacement capture starts successfully, so a failed swap never -// drops below one working mic reader. -func (c *Parec) reconcileMic(ctx context.Context) { +// reconcileMics lists all input sources and diffs them against active mic +// readers by name. Same pattern as reconcileSinks. +func (c *Parec) reconcileMics(ctx context.Context) { if time.Since(c.micLastTry) < c.micBackoff.cur { return } c.micLastTry = time.Now() - resp, err := c.client.GetDefaultSource(ctx, parec.GetDefaultSourceRequest{}) + resp, err := c.client.ListSources(ctx, parec.ListSourcesRequest{}) if err != nil { - slog.WarnContext(ctx, "get default source failed", + slog.WarnContext(ctx, "list sources failed", "err", err, "retryIn", c.micBackoff.next(), ) @@ -167,52 +166,62 @@ func (c *Parec) reconcileMic(ctx context.Context) { } c.micBackoff.reset() + wanted := make(map[string]parec.Source, len(resp.Sources)) + for _, s := range resp.Sources { + wanted[s.Name] = s + } + c.mu.Lock() - name := resp.Source - current := c.mic - currentName := c.micName + current := make(map[string]*reader, len(c.mics)) + maps.Copy(current, c.mics) c.mu.Unlock() - dead := false - if current != nil { + for name, r := range current { + if _, ok := wanted[name]; !ok { + c.stopMic(name, r, "source disappeared") + continue + } select { - case <-current.dead(): - dead = true + case <-r.dead(): + c.stopMic(name, r, "mic reader exited") default: } } - if current != nil && !dead && name == currentName { - return - } + for name := range wanted { + c.mu.Lock() + _, active := c.mics[name] + c.mu.Unlock() + if active { + continue + } - stream, err := c.client.StartCapture(ctx, parec.StartCaptureRequest{ - Device: name, SampleRate: pcm.SampleRate, - }) - if err != nil { - slog.WarnContext(ctx, "mic capture start failed", - "source", name, - "err", err, - ) - return + stream, err := c.client.StartCapture(ctx, parec.StartCaptureRequest{ + Device: name, SampleRate: pcm.SampleRate, + }) + if err != nil { + slog.WarnContext(ctx, "mic capture start failed", + "source", name, + "err", err, + ) + continue + } + + r := startReader(stream) + c.mu.Lock() + c.mics[name] = r + c.mu.Unlock() + slog.InfoContext(ctx, "mic capture started", "source", name) } +} - r := startReader(stream) +func (c *Parec) stopMic(name string, r *reader, reason string) { c.mu.Lock() - c.mic = r - c.micName = name + delete(c.mics, name) c.mu.Unlock() - - if current != nil { - _ = current.stop() - } - - if currentName == "" { - slog.InfoContext(ctx, "mic capture started", "source", name) - } else { - slog.InfoContext(ctx, "default microphone changed", - "old", currentName, - "new", name, - ) - } + _ = r.stop() + slog.InfoContext(context.Background(), "mic capture stopped", + "source", name, + "reason", reason, + ) } diff --git a/internal/protocol/parec/client.go b/internal/protocol/parec/client.go index 5db0338..68737c8 100644 --- a/internal/protocol/parec/client.go +++ b/internal/protocol/parec/client.go @@ -91,6 +91,43 @@ func (c *Client) GetDefaultSource(ctx context.Context, _ GetDefaultSourceRequest }, nil } +// ListSourcesRequest is empty; sources are a system-global query. +type ListSourcesRequest struct{} + +// Source describes one PulseAudio input source (microphone). +type Source struct { + // Name is the source's PulseAudio name. + Name string +} + +// ListSourcesResponse contains all currently known input sources (non-monitor). +type ListSourcesResponse struct { + Sources []Source +} + +// ListSources enumerates all PulseAudio input sources, excluding sink monitors. +func (c *Client) ListSources(ctx context.Context, _ ListSourcesRequest) (ListSourcesResponse, error) { + out, err := c.runner.Output(ctx, "pactl", "--format=json", "list", "sources") + if err != nil { + return ListSourcesResponse{}, fmt.Errorf("pactl list sources: %w", err) + } + var wire []struct { + Name string `json:"name"` + MonitorSource string `json:"monitor_source"` + } + if err := json.Unmarshal(out, &wire); err != nil { + return ListSourcesResponse{}, fmt.Errorf("parse pactl sources json: %w", err) + } + sources := make([]Source, 0, len(wire)) + for _, s := range wire { + if s.MonitorSource != "" { + continue + } + sources = append(sources, Source{Name: s.Name}) + } + return ListSourcesResponse{Sources: sources}, nil +} + // ListSinksRequest is empty; sinks are a system-global query. type ListSinksRequest struct{} diff --git a/internal/protocol/parec/client_test.go b/internal/protocol/parec/client_test.go index 5510642..b25b335 100644 --- a/internal/protocol/parec/client_test.go +++ b/internal/protocol/parec/client_test.go @@ -156,6 +156,62 @@ func TestListSinks_CommandError(t *testing.T) { } } +func TestListSources(t *testing.T) { + runner := &fakeRunner{ + outputFn: func(ctx context.Context, name string, args ...string) ([]byte, error) { + expectedArgs := []string{"--format=json", "list", "sources"} + if len(args) != len(expectedArgs) { + t.Fatalf("expected %d args, got %d: %v", len(expectedArgs), len(args), args) + } + for i, exp := range expectedArgs { + if args[i] != exp { + t.Errorf("arg %d: expected %q, got %q", i, exp, args[i]) + } + } + return []byte(`[ + {"name": "alsa_output.hdmi.monitor", "monitor_source": "alsa_output.hdmi"}, + {"name": "alsa_input.usb-mic", "monitor_source": ""}, + {"name": "bluez_input.headset", "monitor_source": ""} + ]`), nil + }, + } + + client := parec.New(runner) + resp, err := client.ListSources(context.Background(), parec.ListSourcesRequest{}) + if err != nil { + t.Fatal(err) + } + want := []parec.Source{ + {Name: "alsa_input.usb-mic"}, + {Name: "bluez_input.headset"}, + } + if len(resp.Sources) != len(want) { + t.Fatalf("expected %d sources, got %d: %v", len(want), len(resp.Sources), resp.Sources) + } + for i, w := range want { + if resp.Sources[i] != w { + t.Errorf("source %d: expected %+v, got %+v", i, w, resp.Sources[i]) + } + } +} + +func TestListSources_CommandError(t *testing.T) { + runner := &fakeRunner{ + outputFn: func(ctx context.Context, name string, args ...string) ([]byte, error) { + return nil, errors.New("command not found") + }, + } + + client := parec.New(runner) + _, err := client.ListSources(context.Background(), parec.ListSourcesRequest{}) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "pactl list sources") { + t.Errorf("expected wrapped error, got: %v", err) + } +} + func TestStartCapture(t *testing.T) { pcmData := bytes.Repeat([]byte{0x01, 0x02}, 100) closed := false