Add wyoming module

Author: AlexxIT

internal/streams/play.go

@@ -7,7 +7,7 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/core"
 )
 
-func (s *Stream) Play(source string) error {
+func (s *Stream) Play(urlOrProd any) error {
 	s.mu.Lock()
 	for _, producer := range s.producers {
 		if producer.state == stateInternal && producer.conn != nil {
@@ -16,12 +16,18 @@ func (s *Stream) Play(source string) error {
 	}
 	s.mu.Unlock()
 
-	if source == "" {
-		return nil
-	}
-
+	var source string
 	var src core.Producer
 
+	switch urlOrProd.(type) {
+	case string:
+		if source = urlOrProd.(string); source == "" {
+			return nil
+		}
+	case core.Producer:
+		src = urlOrProd.(core.Producer)
+	}
+
 	for _, producer := range s.producers {
 		if producer.conn == nil {
 			continue

internal/wyoming/wyoming.go

@@ -0,0 +1,70 @@
+package wyoming
+
+import (
+	"net"
+
+	"github.com/AlexxIT/go2rtc/internal/app"
+	"github.com/AlexxIT/go2rtc/internal/streams"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/wyoming"
+	"github.com/rs/zerolog"
+)
+
+func Init() {
+	streams.HandleFunc("wyoming", wyoming.Dial)
+
+	// server
+	var cfg struct {
+		Mod map[string]struct {
+			Listen       string  `yaml:"listen"`
+			Name         string  `yaml:"name"`
+			WakeURI      string  `yaml:"wake_uri"`
+			VADThreshold float32 `yaml:"vad_threshold"`
+		} `yaml:"wyoming"`
+	}
+	app.LoadConfig(&cfg)
+
+	log = app.GetLogger("wyoming")
+
+	for name, cfg := range cfg.Mod {
+		stream := streams.Get(name)
+		if stream == nil {
+			log.Warn().Msgf("[wyoming] missing stream: %s", name)
+			continue
+		}
+
+		ln, err := net.Listen("tcp", cfg.Listen)
+		if err != nil {
+			log.Warn().Msgf("[wyoming] listen error: %s", err)
+			continue
+		}
+
+		if cfg.Name == "" {
+			cfg.Name = name
+		}
+
+		srv := wyoming.Server{
+			Name:         cfg.Name,
+			VADThreshold: int16(1000 * cfg.VADThreshold), // 1.0 => 1000
+			WakeURI:      cfg.WakeURI,
+			MicHandler: func(cons core.Consumer) error {
+				if err := stream.AddConsumer(cons); err != nil {
+					return err
+				}
+				// not best solution
+				if i, ok := cons.(interface{ OnClose(func()) }); ok {
+					i.OnClose(func() {
+						stream.RemoveConsumer(cons)
+					})
+				}
+				return nil
+			},
+			SndHandler: func(prod core.Producer) error {
+				return stream.Play(prod)
+			},
+		}
+		go srv.Serve(ln)
+	}
+}
+
+var log zerolog.Logger

main.go

@@ -38,6 +38,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/v4l2"
 	"github.com/AlexxIT/go2rtc/internal/webrtc"
 	"github.com/AlexxIT/go2rtc/internal/webtorrent"
+	"github.com/AlexxIT/go2rtc/internal/wyoming"
 	"github.com/AlexxIT/go2rtc/pkg/shell"
 )
 
@@ -69,6 +70,7 @@ func main() {
 	hass.Init()       // hass source, Hass API server
 	onvif.Init()      // onvif source, ONVIF API server
 	webtorrent.Init() // webtorrent source, WebTorrent module
+	wyoming.Init()
 
 	// 5. Other sources
 

pkg/core/codec.go

@@ -277,7 +277,7 @@ func ParseCodecString(s string) *Codec {
 		codec.ClockRate = uint32(Atoi(ss[1]))
 	}
 	if len(ss) >= 3 {
-		codec.Channels = uint16(Atoi(ss[1]))
+		codec.Channels = uint8(Atoi(ss[1]))
 	}
 
 	return &codec

pkg/pcm/pcm.go

@@ -185,3 +185,23 @@ func Transcode(dst, src *core.Codec) func([]byte) []byte {
 		return writer(samples)
 	}
 }
+
+func ConsumerCodecs() []*core.Codec {
+	return []*core.Codec{
+		{Name: core.CodecPCML},
+		{Name: core.CodecPCM},
+		{Name: core.CodecPCMA},
+		{Name: core.CodecPCMU},
+	}
+}
+
+func ProducerCodecs() []*core.Codec {
+	return []*core.Codec{
+		{Name: core.CodecPCML, ClockRate: 16000},
+		{Name: core.CodecPCM, ClockRate: 16000},
+		{Name: core.CodecPCML, ClockRate: 8000},
+		{Name: core.CodecPCM, ClockRate: 8000},
+		{Name: core.CodecPCMA, ClockRate: 8000},
+		{Name: core.CodecPCMU, ClockRate: 8000},
+	}
+}

pkg/pcm/s16le/s16le.go

@@ -0,0 +1,42 @@
+package s16le
+
+func PeaksRMS(b []byte) int16 {
+	// RMS of sine wave = peak / sqrt2
+	// https://en.wikipedia.org/wiki/Root_mean_square
+	// https://www.youtube.com/watch?v=MUDkL4KZi0I
+	var peaks int32
+	var peaksSum int32
+	var prevSample int16
+	var prevUp bool
+
+	var i int
+	for n := len(b); i < n; {
+		lo := b[i]
+		i++
+		hi := b[i]
+		i++
+
+		sample := int16(hi)<<8 | int16(lo)
+		up := sample >= prevSample
+
+		if i >= 4 {
+			if up != prevUp {
+				if prevSample >= 0 {
+					peaksSum += int32(prevSample)
+				} else {
+					peaksSum -= int32(prevSample)
+				}
+				peaks++
+			}
+		}
+
+		prevSample = sample
+		prevUp = up
+	}
+
+	if peaks == 0 {
+		return 0
+	}
+
+	return int16(peaksSum / peaks)
+}

pkg/wyoming/api.go

@@ -0,0 +1,98 @@
+package wyoming
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"net"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+)
+
+type API struct {
+	conn net.Conn
+	rd   *bufio.Reader
+}
+
+func DialAPI(address string) (*API, error) {
+	conn, err := net.DialTimeout("tcp", address, core.ConnDialTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewAPI(conn), nil
+}
+
+const Version = "1.5.4"
+
+func NewAPI(conn net.Conn) *API {
+	return &API{conn: conn, rd: bufio.NewReader(conn)}
+}
+
+func (w *API) WriteEvent(evt *Event) (err error) {
+	hdr := EventHeader{
+		Type:          evt.Type,
+		Version:       Version,
+		DataLength:    len(evt.Data),
+		PayloadLength: len(evt.Payload),
+	}
+
+	buf, err := json.Marshal(hdr)
+	if err != nil {
+		return err
+	}
+
+	buf = append(buf, '\n')
+	buf = append(buf, evt.Data...)
+	buf = append(buf, evt.Payload...)
+
+	_, err = w.conn.Write(buf)
+	return err
+}
+
+func (w *API) ReadEvent() (*Event, error) {
+	data, err := w.rd.ReadBytes('\n')
+	if err != nil {
+		return nil, err
+	}
+
+	var hdr EventHeader
+	if err = json.Unmarshal(data, &hdr); err != nil {
+		return nil, err
+	}
+
+	evt := Event{Type: hdr.Type}
+
+	if hdr.DataLength > 0 {
+		evt.Data = make([]byte, hdr.DataLength)
+		if _, err = io.ReadFull(w.rd, evt.Data); err != nil {
+			return nil, err
+		}
+	}
+
+	if hdr.PayloadLength > 0 {
+		evt.Payload = make([]byte, hdr.PayloadLength)
+		if _, err = io.ReadFull(w.rd, evt.Payload); err != nil {
+			return nil, err
+		}
+	}
+
+	return &evt, nil
+}
+
+func (w *API) Close() error {
+	return w.conn.Close()
+}
+
+type Event struct {
+	Type    string
+	Data    []byte
+	Payload []byte
+}
+
+type EventHeader struct {
+	Type          string `json:"type"`
+	Version       string `json:"version"`
+	DataLength    int    `json:"data_length,omitempty"`
+	PayloadLength int    `json:"payload_length,omitempty"`
+}

pkg/wyoming/backchannel.go

@@ -0,0 +1,42 @@
+package wyoming
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/pion/rtp"
+)
+
+type Backchannel struct {
+	core.Connection
+	api *API
+}
+
+func (b *Backchannel) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) {
+	return nil, core.ErrCantGetTrack
+}
+
+func (b *Backchannel) AddTrack(media *core.Media, codec *core.Codec, track *core.Receiver) error {
+	sender := core.NewSender(media, codec)
+	sender.Handler = func(pkt *rtp.Packet) {
+		ts := time.Now().Nanosecond()
+		evt := &Event{
+			Type:    "audio-chunk",
+			Data:    []byte(fmt.Sprintf(`{"rate":16000,"width":2,"channels":1,"timestamp":%d}`, ts)),
+			Payload: pkt.Payload,
+		}
+		_ = b.api.WriteEvent(evt)
+	}
+	sender.HandleRTP(track)
+	b.Senders = append(b.Senders, sender)
+	return nil
+}
+
+func (b *Backchannel) Start() error {
+	for {
+		if _, err := b.api.ReadEvent(); err != nil {
+			return err
+		}
+	}
+}

pkg/wyoming/producer.go

@@ -0,0 +1,43 @@
+package wyoming
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/pion/rtp"
+)
+
+type Producer struct {
+	core.Connection
+	api *API
+}
+
+func (p *Producer) Start() error {
+	var seq uint16
+	var ts uint32
+
+	for {
+		evt, err := p.api.ReadEvent()
+		if err != nil {
+			return err
+		}
+
+		if evt.Type != "audio-chunk" {
+			continue
+		}
+
+		p.Recv += len(evt.Payload)
+
+		pkt := &core.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				SequenceNumber: seq,
+				Timestamp:      ts,
+			},
+			Payload: evt.Payload,
+		}
+		p.Receivers[0].WriteRTP(pkt)
+
+		seq++
+		ts += uint32(len(evt.Payload) / 2)
+	}
+}