feat: add PipeWire screen and system audio capture (M2)
Capture the Wayland desktop via xdg-desktop-portal + PipeWire using go2tv.app/screencast (MIT), and stream it to OBS: - internal/capture: Capture/FrameSource/AudioSource interfaces and the PipeWire backend (BGRA frames at monitor resolution, S16 48 kHz stereo system audio) - protocol: EncodeBGRA fast path producing 4:2:0 YCbCr JPEGs - cmd: --source screen|pattern, --audio, --stream-index flags; real capture feeds the existing sender - share one wall-clock reference between the audio and video loops so OBS receives aligned A/V timestamps (avoids multi-second latency) Verified end-to-end: real desktop at 30 fps renders in OBS with sub-second latency.
This commit is contained in:
+219
-101
@@ -1,22 +1,27 @@
|
||||
// Command teleportfling is a standalone sender for the Teleport protocol.
|
||||
//
|
||||
// M1 milestone: protocol proof-of-life. It streams a synthetic test pattern
|
||||
// (colour bars with a moving box) plus a silent stereo 48 kHz audio tone over
|
||||
// TCP and announces itself on the LAN multicast group, so an OBS instance
|
||||
// It captures a Wayland screen (PipeWire via xdg-desktop-portal) plus the
|
||||
// system's default audio output and streams them over TCP as the Teleport
|
||||
// protocol, announcing itself on the LAN multicast group so an OBS instance
|
||||
// with the obs-teleport plugin can discover and decode the stream.
|
||||
//
|
||||
// Usage:
|
||||
//
|
||||
// teleportfling [--name NAME] [--port PORT] [--width W] [--height H]
|
||||
// [--fps N] [--quality 1..100] [--duration SECONDS]
|
||||
// teleportfling [--name NAME] [--port PORT] [--quality 1..100]
|
||||
// [--fps N] [--source screen|pattern] [--audio]
|
||||
// [--stream-index N] [--duration SECONDS]
|
||||
//
|
||||
// M2+ replaces the synthetic sources with real PipeWire screen/audio capture.
|
||||
// --source pattern selects the M1 synthetic test pattern (colour bars with a
|
||||
// moving box) instead of real screen capture, which is useful for testing
|
||||
// without granting screen-share permission.
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"flag"
|
||||
"image"
|
||||
"image/color"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
@@ -25,20 +30,30 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"teleportfling/internal/capture"
|
||||
"teleportfling/internal/discovery"
|
||||
"teleportfling/internal/output"
|
||||
"teleportfling/internal/protocol"
|
||||
)
|
||||
|
||||
const (
|
||||
// sampleRate and speakers describe the captured/encoded audio stream.
|
||||
sampleRate = 48000
|
||||
speakers = 2
|
||||
// audioChunk sets how much audio we packetize per WAVE message (~10 ms).
|
||||
audioChunk = 10 * time.Millisecond
|
||||
)
|
||||
|
||||
func main() {
|
||||
var (
|
||||
name = flag.String("name", "", "announce name (default: hostname)")
|
||||
port = flag.Int("port", 9756, "TCP listening port")
|
||||
width = flag.Int("width", 1920, "frame width")
|
||||
height = flag.Int("height", 1080, "frame height")
|
||||
fps = flag.Int("fps", 30, "video frames per second")
|
||||
quality = flag.Int("quality", 80, "JPEG quality 1..100")
|
||||
duration = flag.Duration("duration", 0, "stream duration (0 = run until interrupted)")
|
||||
name = flag.String("name", "", "announce name (default: hostname)")
|
||||
port = flag.Int("port", 9756, "TCP listening port")
|
||||
quality = flag.Int("quality", 80, "JPEG quality 1..100")
|
||||
fps = flag.Int("fps", 30, "video frames per second")
|
||||
source = flag.String("source", "screen", "capture source: screen or pattern")
|
||||
withAudio = flag.Bool("audio", true, "capture and stream system audio")
|
||||
streamIndex = flag.Int("stream-index", 0, "monitor index to capture (screen source)")
|
||||
duration = flag.Duration("duration", 0, "stream duration (0 = run until interrupted)")
|
||||
)
|
||||
flag.Parse()
|
||||
|
||||
@@ -49,114 +64,68 @@ func main() {
|
||||
log.Fatalf("output: listen: %v", err)
|
||||
}
|
||||
announcer := discovery.Start(*name, p)
|
||||
log.Printf("teleportfling: advertising on %d, capturing %dx%d @ %d fps", p, *width, *height, *fps)
|
||||
|
||||
// Pipeline state.
|
||||
var (
|
||||
totalFrames atomic.Int64
|
||||
encoder = mustNewEncoder()
|
||||
frameInterval = time.Second / time.Duration(*fps)
|
||||
audioInterval = 100 * time.Millisecond
|
||||
sampleRate = 48000
|
||||
speakers = 2
|
||||
start = time.Now()
|
||||
audioStart time.Time
|
||||
deadline time.Time
|
||||
stop = make(chan struct{})
|
||||
totalFrames atomic.Int64
|
||||
encoder = mustNewEncoder()
|
||||
start = time.Now()
|
||||
stop = make(chan struct{})
|
||||
)
|
||||
|
||||
if *duration > 0 {
|
||||
deadline = start.Add(*duration)
|
||||
var (
|
||||
cam capture.Capture
|
||||
loop frameSource
|
||||
)
|
||||
|
||||
switch *source {
|
||||
case "screen":
|
||||
cam, err = capture.OpenPipeWire(*streamIndex, *withAudio)
|
||||
if err != nil {
|
||||
log.Fatalf("capture: %v", err)
|
||||
}
|
||||
loop = captureLoop{cam}
|
||||
log.Printf("teleportfling: advertising on %d, capturing screen via PipeWire", p)
|
||||
case "pattern":
|
||||
loop = &patternLoop{
|
||||
w: 1920,
|
||||
h: 1080,
|
||||
fps: *fps,
|
||||
}
|
||||
log.Printf("teleportfling: advertising on %d, streaming test pattern", p)
|
||||
default:
|
||||
log.Fatalf("teleportfling: unknown source %q (want screen or pattern)", *source)
|
||||
}
|
||||
|
||||
// Interrupt / SIGTERM handling.
|
||||
sigc := make(chan os.Signal, 1)
|
||||
signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
|
||||
|
||||
// — Audio goroutine: ~100 ms chunks of a silent stereo float32 tone —
|
||||
// A silent master tone keeps OBS's audio pipeline alive without needing
|
||||
// a mic. M2 will replace this with real captured audio.
|
||||
// — Audio loop —
|
||||
audioDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(audioDone)
|
||||
audioStart = time.Now()
|
||||
tick := time.NewTicker(audioInterval)
|
||||
defer tick.Stop()
|
||||
|
||||
framesPerChunk := int32(float64(sampleRate) * audioInterval.Seconds())
|
||||
pcm := make([]byte, 0, framesPerChunk*int32(speakers)*4)
|
||||
|
||||
sendAudio := func(ts uint64, chunkFrames int32) {
|
||||
// Right = left = 0 → digital silence.
|
||||
pcm = pcm[:0]
|
||||
for f := 0; f < int(chunkFrames); f++ {
|
||||
pcm = append(pcm, 0, 0, 0, 0, 0, 0, 0, 0)
|
||||
}
|
||||
packet, err := protocol.BuildWavePacket(ts, protocol.AudioFormatF32, int32(sampleRate), int32(speakers), chunkFrames, pcm)
|
||||
if err != nil {
|
||||
log.Printf("teleportfling: wave: %v", err)
|
||||
return
|
||||
}
|
||||
sender.Send(packet)
|
||||
var src io.ReadCloser
|
||||
switch {
|
||||
case cam == nil:
|
||||
// Pattern source: synthesize silence to keep the audio pipeline alive.
|
||||
src = capture.NewSilenceSource()
|
||||
case cam.Audio() != nil:
|
||||
src = cam.Audio()
|
||||
default:
|
||||
log.Printf("teleportfling: system audio unavailable, streaming silence")
|
||||
src = capture.NewSilenceSource()
|
||||
}
|
||||
defer func() { _ = src.Close() }()
|
||||
|
||||
for {
|
||||
select {
|
||||
case now := <-tick.C:
|
||||
ts := uint64(now.Sub(audioStart))
|
||||
// Compute frames elapsed since audioStart so timestamps are a
|
||||
// continuous stream (not aligned to tick boundaries).
|
||||
elapsedFrames := int64(now.Sub(audioStart) / (time.Second / time.Duration(sampleRate)))
|
||||
// Cumulative frames sent so far.
|
||||
sendAudio(ts, framesPerChunk)
|
||||
_ = elapsedFrames
|
||||
case <-stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
audioLoop(sender, src, start, stop)
|
||||
}()
|
||||
|
||||
// — Video loop —
|
||||
videoDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(videoDone)
|
||||
ticker := time.NewTicker(frameInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
var frameNum int64
|
||||
for {
|
||||
select {
|
||||
case now := <-ticker.C:
|
||||
ts := uint64(now.Sub(start))
|
||||
img := testPattern(*width, *height, int(frameNum))
|
||||
frameNum++
|
||||
|
||||
frameStart := time.Now()
|
||||
buf, err := encoder.Encode(img, *quality)
|
||||
if err != nil {
|
||||
log.Printf("teleportfling: jpeg: %v", err)
|
||||
continue
|
||||
}
|
||||
encodeDur := time.Since(frameStart)
|
||||
|
||||
packet, err := protocol.WritePacket(
|
||||
protocol.Header{Type: protocol.VideoType, Timestamp: ts, Size: int32(len(buf))},
|
||||
ptr(protocol.DefaultBT709Full()),
|
||||
nil,
|
||||
buf,
|
||||
)
|
||||
if err != nil {
|
||||
log.Printf("teleportfling: packet: %v", err)
|
||||
continue
|
||||
}
|
||||
sender.Send(packet)
|
||||
totalFrames.Add(1)
|
||||
|
||||
_ = encodeDur // stats below
|
||||
case <-stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
videoLoop(sender, encoder, loop, *fps, *quality, start, stop, &totalFrames)
|
||||
}()
|
||||
|
||||
// — Stats ticker —
|
||||
@@ -183,7 +152,7 @@ func main() {
|
||||
case <-func() <-chan struct{} {
|
||||
if *duration > 0 {
|
||||
ch := make(chan struct{})
|
||||
time.AfterFunc(time.Until(deadline), func() { close(ch) })
|
||||
time.AfterFunc(*duration, func() { close(ch) })
|
||||
return ch
|
||||
}
|
||||
return nil
|
||||
@@ -199,9 +168,158 @@ func main() {
|
||||
announcer.Stop()
|
||||
sender.Close()
|
||||
encoder.Close()
|
||||
if cam != nil {
|
||||
if err := cam.Close(); err != nil {
|
||||
log.Printf("teleportfling: capture close: %v", err)
|
||||
}
|
||||
}
|
||||
log.Printf("teleportfling: stopped after %s", time.Since(start).Round(time.Millisecond))
|
||||
}
|
||||
|
||||
// audioLoop reads raw PCM from src and emits WAVE packets in audioChunk-sized
|
||||
// pieces. PCM is assumed interleaved signed-16-bit at sampleRate/speakers.
|
||||
//
|
||||
// start is the shared reference clock used by the video loop: audio and video
|
||||
// timestamps must share one time base, otherwise a constant skew between them
|
||||
// makes OBS buffer one stream to re-sync the other, adding latency.
|
||||
func audioLoop(sender *output.Sender, src io.Reader, start time.Time, stop <-chan struct{}) {
|
||||
framesPerChunk := int(sampleRate) * int(audioChunk) / int(time.Second)
|
||||
chunkBytes := framesPerChunk * speakers * 2 // S16
|
||||
buf := make([]byte, chunkBytes)
|
||||
|
||||
for {
|
||||
n, err := io.ReadFull(src, buf)
|
||||
if n > 0 {
|
||||
frames := n / (speakers * 2)
|
||||
ts := uint64(time.Since(start))
|
||||
packet, perr := protocol.BuildWavePacket(ts, protocol.AudioFormatS16, sampleRate, speakers, int32(frames), buf[:n])
|
||||
if perr != nil {
|
||||
log.Printf("teleportfling: wave: %v", perr)
|
||||
} else {
|
||||
sender.Send(packet)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
|
||||
log.Printf("teleportfling: audio: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// videoLoop pulls frames from loop and sends them at fps, encoding each to
|
||||
// JPEG with the given quality.
|
||||
func videoLoop(sender *output.Sender, encoder *protocol.JPEGEncoder, loop frameSource, fps, quality int, start time.Time, stop <-chan struct{}, total *atomic.Int64) {
|
||||
frameInterval := time.Second / time.Duration(fps)
|
||||
next := start
|
||||
|
||||
for {
|
||||
frame, err := loop.Next()
|
||||
if err != nil {
|
||||
if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
|
||||
log.Printf("teleportfling: capture: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Drop frames if we're running ahead of the target fps (e.g. a 60 Hz
|
||||
// monitor captured at 30 fps) to keep timestamps monotonic.
|
||||
now := time.Now()
|
||||
if now.Before(next) {
|
||||
continue
|
||||
}
|
||||
next = now.Add(frameInterval)
|
||||
|
||||
ts := uint64(now.Sub(start))
|
||||
buf, err := encodeFrame(encoder, frame, quality)
|
||||
if err != nil {
|
||||
log.Printf("teleportfling: jpeg: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
packet, err := protocol.WritePacket(
|
||||
protocol.Header{Type: protocol.VideoType, Timestamp: ts, Size: int32(len(buf))},
|
||||
ptr(protocol.DefaultBT709Full()),
|
||||
nil,
|
||||
buf,
|
||||
)
|
||||
if err != nil {
|
||||
log.Printf("teleportfling: packet: %v", err)
|
||||
continue
|
||||
}
|
||||
sender.Send(packet)
|
||||
total.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
// encodeFrame compresses a captured frame based on its concrete type.
|
||||
func encodeFrame(encoder *protocol.JPEGEncoder, frame *capture.VideoFrame, quality int) ([]byte, error) {
|
||||
if frame.Pix != nil {
|
||||
// BGRA from the PipeWire backend.
|
||||
return encoder.EncodeBGRA(frame.Pix, frame.Width, frame.Height, quality)
|
||||
}
|
||||
return nil, errors.New("capture: unsupported frame type")
|
||||
}
|
||||
|
||||
// frameSource abstracts the frame source: real capture or the test pattern.
|
||||
type frameSource interface {
|
||||
Next() (*capture.VideoFrame, error)
|
||||
}
|
||||
|
||||
// captureLoop wraps the PipeWire capture backend.
|
||||
type captureLoop struct {
|
||||
cam capture.Capture
|
||||
}
|
||||
|
||||
func (c captureLoop) Next() (*capture.VideoFrame, error) {
|
||||
return c.cam.Video().NextFrame()
|
||||
}
|
||||
|
||||
// patternLoop synthesizes the M1 test pattern (colour bars + moving box).
|
||||
type patternLoop struct {
|
||||
w, h int
|
||||
fps int
|
||||
seq int64
|
||||
}
|
||||
|
||||
func (p *patternLoop) Next() (*capture.VideoFrame, error) {
|
||||
img := testPattern(p.w, p.h, int(p.seq))
|
||||
p.seq++
|
||||
return ycrcbToBGRA(img), nil
|
||||
}
|
||||
|
||||
// ycrcbToBGRA converts a YCbCr image to a BGRA VideoFrame so both sources
|
||||
// share the encode path (EncodeBGRA).
|
||||
func ycrcbToBGRA(img *image.YCbCr) *capture.VideoFrame {
|
||||
w, h := img.Rect.Dx(), img.Rect.Dy()
|
||||
frame := &capture.VideoFrame{
|
||||
Pix: make([]byte, w*h*4),
|
||||
Width: w,
|
||||
Height: h,
|
||||
Stride: w * 4,
|
||||
}
|
||||
for y := 0; y < h; y++ {
|
||||
for x := 0; x < w; x++ {
|
||||
yi := y*img.YStride + x
|
||||
ci := (y/2)*img.CStride + x/2
|
||||
r, g, b := color.YCbCrToRGB(img.Y[yi], img.Cb[ci], img.Cr[ci])
|
||||
off := (y*w + x) * 4
|
||||
frame.Pix[off], frame.Pix[off+1], frame.Pix[off+2], frame.Pix[off+3] = b, g, r, 255
|
||||
}
|
||||
}
|
||||
return frame
|
||||
}
|
||||
|
||||
// mustNewEncoder creates a JPEG encoder or panics.
|
||||
func mustNewEncoder() *protocol.JPEGEncoder {
|
||||
enc, err := protocol.NewJPEGEncoder()
|
||||
@@ -222,7 +340,7 @@ func addr(port int) string {
|
||||
// testPattern renders a standard SMPTE colour bar with a moving white box at
|
||||
// the given frame index. The result is a *image.YCbCr 4:2:0 image so the
|
||||
// encoder uses the YUV path — closest to what real PipeWire capture will
|
||||
// produce in M2.
|
||||
// produce.
|
||||
func testPattern(w, h, frame int) *image.YCbCr {
|
||||
img := image.NewYCbCr(image.Rect(0, 0, w, h), image.YCbCrSubsampleRatio420)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user