Refactor the device model for streaming

The rewrite is working toward having a simpler model that can
represent streaming capture/output devices. The device package
determines how the device works when one is created and configured.

v4l2/device/device.go → device/device.go

@@ -14,25 +14,20 @@ type Device struct {
 	path         string
 	file         *os.File
 	fd           uintptr
-	config       Config
+	config       config
 	bufType      v4l2.BufType
 	cap          v4l2.Capability
 	cropCap      v4l2.CropCapability
 	buffers      [][]byte
 	requestedBuf v4l2.RequestBuffers
 	streaming    bool
+	output chan []byte
 }
 
-// Open creates opens the underlying device at specified path
-// and returns a *Device or an error if unable to open device.
+// Open creates opens the underlying device at specified path for streaming.
+// It returns a *Device or an error if unable to open device.
 func Open(path string, options ...Option) (*Device, error) {
-	file, err := os.OpenFile(path, sys.O_RDWR|sys.O_NONBLOCK, 0644)
-	//file, err := os.OpenFile(path, sys.O_RDWR, 0644)
-	if err != nil {
-		return nil, fmt.Errorf("device open: %w", err)
-	}
-	dev := &Device{path: path, file: file, fd: file.Fd(), config: Config{}}
-
+	dev := &Device{path: path, config: config{}}
 	// apply options
 	if len(options) > 0 {
 		for _, o := range options {
@@ -40,12 +35,15 @@ func Open(path string, options ...Option) (*Device, error) {
 		}
 	}
 
-	// ensures IOType is set
-	if reflect.ValueOf(dev.config.ioType).IsZero() {
-		dev.config.ioType = v4l2.IOTypeMMAP
+	file, err := os.OpenFile(path, sys.O_RDWR|sys.O_NONBLOCK, 0644)
+	//file, err := os.OpenFile(path, sys.O_RDWR, 0644)
+	if err != nil {
+		return nil, fmt.Errorf("device open: %w", err)
 	}
+	dev.file = file
+	dev.fd = file.Fd()
 
-	// set capability
+	// get capability
 	cap, err := v4l2.GetCapability(file.Fd())
 	if err != nil {
 		if err := file.Close(); err != nil {
@@ -55,9 +53,21 @@ func Open(path string, options ...Option) (*Device, error) {
 	}
 	dev.cap = cap
 
+	// set preferred device buffer size
+	if reflect.ValueOf(dev.config.bufSize).IsZero() {
+		dev.config.bufSize = 2
+	}
+
+	// only supports streaming IO model right now
+	if !dev.cap.IsStreamingSupported() {
+		return nil, fmt.Errorf("device open: only streaming IO is supported")
+	}
+
 	switch {
 	case cap.IsVideoCaptureSupported():
+		// setup capture parameters and chan for captured data
 		dev.bufType = v4l2.BufTypeVideoCapture
+		dev.output = make(chan []byte, dev.config.bufSize)
 	case cap.IsVideoOutputSupported():
 		dev.bufType = v4l2.BufTypeVideoOutput
 	default:
@@ -67,21 +77,24 @@ func Open(path string, options ...Option) (*Device, error) {
 		return nil, fmt.Errorf("device open: %s: %w", path, v4l2.ErrorUnsupportedFeature)
 	}
 
-	// set crop
-	cropCap, err := v4l2.GetCropCapability(file.Fd(), dev.bufType)
-	if err != nil {
-		if err := file.Close(); err != nil {
-			return nil, fmt.Errorf("device open: %s: closing after failure: %s", path, err)
-		}
-		return nil, fmt.Errorf("device open: %s: %w", path, err)
+	if !reflect.ValueOf(dev.config.bufType).IsZero() && dev.config.bufType != dev.bufType {
+		return nil, fmt.Errorf("device open: does not support buffer stream type")
+	}
+
+	// ensures IOType is set
+	if reflect.ValueOf(dev.config.ioType).IsZero() {
+		dev.config.ioType = v4l2.IOTypeMMAP
 	}
-	dev.cropCap = cropCap
 
 	// set pix format
 	if !reflect.ValueOf(dev.config.pixFormat).IsZero() {
 		if err := dev.SetPixFormat(dev.config.pixFormat); err != nil {
 			fmt.Errorf("device open: %s: set format: %w", path, err)
 		}
+	} else {
+		if dev.config.pixFormat, err = v4l2.GetPixFormat(dev.fd); err != nil {
+			fmt.Errorf("device open: %s: get pix format: %w", path, err)
+		}
 	}
 
 	// set fps
@@ -89,12 +102,12 @@ func Open(path string, options ...Option) (*Device, error) {
 		if err := dev.SetFrameRate(dev.config.fps); err != nil {
 			fmt.Errorf("device open: %s: set fps: %w", path, err)
 		}
+	} else {
+		if dev.config.fps, err = dev.GetFrameRate(); err != nil {
+			fmt.Errorf("device open: %s: get fps: %w", path, err)
+		}
 	}
 
-	// set preferred device buffer size
-	if reflect.ValueOf(dev.config.bufSize).IsZero() {
-		dev.config.bufSize = 2
-	}
 
 	return dev, nil
 }
@@ -102,7 +115,7 @@ func Open(path string, options ...Option) (*Device, error) {
 // Close closes the underlying device associated with `d` .
 func (d *Device) Close() error {
 	if d.streaming {
-		if err := d.StopStream(); err != nil {
+		if err := d.Stop(); err != nil {
 			return err
 		}
 	}
@@ -115,8 +128,8 @@ func (d *Device) Name() string {
 	return d.path
 }
 
-// FileDescriptor returns the file descriptor value for the device
-func (d *Device) FileDescriptor() uintptr {
+// Fd returns the file descriptor value for the device
+func (d *Device) Fd() uintptr {
 	return d.fd
 }
 
@@ -148,6 +161,18 @@ func (d *Device) MemIOType() v4l2.IOType {
 	return d.config.ioType
 }
 
+// GetOutput returns the channel that outputs streamed data that is
+// captured from the underlying device driver.
+func (d *Device) GetOutput() <-chan []byte {
+	return d.output
+}
+
+// SetInput sets up an input channel for data this sent for output to the
+// underlying device driver.
+func (d *Device) SetInput(in <-chan []byte) {
+
+}
+
 // GetCropCapability returns cropping info for device
 func (d *Device) GetCropCapability() (v4l2.CropCapability, error) {
 	if !d.cap.IsVideoCaptureSupported() {
@@ -251,6 +276,10 @@ func (d *Device) SetStreamParam(param v4l2.StreamParam) error {
 
 // SetFrameRate sets the FPS rate value of the device
 func (d *Device) SetFrameRate(fps uint32) error {
+	if !d.cap.IsStreamingSupported() {
+		return fmt.Errorf("set frame rate: %w", v4l2.ErrorUnsupportedFeature)
+	}
+
 	var param v4l2.StreamParam
 	switch {
 	case d.cap.IsVideoCaptureSupported():
@@ -292,57 +321,93 @@ func (d *Device) GetMediaInfo() (v4l2.MediaDeviceInfo, error) {
 	return v4l2.GetMediaDeviceInfo(d.fd)
 }
 
-func (d *Device) StartStream(ctx context.Context) (<-chan []byte, error) {
+func (d *Device) Start(ctx context.Context)  error {
 	if ctx.Err() != nil {
-		return nil, ctx.Err()
+		return ctx.Err()
 	}
 
 	if !d.cap.IsStreamingSupported() {
-		return nil, fmt.Errorf("device: start stream: %s", v4l2.ErrorUnsupportedFeature)
+		return fmt.Errorf("device: start stream: %s", v4l2.ErrorUnsupportedFeature)
 	}
 
 	if d.streaming {
-		return nil, fmt.Errorf("device: stream already started")
+		return fmt.Errorf("device: stream already started")
 	}
 
 	// allocate device buffers
 	bufReq, err := v4l2.InitBuffers(d)
 	if err != nil {
-		return nil, fmt.Errorf("device: init buffers: %w", err)
+		return fmt.Errorf("device: init buffers: %w", err)
 	}
 	d.config.bufSize = bufReq.Count // update with granted buf size
 	d.requestedBuf = bufReq
 
 	// for each allocated device buf, map into local space
-	if d.buffers, err = v4l2.MakeMappedBuffers(d); err != nil {
-		return nil, fmt.Errorf("device: make mapped buffers: %s", err)
+	if d.buffers, err = v4l2.MapMemoryBuffers(d); err != nil {
+		return fmt.Errorf("device: make mapped buffers: %s", err)
 	}
 
 	// Initial enqueue of buffers for capture
 	for i := 0; i < int(d.config.bufSize); i++ {
 		_, err := v4l2.QueueBuffer(d.fd, d.config.ioType, d.bufType, uint32(i))
 		if err != nil {
-			return nil, fmt.Errorf("device: initial buffer queueing: %w", err)
+			return fmt.Errorf("device: initial buffer queueing: %w", err)
 		}
 	}
 
-	dataChan, err := v4l2.StartStreamLoop(ctx, d)
-	if err != nil {
-		return nil, fmt.Errorf("device: start stream loop: %s", err)
+	if err := d.startStreamLoop(ctx); err != nil {
+		return fmt.Errorf("device: start stream loop: %s", err)
 	}
 
 	d.streaming = true
 
-	return dataChan, nil
+	return nil
 }
 
-func (d *Device) StopStream() error {
+func (d *Device) Stop() error {
 	d.streaming = false
-	if err := v4l2.UnmapBuffers(d); err != nil {
-		return fmt.Errorf("device: stop stream: %s", err)
+	if err := v4l2.UnmapMemoryBuffers(d); err != nil {
+		return fmt.Errorf("device: stop: %s", err)
 	}
-	if err := v4l2.StopStreamLoop(d); err != nil {
-		return fmt.Errorf("device: stop stream: %w", err)
+	if err := v4l2.StreamOff(d); err != nil {
+		return fmt.Errorf("device: stop: %w", err)
 	}
 	return nil
 }
+
+func (d *Device) startStreamLoop(ctx context.Context) error {
+	if err := v4l2.StreamOn(d); err != nil {
+		return fmt.Errorf("stream loop: stream on: %w", err)
+	}
+
+
+	go func() {
+		defer close(d.output)
+
+		fd := d.Fd()
+		ioMemType := d.MemIOType()
+		bufType := d.BufferType()
+
+		for {
+			select {
+			// handle stream capture (read from driver)
+			case <-v4l2.WaitForRead(d):
+				//TODO add better error-handling, for now just panic
+				buff, err  := v4l2.CaptureBuffer(fd, ioMemType, bufType)
+				if err != nil {
+					panic(fmt.Errorf("stream loop: buffer capture: %s", err).Error())
+				}
+
+				select {
+				case d.output <-d.Buffers()[buff.Index][:buff.BytesUsed]:
+				case <-ctx.Done():
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return nil
+}

v4l2/device/device_config.go → device/device_config.go

@@ -4,35 +4,48 @@ import (
 	"github.com/vladimirvivien/go4vl/v4l2"
 )
 
-type Config struct {
+type config struct {
 	ioType v4l2.IOType
 	pixFormat v4l2.PixFormat
 	bufSize uint32
 	fps uint32
+	bufType uint32
 }
 
-type Option func(*Config)
+type Option func(*config)
 
 func WithIOType(ioType v4l2.IOType) Option {
-	return func(o *Config) {
+	return func(o *config) {
 		o.ioType = ioType
 	}
 }
 
 func WithPixFormat(pixFmt v4l2.PixFormat) Option {
-	return func(o *Config) {
+	return func(o *config) {
 		o.pixFormat = pixFmt
 	}
 }
 
 func WithBufferSize(size uint32) Option {
-	return func(o *Config) {
+	return func(o *config) {
 		o.bufSize = size
 	}
 }
 
 func WithFPS(fps uint32) Option {
-	return func(o *Config) {
+	return func(o *config) {
 		o.fps = fps
 	}
+}
+
+func WithVideoCaptureEnabled() Option {
+	return func(o *config) {
+		o.bufType = v4l2.BufTypeVideoCapture
+	}
+}
+
+func WithVideoOutputEnabled() Option {
+	return func(o *config) {
+		o.bufType = v4l2.BufTypeVideoOutput
+	}
 }

device/doc.go

@@ -0,0 +1,2 @@
+// Package device provides a device abstraction that supports video streaming.
+package device

examples/capture/capture.go

@@ -7,8 +7,8 @@ import (
 	"log"
 	"os"
 
+	"github.com/vladimirvivien/go4vl/device"
 	"github.com/vladimirvivien/go4vl/v4l2"
-	"github.com/vladimirvivien/go4vl/v4l2/device"
 )
 
 func main() {
@@ -59,7 +59,7 @@ func main() {
 		log.Fatalf("device does not support any of %#v", preferredFmts)
 	}
 	log.Printf("Found preferred fmt: %s", fmtDesc)
-	frameSizes, err := v4l2.GetFormatFrameSizes(device.FileDescriptor(), fmtDesc.PixelFormat)
+	frameSizes, err := v4l2.GetFormatFrameSizes(device.Fd(), fmtDesc.PixelFormat)
 	if err!=nil{
 		log.Fatalf("failed to get framesize info: %s", err)
 	}
@@ -96,8 +96,7 @@ func main() {
 
 	// start stream
 	ctx, cancel := context.WithCancel(context.TODO())
-	frameChan, err := device.StartStream(ctx)
-	if err != nil {
+	if err := device.Start(ctx); err != nil {
 		log.Fatalf("failed to stream: %s", err)
 	}
 
@@ -106,7 +105,7 @@ func main() {
 	totalFrames := 10
 	count := 0
 	log.Printf("Capturing %d frames at %d fps...", totalFrames, fps)
-	for frame := range frameChan {
+	for frame := range device.GetOutput() {
 		fileName := fmt.Sprintf("capture_%d.jpg", count)
 		file, err := os.Create(fileName)
 		if err != nil {
@@ -128,7 +127,7 @@ func main() {
 	}
 
 	cancel() // stop capture
-	if err := device.StopStream(); err != nil {
+	if err := device.Stop(); err != nil {
 		fmt.Println(err)
 		os.Exit(1)
 	}

examples/device_info/devinfo.go

@@ -7,8 +7,8 @@ import (
 	"os"
 	"strings"
 
+	device2 "github.com/vladimirvivien/go4vl/device"
 	"github.com/vladimirvivien/go4vl/v4l2"
-	"github.com/vladimirvivien/go4vl/v4l2/device"
 )
 
 var template = "\t%-24s : %s\n"
@@ -27,7 +27,7 @@ func main() {
 		os.Exit(0)
 	}
 
-	device, err := device.Open(devName)
+	device, err := device2.Open(devName)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -64,12 +64,12 @@ func main() {
 }
 
 func listDevices() error {
-	paths, err := device.GetAllDevicePaths()
+	paths, err := device2.GetAllDevicePaths()
 	if err != nil {
 		return err
 	}
 	for _, path := range paths {
-		dev, err := device.Open(path)
+		dev, err := device2.Open(path)
 		if err != nil {
 			log.Print(err)
 			continue
@@ -101,17 +101,17 @@ func listDevices() error {
 			continue
 		}
 
-		fmt.Printf("Device [%s]: %s: %s\n", path, card, busInfo)
+		fmt.Printf("v4l2Device [%s]: %s: %s\n", path, card, busInfo)
 
 	}
 	return nil
 }
 
-func printDeviceDriverInfo(dev *device.Device) error {
+func printDeviceDriverInfo(dev *device2.Device) error {
 	caps := dev.Capability()
 
 	// print driver info
-	fmt.Println("Device Info:")
+	fmt.Println("v4l2Device Info:")
 	fmt.Printf(template, "Driver name", caps.Driver)
 	fmt.Printf(template, "Card name", caps.Card)
 	fmt.Printf(template, "Bus info", caps.BusInfo)
@@ -123,7 +123,7 @@ func printDeviceDriverInfo(dev *device.Device) error {
 		fmt.Printf("\t\t%s\n", desc.Desc)
 	}
 
-	fmt.Printf("\t%-16s : %0x\n", "Device capabilities", caps.Capabilities)
+	fmt.Printf("\t%-16s : %0x\n", "v4l2Device capabilities", caps.Capabilities)
 	for _, desc := range caps.GetDeviceCapDescriptions() {
 		fmt.Printf("\t\t%s\n", desc.Desc)
 	}
@@ -131,7 +131,7 @@ func printDeviceDriverInfo(dev *device.Device) error {
 	return nil
 }
 
-func printVideoInputInfo(dev *device.Device) error {
+func printVideoInputInfo(dev *device2.Device) error {
 	// first get current input
 	index, err := dev.GetVideoInputIndex()
 	if err != nil {
@@ -152,7 +152,7 @@ func printVideoInputInfo(dev *device.Device) error {
 	return nil
 }
 
-func printFormatInfo(dev *device.Device) error {
+func printFormatInfo(dev *device2.Device) error {
 	pixFmt, err := dev.GetPixFormat()
 	if err != nil {
 		return fmt.Errorf("video capture format: %w", err)
@@ -194,14 +194,14 @@ func printFormatInfo(dev *device.Device) error {
 	return printFormatDesc(dev)
 }
 
-func printFormatDesc(dev *device.Device) error {
+func printFormatDesc(dev *device2.Device) error {
 	descs, err := dev.GetFormatDescriptions()
 	if err != nil {
 		return fmt.Errorf("format desc: %w", err)
 	}
 	fmt.Println("Supported formats:")
 	for i, desc := range descs {
-		frmSizes, err := v4l2.GetFormatFrameSizes(dev.FileDescriptor(), desc.PixelFormat)
+		frmSizes, err := v4l2.GetFormatFrameSizes(dev.Fd(), desc.PixelFormat)
 		if err != nil {
 			return fmt.Errorf("format desc: %w", err)
 		}
@@ -215,7 +215,7 @@ func printFormatDesc(dev *device.Device) error {
 	return nil
 }
 
-func printCropInfo(dev *device.Device) error {
+func printCropInfo(dev *device2.Device) error {
 	crop, err := dev.GetCropCapability()
 	if err != nil {
 		return fmt.Errorf("crop capability: %w", err)
@@ -242,7 +242,7 @@ func printCropInfo(dev *device.Device) error {
 	return nil
 }
 
-func printCaptureParam(dev *device.Device) error {
+func printCaptureParam(dev *device2.Device) error {
 	params, err := dev.GetStreamParam()
 	if err != nil {
 		return fmt.Errorf("stream capture param: %w", err)
@@ -267,7 +267,7 @@ func printCaptureParam(dev *device.Device) error {
 }
 
 
-func printOutputParam(dev *device.Device) error {
+func printOutputParam(dev *device2.Device) error {
 	params, err := dev.GetStreamParam()
 	if err != nil {
 		return fmt.Errorf("stream output param: %w", err)

examples/format/devfmt.go

@@ -5,8 +5,8 @@ import (
 	"log"
 	"strings"
 
+	device2 "github.com/vladimirvivien/go4vl/device"
 	"github.com/vladimirvivien/go4vl/v4l2"
-	"github.com/vladimirvivien/go4vl/v4l2/device"
 )
 
 func main() {
@@ -31,10 +31,10 @@ func main() {
 		fmtEnc = v4l2.PixelFmtYUYV
 	}
 
-	device, err := device.Open(
+	device, err := device2.Open(
 		devName,
-		device.WithPixFormat(v4l2.PixFormat{Width: uint32(width), Height: uint32(height), PixelFormat: fmtEnc, Field: v4l2.FieldNone}),
-		device.WithFPS(15),
+		device2.WithPixFormat(v4l2.PixFormat{Width: uint32(width), Height: uint32(height), PixelFormat: fmtEnc, Field: v4l2.FieldNone}),
+		device2.WithFPS(15),
 	)
 	if err != nil {
 		log.Fatalf("failed to open device: %s", err)

examples/webcam/webcam.go

@@ -11,13 +11,13 @@ import (
 	"strings"
 	"time"
 
+	"github.com/vladimirvivien/go4vl/device"
 	"github.com/vladimirvivien/go4vl/imgsupport"
 	"github.com/vladimirvivien/go4vl/v4l2"
-	"github.com/vladimirvivien/go4vl/v4l2/device"
 )
 
 var (
-	frames <-chan []byte
+	frames chan []byte
 	fps    uint32 = 30
 	pixfmt v4l2.FourCCType
 )
@@ -157,15 +157,28 @@ func main() {
 
 	// start capture
 	ctx, cancel := context.WithCancel(context.TODO())
-	f, err := device.StartStream(ctx)
-	if err != nil {
+	if err := device.Start(ctx); err != nil {
 		log.Fatalf("stream capture: %s", err)
 	}
 	defer func() {
 		cancel()
 		device.Close()
 	}()
-	frames = f // make frames available.
+
+	// buffer captured video stream into a local channel of bytes
+	frames = make(chan []byte, 1024)
+	go func() {
+		defer close(frames)
+		for {
+			select {
+			case frame := <-device.GetOutput():
+				frames <- frame
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
 	log.Println("device capture started, frames available")
 
 	log.Printf("starting server on port %s", port)

imgsupport/converters.go

@@ -2,6 +2,7 @@ package imgsupport
 
 import (
 	"bytes"
+	"fmt"
 	"image"
 	"image/jpeg"
 )
@@ -9,6 +10,9 @@ import (
 // Yuyv2Jpeg attempts to convert the YUYV image using Go's built-in
 // YCbCr encoder
 func Yuyv2Jpeg(width, height int, frame []byte) ([]byte, error) {
+	if true {
+		return nil, fmt.Errorf("unsupported")
+	}
 	//size := len(frame)
 	ycbr := image.NewYCbCr(image.Rect(0, 0, width, height), image.YCbCrSubsampleRatio422)
 

v4l2/streaming.go

@@ -123,9 +123,9 @@ type PlaneInfo struct {
 // StreamOn requests streaming to be turned on for
 // capture (or output) that uses memory map, user ptr, or DMA buffers.
 // https://www.kernel.org/doc/html/latest/userspace-api/media/v4l/vidioc-streamon.html
-func StreamOn(dev Device) error {
+func StreamOn(dev StreamingDevice) error {
 	bufType := dev.BufferType()
-	if err := send(dev.FileDescriptor(), C.VIDIOC_STREAMON, uintptr(unsafe.Pointer(&bufType))); err != nil {
+	if err := send(dev.Fd(), C.VIDIOC_STREAMON, uintptr(unsafe.Pointer(&bufType))); err != nil {
 		return fmt.Errorf("stream on: %w", err)
 	}
 	return nil
@@ -134,9 +134,9 @@ func StreamOn(dev Device) error {
 // StreamOff requests streaming to be turned off for
 // capture (or output) that uses memory map, user ptr, or DMA buffers.
 // https://www.kernel.org/doc/html/latest/userspace-api/media/v4l/vidioc-streamon.html
-func StreamOff(dev Device) error {
+func StreamOff(dev StreamingDevice) error {
 	bufType := dev.BufferType()
-	if err := send(dev.FileDescriptor(), C.VIDIOC_STREAMOFF, uintptr(unsafe.Pointer(&bufType))); err != nil {
+	if err := send(dev.Fd(), C.VIDIOC_STREAMOFF, uintptr(unsafe.Pointer(&bufType))); err != nil {
 		return fmt.Errorf("stream off: %w", err)
 	}
 	return nil
@@ -145,7 +145,7 @@ func StreamOff(dev Device) error {
 // InitBuffers sends buffer allocation request to initialize buffer IO
 // for video capture or video output when using either mem map, user pointer, or DMA buffers.
 // See https://www.kernel.org/doc/html/latest/userspace-api/media/v4l/vidioc-reqbufs.html#vidioc-reqbufs
-func InitBuffers(dev Device) (RequestBuffers, error) {
+func InitBuffers(dev StreamingDevice) (RequestBuffers, error) {
 	if dev.MemIOType() != IOTypeMMAP && dev.MemIOType() != IOTypeDMABuf {
 		return RequestBuffers{}, fmt.Errorf("request buffers: %w", ErrorUnsupported)
 	}
@@ -154,7 +154,7 @@ func InitBuffers(dev Device) (RequestBuffers, error) {
 	req._type = C.uint(dev.BufferType())
 	req.memory = C.uint(dev.MemIOType())
 
-	if err := send(dev.FileDescriptor(), C.VIDIOC_REQBUFS, uintptr(unsafe.Pointer(&req))); err != nil {
+	if err := send(dev.Fd(), C.VIDIOC_REQBUFS, uintptr(unsafe.Pointer(&req))); err != nil {
 		return RequestBuffers{}, fmt.Errorf("request buffers: %w", err)
 	}
 
@@ -163,21 +163,21 @@ func InitBuffers(dev Device) (RequestBuffers, error) {
 
 // GetBuffer retrieves buffer info for allocated buffers at provided index.
 // This call should take place after buffers are allocated with RequestBuffers (for mmap for instance).
-func GetBuffer(dev Device, index uint32) (Buffer, error) {
+func GetBuffer(dev StreamingDevice, index uint32) (Buffer, error) {
 	var v4l2Buf C.struct_v4l2_buffer
 	v4l2Buf._type = C.uint(dev.BufferType())
 	v4l2Buf.memory = C.uint(dev.MemIOType())
 	v4l2Buf.index = C.uint(index)
 
-	if err := send(dev.FileDescriptor(), C.VIDIOC_QUERYBUF, uintptr(unsafe.Pointer(&v4l2Buf))); err != nil {
+	if err := send(dev.Fd(), C.VIDIOC_QUERYBUF, uintptr(unsafe.Pointer(&v4l2Buf))); err != nil {
 		return Buffer{}, fmt.Errorf("query buffer: %w", err)
 	}
 
 	return makeBuffer(v4l2Buf), nil
 }
 
-// MapMemoryBuffer creates a local buffer mapped to the address space of the device specified by fd.
-func MapMemoryBuffer(fd uintptr, offset int64, len int) ([]byte, error) {
+// mapMemoryBuffer creates a local buffer mapped to the address space of the device specified by fd.
+func mapMemoryBuffer(fd uintptr, offset int64, len int) ([]byte, error) {
 	data, err := sys.Mmap(int(fd), offset, len, sys.PROT_READ|sys.PROT_WRITE, sys.MAP_SHARED)
 	if err != nil {
 		return nil, fmt.Errorf("map memory buffer: %w", err)
@@ -185,8 +185,8 @@ func MapMemoryBuffer(fd uintptr, offset int64, len int) ([]byte, error) {
 	return data, nil
 }
 
-// MakeMappedBuffers creates mapped memory buffers for specified buffer count of device.
-func MakeMappedBuffers(dev Device)([][]byte, error) {
+// MapMemoryBuffers creates mapped memory buffers for specified buffer count of device.
+func MapMemoryBuffers(dev StreamingDevice)([][]byte, error) {
 	bufCount := int(dev.BufferCount())
 	buffers := make([][]byte, bufCount)
 	for i := 0; i < bufCount; i++ {
@@ -195,9 +195,11 @@ func MakeMappedBuffers(dev Device)([][]byte, error) {
 			return nil, fmt.Errorf("mapped buffers: %w", err)
 		}
 
+		// TODO check buffer flags for errors etc
+
 		offset := buffer.Info.Offset
 		length := buffer.Length
-		mappedBuf, err := MapMemoryBuffer(dev.FileDescriptor(), int64(offset), int(length))
+		mappedBuf, err := mapMemoryBuffer(dev.Fd(), int64(offset), int(length))
 		if err != nil {
 			return nil, fmt.Errorf("mapped buffers: %w", err)
 		}
@@ -206,21 +208,21 @@ func MakeMappedBuffers(dev Device)([][]byte, error) {
 	return buffers, nil
 }
 
-// UnmapMemoryBuffer removes the buffer that was previously mapped.
-func UnmapMemoryBuffer(buf []byte) error {
+// unmapMemoryBuffer removes the buffer that was previously mapped.
+func unmapMemoryBuffer(buf []byte) error {
 	if err := sys.Munmap(buf); err != nil {
 		return fmt.Errorf("unmap memory buffer: %w", err)
 	}
 	return nil
 }
 
-// UnmapBuffers unmaps all mapped memory buffer for device
-func UnmapBuffers(dev Device) error {
+// UnmapMemoryBuffers unmaps all mapped memory buffer for device
+func UnmapMemoryBuffers(dev StreamingDevice) error {
 	if dev.Buffers() == nil {
 		return fmt.Errorf("unmap buffers: uninitialized buffers")
 	}
 	for i := 0; i < len(dev.Buffers()); i++ {
-		if err := UnmapMemoryBuffer(dev.Buffers()[i]); err != nil {
+		if err := unmapMemoryBuffer(dev.Buffers()[i]); err != nil {
 			return fmt.Errorf("unmap buffers: %w", err)
 		}
 	}
@@ -261,22 +263,18 @@ func DequeueBuffer(fd uintptr, ioType IOType, bufType BufType) (Buffer, error) {
 	return makeBuffer(v4l2Buf), nil
 }
 
-// CaptureFrame captures a frame buffer from the device
-func CaptureFrame(dev Device) ([]byte, error) {
-	bufInfo, err := DequeueBuffer(dev.FileDescriptor(), dev.MemIOType(), dev.BufferType())
+// CaptureBuffer captures a frame buffer from the device
+func CaptureBuffer(fd uintptr, ioType IOType, bufType BufType) (Buffer, error) {
+	bufInfo, err := DequeueBuffer(fd, ioType, bufType)
 	if err != nil {
-		return nil, fmt.Errorf("capture frame: dequeue: %w", err)
-	}
-	// assert dequeued buffer is in proper range
-	if !(bufInfo.Index < dev.BufferCount()) {
-		return nil, fmt.Errorf("capture frame: buffer with unexpected index: %d (out of %d)", bufInfo.Index, dev.BufferCount())
+		return Buffer{}, fmt.Errorf("capture frame: dequeue: %w", err)
 	}
 
 	// requeue/clear used buffer, prepare for next read
-	if _, err := QueueBuffer(dev.FileDescriptor(), dev.MemIOType(), dev.BufferType(), bufInfo.Index); err != nil {
-		return nil, fmt.Errorf("capture frame: queue: %w", err)
+	if _, err := QueueBuffer(fd, ioType, bufType, bufInfo.Index); err != nil {
+		return Buffer{}, fmt.Errorf("capture frame: queue: %w", err)
 	}
 
 	// return captured buffer
-	return dev.Buffers()[bufInfo.Index][:bufInfo.BytesUsed], nil
+	return bufInfo, nil
 }

v4l2/streaming_loop.go

@@ -1,47 +1,13 @@
 package v4l2
 
 import (
-	"context"
 	"fmt"
 
 	sys "golang.org/x/sys/unix"
 )
 
-// StartStreamLoop issue a streaming request for the device and sets up
-// a loop to capture incoming buffers from the device.
-func StartStreamLoop(ctx context.Context, dev Device) (chan []byte, error) {
-	if err := StreamOn(dev); err != nil {
-		return nil, fmt.Errorf("stream loop: driver stream on: %w", err)
-	}
-
-	dataChan := make(chan []byte, dev.BufferCount())
-
-	go func() {
-		defer close(dataChan)
-		for {
-			select {
-			case <-WaitForRead(dev):
-				//TODO add better error-handling, for now just panic
-				frame, err  := CaptureFrame(dev)
-				if err != nil {
-					panic(fmt.Errorf("stream loop: frame capture: %s", err).Error())
-				}
-				select {
-				case dataChan <-frame:
-				case <-ctx.Done():
-					return
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
-
-	return dataChan, nil
-}
-
 // StopStreamLoop unmaps allocated IO memory and signal device to stop streaming
-func StopStreamLoop(dev Device) error {
+func StopStreamLoop(dev StreamingDevice) error {
 	if dev.Buffers() == nil {
 		return fmt.Errorf("stop loop: failed to stop loop: buffers uninitialized")
 	}
@@ -55,7 +21,7 @@ func StopStreamLoop(dev Device) error {
 func WaitForRead(dev Device) <-chan struct{} {
 	sigChan := make(chan struct{})
 
-	fd := dev.FileDescriptor()
+	fd := dev.Fd()
 
 	go func() {
 		defer close(sigChan)

v4l2/types.go

@@ -1,11 +1,39 @@
 package v4l2
 
+import (
+	"context"
+)
+
+// Device is the base interface for a v4l2 device
 type Device interface {
 	Name() string
-	FileDescriptor() uintptr
+	Fd() uintptr
 	Capability() Capability
+	MemIOType() IOType
+	GetOutput() <-chan []byte
+	SetInput(<-chan []byte)
+	Close() error
+}
+
+// StreamingDevice represents device that supports streaming IO
+// via mapped buffer sharing.
+type StreamingDevice interface {
+	Device
 	Buffers() [][]byte
 	BufferType() BufType
 	BufferCount() uint32
-	MemIOType() IOType
+	Start(context.Context) error
+	Stop() error
 }
+
+//// CaptureDevice represents a device that captures video from an underlying device
+//type CaptureDevice interface {
+//	StreamingDevice
+//	StartCapture(context.Context) (<-chan []byte, error)
+//}
+//
+//// OutputDevice represents a device that can output video data to an underlying device driver
+//type OutputDevice interface {
+//	StreamingDevice
+//	StartOutput(context.Context, chan<- []byte) error
+//}