Merge #16 branch 'diamondburned-pa-fix'

go.mod

@@ -4,7 +4,7 @@ go 1.18
 
 require (
 	github.com/integrii/flaggy v1.4.4
-	github.com/lawl/pulseaudio v0.0.0-20210928141934-ed754c0c6618
+	github.com/noisetorch/pulseaudio v0.0.0-20220603053345-9303200c3861
 	github.com/nsf/termbox-go v1.1.1
 	github.com/pkg/errors v0.9.1
 	gonum.org/v1/gonum v0.11.0

go.sum

@@ -1,9 +1,9 @@
 github.com/integrii/flaggy v1.4.4 h1:8fGyiC14o0kxhTqm2VBoN19fDKPZsKipP7yggreTMDc=
 github.com/integrii/flaggy v1.4.4/go.mod h1:tnTxHeTJbah0gQ6/K0RW0J7fMUBk9MCF5blhm43LNpI=
-github.com/lawl/pulseaudio v0.0.0-20210928141934-ed754c0c6618 h1:lktbhQBHluc1oWEDow4DEv13qkWJ8zm/dTUSKer2iKk=
-github.com/lawl/pulseaudio v0.0.0-20210928141934-ed754c0c6618/go.mod h1:9h36x4KH7r2V8DOCKoPMt87IXZ++X90y8D5nnuwq290=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/noisetorch/pulseaudio v0.0.0-20220603053345-9303200c3861 h1:Xng5X+MlNK7Y/Ede75B86wJgaFMFvuey1K4Suh9k2E4=
+github.com/noisetorch/pulseaudio v0.0.0-20220603053345-9303200c3861/go.mod h1:/zosM8PSkhuVyfJ9c/qzBhPSm3k06m9U4y4SDfH0jeA=
 github.com/nsf/termbox-go v1.1.1 h1:nksUPLCb73Q++DwbYUBEglYBRPZyoXJdrj5L+TkjyZY=
 github.com/nsf/termbox-go v1.1.1/go.mod h1:T0cTdVuOwf7pHQNtfhnEbzHbcNyCEcVU4YPpouCbVxo=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

input/all/all.go

@@ -4,4 +4,5 @@ package all
 import (
 	_ "github.com/noriah/catnip/input/ffmpeg"
 	_ "github.com/noriah/catnip/input/parec"
+	_ "github.com/noriah/catnip/input/pipewire"
 )

input/common/execread/execread.go

@@ -9,13 +9,17 @@ import (
 	"os"
 	"os/exec"
 	"sync"
+	"time"
 
 	"github.com/noriah/catnip/input"
 	"github.com/pkg/errors"
 )
 
 // Session is a session that reads floating-point audio values from a Cmd.
 type Session struct {
+	// OnStart is called when the session starts. Nil by default.
+	OnStart func(ctx context.Context, cmd *exec.Cmd) error
+
 	argv []string
 	cfg  input.SessionConfig
 
@@ -26,38 +30,47 @@ type Session struct {
 }
 
 // NewSession creates a new execread session. It never returns an error.
-func NewSession(argv []string, f32mode bool, cfg input.SessionConfig) (*Session, error) {
+func NewSession(argv []string, f32mode bool, cfg input.SessionConfig) *Session {
 	if len(argv) < 1 {
-		return nil, errors.New("argv has no arg0")
+		panic("argv has no arg0")
 	}
 
 	return &Session{
 		argv:    argv,
 		cfg:     cfg,
 		f32mode: f32mode,
 		samples: cfg.SampleSize * cfg.FrameSize,
-	}, nil
+	}
 }
 
 func (s *Session) Start(ctx context.Context, dst [][]input.Sample, kickChan chan bool, mu *sync.Mutex) error {
 	if !input.EnsureBufferLen(s.cfg, dst) {
 		return errors.New("invalid dst length given")
 	}
 
-	// Take argv and free it soon after, since we won't be needing it again.
 	cmd := exec.CommandContext(ctx, s.argv[0], s.argv[1:]...)
 	cmd.Stderr = os.Stderr
-	s.argv = nil
 
 	o, err := cmd.StdoutPipe()
 	if err != nil {
 		return errors.Wrap(err, "failed to get stdout pipe")
 	}
 	defer o.Close()
 
-	bufsz := s.samples * 4
-	if !s.f32mode {
-		bufsz *= 2
+	// We need o as an *os.File for SetWriteDeadline.
+	of, ok := o.(*os.File)
+	if !ok {
+		return errors.New("stdout pipe is not an *os.File (bug)")
+	}
+
+	if err := cmd.Start(); err != nil {
+		return errors.Wrap(err, "failed to start "+s.argv[0])
+	}
+
+	if s.OnStart != nil {
+		if err := s.OnStart(ctx, cmd); err != nil {
+			return err
+		}
 	}
 
 	framesz := s.cfg.FrameSize
@@ -66,60 +79,94 @@ func (s *Session) Start(ctx context.Context, dst [][]input.Sample, kickChan chan
 		f64:   !s.f32mode,
 	}
 
-	// Allocate 4 times the buffer. We should ensure that we can read some of
-	// the overflow.
-	raw := make([]byte, bufsz)
-
-	if err := cmd.Start(); err != nil {
-		return errors.Wrap(err, "failed to start ffmpeg")
+	bufsz := s.samples
+	if !s.f32mode {
+		bufsz *= 2
 	}
 
-	for {
-		reader.reset(raw)
+	raw := make([]byte, bufsz*4)
 
-		mu.Lock()
-		for n := 0; n < s.samples; n++ {
-			dst[n%framesz][n/framesz] = reader.next()
-		}
-		mu.Unlock()
+	// We double this as a workaround because sampleDuration is less than the
+	// actual time that ReadFull blocks for some reason, probably because the
+	// process decides to discard audio when it overflows.
+	sampleDuration := time.Duration(
+		float64(s.cfg.SampleSize) / s.cfg.SampleRate * float64(time.Second))
+	// We also keep track of whether the deadline was hit once so we can half
+	// the sample duration. This smooths out the jitter.
+	var readExpired bool
 
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-			// default:
-		case kickChan <- true:
+	for {
+		// Set us a read deadline. If the deadline is reached, we'll write zeros
+		// to the buffer.
+		timeout := sampleDuration
+		if !readExpired {
+			timeout *= 2
+		}
+		if err := of.SetReadDeadline(time.Now().Add(timeout)); err != nil {
+			return errors.Wrap(err, "failed to set read deadline")
 		}
 
 		_, err := io.ReadFull(o, raw)
 		if err != nil {
-			if errors.Is(err, io.EOF) {
+			switch {
+			case errors.Is(err, io.EOF):
 				return nil
+			case errors.Is(err, os.ErrDeadlineExceeded):
+				readExpired = true
+			default:
+				return err
 			}
-			return err
+		} else {
+			readExpired = false
+		}
+
+		if readExpired {
+			mu.Lock()
+			// We can write directly to dst just so we can avoid parsing zero
+			// bytes to floats.
+			for _, buf := range dst {
+				// Go should optimize this to a memclr.
+				for i := range buf {
+					buf[i] = 0
+				}
+			}
+			mu.Unlock()
+		} else {
+			reader.reset(raw)
+			mu.Lock()
+			for n := 0; n < s.samples; n++ {
+				dst[n%framesz][n/framesz] = reader.next()
+			}
+			mu.Unlock()
+		}
+
+		// Signal that we've written to dst.
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case kickChan <- true:
 		}
 	}
 }
 
 type floatReader struct {
 	order binary.ByteOrder
 	buf   []byte
-	n     int64
 	f64   bool
 }
 
 func (f *floatReader) reset(b []byte) {
-	f.n = 0
 	f.buf = b
 }
 
 func (f *floatReader) next() float64 {
-	n := f.n
-
 	if f.f64 {
-		f.n += 8
-		return math.Float64frombits(f.order.Uint64(f.buf[n:]))
+		b := f.buf[:8]
+		f.buf = f.buf[8:]
+		return math.Float64frombits(f.order.Uint64(b))
 	}
 
-	f.n += 4
-	return float64(math.Float32frombits(f.order.Uint32(f.buf[n:])))
+	b := f.buf[:4]
+	f.buf = f.buf[4:]
+	return float64(math.Float32frombits(f.order.Uint32(b)))
 }

input/ffmpeg/ffmpeg.go

@@ -21,5 +21,5 @@ func NewSession(b FFmpegBackend, cfg input.SessionConfig) (*execread.Session, er
 		"-",
 	)
 
-	return execread.NewSession(args, false, cfg)
+	return execread.NewSession(args, false, cfg), nil
 }

input/parec/parec.go

@@ -3,7 +3,7 @@ package parec
 import (
 	"fmt"
 
-	"github.com/lawl/pulseaudio"
+	"github.com/noisetorch/pulseaudio"
 	"github.com/noriah/catnip/input"
 	"github.com/noriah/catnip/input/common/execread"
 	"github.com/pkg/errors"
@@ -83,5 +83,5 @@ func NewSession(cfg input.SessionConfig) (*execread.Session, error) {
 		args = append(args, "-d", dv.String())
 	}
 
-	return execread.NewSession(args, true, cfg)
+	return execread.NewSession(args, true, cfg), nil
 }

input/pipewire/dump.go

@@ -0,0 +1,140 @@
+package pipewire
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"os/exec"
+
+	"github.com/pkg/errors"
+)
+
+type pwObjects []pwObject
+
+func pwDump(ctx context.Context) (pwObjects, error) {
+	cmd := exec.CommandContext(ctx, "pw-dump")
+	cmd.Stderr = os.Stderr
+
+	dumpOutput, err := cmd.Output()
+	if err != nil {
+		var execErr *exec.ExitError
+		if errors.As(err, &execErr) {
+			return nil, errors.Wrapf(err, "failed to run pw-dump: %s", execErr.Stderr)
+		}
+		return nil, errors.Wrap(err, "failed to run pw-dump")
+	}
+
+	var dump pwObjects
+	if err := json.Unmarshal(dumpOutput, &dump); err != nil {
+		return nil, errors.Wrap(err, "failed to parse pw-dump output")
+	}
+
+	return dump, nil
+}
+
+// Filter filters for the devices that satisfies f.
+func (d pwObjects) Filter(fns ...func(pwObject) bool) pwObjects {
+	filtered := make(pwObjects, 0, len(d))
+loop:
+	for _, device := range d {
+		for _, f := range fns {
+			if !f(device) {
+				continue loop
+			}
+		}
+		filtered = append(filtered, device)
+	}
+	return filtered
+}
+
+// Find returns the first object that satisfies f.
+func (d pwObjects) Find(f func(pwObject) bool) *pwObject {
+	for i, device := range d {
+		if f(device) {
+			return &d[i]
+		}
+	}
+	return nil
+}
+
+// ResolvePorts returns all PipeWire port objects that belong to the given
+// object.
+func (d pwObjects) ResolvePorts(object *pwObject, dir pwPortDirection) pwObjects {
+	return d.Filter(
+		func(o pwObject) bool { return o.Type == pwInterfacePort },
+		func(o pwObject) bool {
+			return o.Info.Props.NodeID == object.ID && o.Info.Props.PortDirection == dir
+		},
+	)
+}
+
+type pwObjectID int64
+
+type pwObjectType string
+
+const (
+	pwInterfaceDevice pwObjectType = "PipeWire:Interface:Device"
+	pwInterfaceNode   pwObjectType = "PipeWire:Interface:Node"
+	pwInterfacePort   pwObjectType = "PipeWire:Interface:Port"
+	pwInterfaceLink   pwObjectType = "PipeWire:Interface:Link"
+)
+
+type pwObject struct {
+	ID   pwObjectID   `json:"id"`
+	Type pwObjectType `json:"type"`
+	Info struct {
+		Props pwInfoProps `json:"props"`
+	} `json:"info"`
+}
+
+type pwInfoProps struct {
+	pwDeviceProps
+	pwNodeProps
+	pwPortProps
+	MediaClass string `json:"media.class"`
+
+	JSON json.RawMessage `json:"-"`
+}
+
+func (p *pwInfoProps) UnmarshalJSON(data []byte) error {
+	type Alias pwInfoProps
+	if err := json.Unmarshal(data, (*Alias)(p)); err != nil {
+		return err
+	}
+	p.JSON = append([]byte(nil), data...)
+	return nil
+}
+
+type pwDeviceProps struct {
+	DeviceName string `json:"device.name"`
+}
+
+// pwNodeProps is for Audio/Sink only.
+type pwNodeProps struct {
+	NodeName        string `json:"node.name"`
+	NodeNick        string `json:"node.nick"`
+	NodeDescription string `json:"node.description"`
+}
+
+// Constants for MediaClass.
+const (
+	pwAudioDevice       string = "Audio/Device"
+	pwAudioSink         string = "Audio/Sink"
+	pwStreamOutputAudio string = "Stream/Output/Audio"
+)
+
+type pwPortDirection string
+
+const (
+	pwPortIn  = "in"
+	pwPortOut = "out"
+)
+
+type pwPortProps struct {
+	PortID        pwObjectID      `json:"port.id"`
+	PortName      string          `json:"port.name"`
+	PortAlias     string          `json:"port.alias"`
+	PortDirection pwPortDirection `json:"port.direction"`
+	NodeID        pwObjectID      `json:"node.id"`
+	ObjectPath    string          `json:"object.path"`
+}