Remove GetImages streaming feature flag and make it the default; Upda…

…te tests; Make lint

components/camera/camera.go

@@ -230,7 +230,6 @@ type Camera interface {
 	Properties(ctx context.Context) (Properties, error)
 }
 
-
 // VideoSource is a camera that has `Stream` embedded to directly integrate with gostream.
 // Note that generally, when writing camera components from scratch, embedding `Stream` is an anti-pattern.
 type VideoSource interface {

components/camera/camera_test.go

@@ -262,27 +262,6 @@ func TestCameraWithProjector(t *testing.T) {
 	test.That(t, cam2.Close(context.Background()), test.ShouldBeNil)
 }
 
-// verifyDecodedImage verifies that decoded image bytes match the original image.
-func verifyDecodedImage(t *testing.T, imgBytes []byte, mimeType string, originalImg image.Image) {
-	t.Helper()
-	test.That(t, len(imgBytes), test.ShouldBeGreaterThan, 0)
-
-	// For JPEG, compare the raw bytes instead of the decoded image since the decoded image is
-	// not guaranteed to be the same as the original image due to lossy compression.
-	if mimeType == rutils.MimeTypeJPEG {
-		expectedBytes, err := rimage.EncodeImage(context.Background(), originalImg, mimeType)
-		test.That(t, err, test.ShouldBeNil)
-		test.That(t, imgBytes, test.ShouldResemble, expectedBytes)
-		return
-	}
-
-	// For other formats, compare the decoded images
-	decodedImg, err := rimage.DecodeImage(context.Background(), imgBytes, mimeType)
-	test.That(t, err, test.ShouldBeNil)
-	test.That(t, rimage.ImagesExactlyEqual(decodedImg, originalImg), test.ShouldBeTrue)
-}
-
-
 // TestImages asserts the core expected behavior of the Images API.
 func TestImages(t *testing.T) {
 	ctx := context.Background()

components/camera/client_test.go

@@ -157,7 +157,11 @@ func TestClient(t *testing.T) {
 	injectCamera2.ProjectorFunc = func(ctx context.Context) (transform.Projector, error) {
 		return nil, errCameraProjectorFailed
 	}
-	injectCamera2.ImagesFunc = func(ctx context.Context, filterSourceNames []string, extra map[string]interface{}) ([]camera.NamedImage, resource.ResponseMetadata, error) {
+	injectCamera2.ImagesFunc = func(
+		ctx context.Context,
+		filterSourceNames []string,
+		extra map[string]interface{},
+	) ([]camera.NamedImage, resource.ResponseMetadata, error) {
 		return nil, resource.ResponseMetadata{}, errGetImageFailed
 	}
 
@@ -196,7 +200,7 @@ func TestClient(t *testing.T) {
 		test.That(t, len(namedImages) > 0, test.ShouldBeTrue)
 		frame, err := namedImages[0].Image(context.Background())
 		test.That(t, err, test.ShouldBeNil)
-		compVal, _, err := rimage.CompareImages(img, frame)
+		compVal, _, err := rimage.CompareImages(expectedColor, frame)
 		test.That(t, err, test.ShouldBeNil)
 		test.That(t, compVal, test.ShouldEqual, 0) // exact copy, no color conversion
 

components/camera/collectors_test.go

@@ -185,7 +185,11 @@ func newCamera(
 	pcd pointcloud.PointCloud,
 ) camera.Camera {
 	v := &inject.Camera{}
-	v.ImagesFunc = func(ctx context.Context, filterSourceNames []string, extra map[string]interface{}) ([]camera.NamedImage, resource.ResponseMetadata, error) {
+	v.ImagesFunc = func(
+		ctx context.Context,
+		filterSourceNames []string,
+		extra map[string]interface{},
+	) ([]camera.NamedImage, resource.ResponseMetadata, error) {
 		viamLogoJpegBytes, err := io.ReadAll(base64.NewDecoder(base64.StdEncoding, bytes.NewReader(viamLogoJpegB64)))
 		if err != nil {
 			return nil, resource.ResponseMetadata{}, err

components/camera/server.go

@@ -3,7 +3,6 @@ package camera
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/pkg/errors"
 	"go.opencensus.io/trace"
@@ -21,22 +20,17 @@ import (
 // serviceServer implements the CameraService from camera.proto.
 type serviceServer struct {
 	pb.UnimplementedCameraServiceServer
-	coll resource.APIResourceGetter[Camera]
-
-	imgTypesMu sync.RWMutex
-	imgTypes   map[string]ImageType
-	logger     logging.Logger
+	coll   resource.APIResourceGetter[Camera]
+	logger logging.Logger
 }
 
 // NewRPCServiceServer constructs an camera gRPC service server.
 // It is intentionally untyped to prevent use outside of tests.
 func NewRPCServiceServer(coll resource.APIResourceGetter[Camera]) interface{} {
 	logger := logging.NewLogger("camserver")
-	imgTypes := make(map[string]ImageType)
 	return &serviceServer{
-		coll:     coll,
-		logger:   logger,
-		imgTypes: imgTypes,
+		coll:   coll,
+		logger: logger,
 	}
 }
 

components/camera/server_test.go

@@ -26,7 +26,6 @@ import (
 )
 
 var (
-	errInvalidMimeType          = errors.New("invalid mime type")
 	errGeneratePointCloudFailed = errors.New("can't generate next point cloud")
 	errPropertiesFailed         = errors.New("can't get camera properties")
 	errCameraProjectorFailed    = errors.New("can't get camera properties")
@@ -168,7 +167,11 @@ func TestServer(t *testing.T) {
 	injectCamera2.ProjectorFunc = func(ctx context.Context) (transform.Projector, error) {
 		return nil, errCameraProjectorFailed
 	}
-	injectCamera2.ImagesFunc = func(ctx context.Context, filterSourceNames []string, extra map[string]interface{}) ([]camera.NamedImage, resource.ResponseMetadata, error) {
+	injectCamera2.ImagesFunc = func(
+		ctx context.Context,
+		filterSourceNames []string,
+		extra map[string]interface{},
+	) ([]camera.NamedImage, resource.ResponseMetadata, error) {
 		return nil, resource.ResponseMetadata{}, errGetImageFailed
 	}
 
@@ -413,4 +416,3 @@ func TestServer(t *testing.T) {
 		test.That(t, err.Error(), test.ShouldContainSubstring, errGetImageFailed.Error())
 	})
 }
-

components/camera/transformpipeline/pipeline_test.go

@@ -2,7 +2,6 @@ package transformpipeline
 
 import (
 	"context"
-	"errors"
 	"testing"
 
 	"github.com/pion/mediadevices/pkg/prop"
@@ -213,14 +212,23 @@ func TestTransformPipelineValidateFail(t *testing.T) {
 
 func TestVideoSourceFromCameraError(t *testing.T) {
 	malformedCam := &inject.Camera{
-		ImagesFunc: func(ctx context.Context, filterSourceNames []string, extra map[string]interface{}) ([]camera.NamedImage, resource.ResponseMetadata, error) {
+		ImagesFunc: func(
+			ctx context.Context,
+			filterSourceNames []string,
+			extra map[string]any,
+		) ([]camera.NamedImage, resource.ResponseMetadata, error) {
 			namedImg, _ := camera.NamedImageFromBytes([]byte("not a valid image"), "", utils.MimeTypePNG)
 			return []camera.NamedImage{namedImg}, resource.ResponseMetadata{}, nil
 		},
 	}
 
 	vs, err := videoSourceFromCamera(context.Background(), malformedCam)
+	test.That(t, err, test.ShouldBeNil)
+	test.That(t, vs, test.ShouldNotBeNil)
+
+	stream, err := vs.Stream(context.Background())
+	test.That(t, err, test.ShouldBeNil)
+	defer stream.Close(context.Background())
+	_, _, err = stream.Next(context.Background())
 	test.That(t, err, test.ShouldNotBeNil)
-	test.That(t, errors.Is(err, ErrVideoSourceCreation), test.ShouldBeTrue)
-	test.That(t, vs, test.ShouldBeNil)
 }

components/camera/videosource/webcam.go

@@ -422,7 +422,6 @@ func (c *webcam) ensureActive() error {
 	return nil
 }
 
-
 func (c *webcam) NextPointCloud(ctx context.Context) (pointcloud.PointCloud, error) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()

robot/client/client_test.go

@@ -347,7 +347,11 @@ func TestStatusClient(t *testing.T) {
 	var imgBuf bytes.Buffer
 	test.That(t, png.Encode(&imgBuf, img), test.ShouldBeNil)
 
-	injectCamera.ImagesFunc = func(ctx context.Context, filterSourceNames []string, extra map[string]interface{}) ([]camera.NamedImage, resource.ResponseMetadata, error) {
+	injectCamera.ImagesFunc = func(
+		ctx context.Context,
+		filterSourceNames []string,
+		extra map[string]interface{},
+	) ([]camera.NamedImage, resource.ResponseMetadata, error) {
 		namedImg, err := camera.NamedImageFromBytes(imgBuf.Bytes(), "", rutils.MimeTypePNG)
 		if err != nil {
 			return nil, resource.ResponseMetadata{}, err

robot/web/stream/camera/camera.go

@@ -10,51 +10,158 @@ import (
 
 	"go.viam.com/rdk/components/camera"
 	"go.viam.com/rdk/gostream"
+	"go.viam.com/rdk/resource"
+	"go.viam.com/rdk/rimage"
 	"go.viam.com/rdk/robot"
+	rutils "go.viam.com/rdk/utils"
 )
 
+var streamableImageMIMETypes = map[string]interface{}{
+	rutils.MimeTypeRawRGBA:  nil,
+	rutils.MimeTypeRawDepth: nil,
+	rutils.MimeTypeJPEG:     nil,
+	rutils.MimeTypePNG:      nil,
+	rutils.MimeTypeQOI:      nil,
+}
+
+// cropToEvenDimensions crops an image to even dimensions for x264 compatibility.
+// x264 only supports even resolutions. This ensures all streamed images work with x264.
+func cropToEvenDimensions(img image.Image) (image.Image, error) {
+	if img, ok := img.(*rimage.LazyEncodedImage); ok {
+		if err := img.DecodeConfig(); err != nil {
+			return nil, err
+		}
+	}
+
+	hasOddWidth := img.Bounds().Dx()%2 != 0
+	hasOddHeight := img.Bounds().Dy()%2 != 0
+	if !hasOddWidth && !hasOddHeight {
+		return img, nil
+	}
+
+	rImg := rimage.ConvertImage(img)
+	newWidth := rImg.Width()
+	newHeight := rImg.Height()
+	if hasOddWidth {
+		newWidth--
+	}
+	if hasOddHeight {
+		newHeight--
+	}
+	return rImg.SubImage(image.Rect(0, 0, newWidth, newHeight)), nil
+}
+
 // Camera returns the camera from the robot (derived from the stream) or
 // an error if it has no camera.
 func Camera(robot robot.Robot, stream gostream.Stream) (camera.Camera, error) {
 	// Stream names are slightly modified versions of the resource short name
-	shortName := stream.Name()
+	shortName := resource.SDPTrackNameToShortName(stream.Name())
 	cam, err := camera.FromRobot(robot, shortName)
 	if err != nil {
 		return nil, err
 	}
 	return cam, nil
 }
 
+// GetStreamableNamedImageFromCamera returns the first named image it finds from the camera that is supported for streaming.
+func GetStreamableNamedImageFromCamera(ctx context.Context, cam camera.Camera) (camera.NamedImage, error) {
+	namedImages, _, err := cam.Images(ctx, nil, nil)
+	if err != nil {
+		return camera.NamedImage{}, err
+	}
+	if len(namedImages) == 0 {
+		return camera.NamedImage{}, fmt.Errorf("no images received for camera %q", cam.Name())
+	}
+
+	for _, namedImage := range namedImages {
+		if _, ok := streamableImageMIMETypes[namedImage.MimeType()]; ok {
+			return namedImage, nil
+		}
+	}
+	return camera.NamedImage{}, fmt.Errorf("no images were found with a streamable mime type for camera %q", cam.Name())
+}
+
+// getImageBySourceName retrieves a specific named image from the camera by source name.
+func getImageBySourceName(ctx context.Context, cam camera.Camera, sourceName string) (camera.NamedImage, error) {
+	filterSourceNames := []string{sourceName}
+	namedImages, _, err := cam.Images(ctx, filterSourceNames, nil)
+	if err != nil {
+		return camera.NamedImage{}, err
+	}
+
+	switch len(namedImages) {
+	case 0:
+		return camera.NamedImage{}, fmt.Errorf("no images found for requested source name: %s", sourceName)
+	case 1:
+		namedImage := namedImages[0]
+		if namedImage.SourceName != sourceName {
+			return camera.NamedImage{}, fmt.Errorf("mismatched source name: requested %q, got %q", sourceName, namedImage.SourceName)
+		}
+		return namedImage, nil
+	default:
+		// At this point, multiple images were returned. This can happen if the camera is on an older version of the API and does not support
+		// filtering by source name, or if there is a bug in the camera resource's filtering logic. In this unfortunate case, we'll match the
+		// requested source name and tank the performance costs.
+		responseSourceNames := []string{}
+		for _, namedImage := range namedImages {
+			if namedImage.SourceName == sourceName {
+				return namedImage, nil
+			}
+			responseSourceNames = append(responseSourceNames, namedImage.SourceName)
+		}
+		return camera.NamedImage{},
+			fmt.Errorf("no matching source name found for multiple returned images: requested %q, got %q", sourceName, responseSourceNames)
+	}
+}
+
 // VideoSourceFromCamera converts a camera resource into a gostream VideoSource.
 // This is useful for streaming video from a camera resource.
 func VideoSourceFromCamera(ctx context.Context, cam camera.Camera) (gostream.VideoSource, error) {
+	// The reader callback uses a small state machine to determine which image to request from the camera.
+	// A `sourceName` is used to track the selected image source. On the first call, `sourceName` is nil,
+	// so the first available streamable image is chosen. On subsequent successful calls, the same `sourceName`
+	// is used. If any errors occur while getting an image, `sourceName` is reset to nil, and the selection
+	// process starts over on the next call. This allows the stream to recover if a source becomes unavailable.
+	var sourceName *string
 	reader := gostream.VideoReaderFunc(func(ctx context.Context) (image.Image, func(), error) {
-		namedImages, _, err := cam.Images(ctx, nil, nil)
+		var respNamedImage camera.NamedImage
+
+		if sourceName == nil {
+			namedImage, err := GetStreamableNamedImageFromCamera(ctx, cam)
+			if err != nil {
+				return nil, func() {}, err
+			}
+			respNamedImage = namedImage
+			sourceName = &namedImage.SourceName
+		} else {
+			var err error
+			respNamedImage, err = getImageBySourceName(ctx, cam, *sourceName)
+			if err != nil {
+				sourceName = nil
+				return nil, func() {}, err
+			}
+		}
+
+		img, err := respNamedImage.Image(ctx)
 		if err != nil {
+			sourceName = nil
 			return nil, func() {}, err
 		}
-		if len(namedImages) == 0 {
-			return nil, func() {}, fmt.Errorf("no images returned from camera")
-		}
-		img, err := namedImages[0].Image(ctx)
+
+		img, err = cropToEvenDimensions(img)
 		if err != nil {
+			sourceName = nil
 			return nil, func() {}, err
 		}
+
 		return img, func() {}, nil
 	})
 
-	namedImages, _, err := cam.Images(ctx, nil, nil)
+	img, _, err := reader(ctx)
 	if err != nil {
 		// Okay to return empty prop because processInputFrames will tick and set them
 		return gostream.NewVideoSource(reader, prop.Video{}), nil //nolint:nilerr
 	}
-	if len(namedImages) == 0 {
-		return gostream.NewVideoSource(reader, prop.Video{}), nil
-	}
-	img, err := namedImages[0].Image(ctx)
-	if err != nil {
-		return gostream.NewVideoSource(reader, prop.Video{}), nil //nolint:nilerr
-	}
 
 	return gostream.NewVideoSource(reader, prop.Video{Width: img.Bounds().Dx(), Height: img.Bounds().Dy()}), nil
 }