DATA-3072-split-prog-and-capture-files

cli/data.go

@@ -602,7 +602,7 @@ func filenameForDownload(meta *datapb.BinaryMetadata) string {
 	}
 
 	// Replace reserved characters.
-	fileName = data.CaptureFilePathWithReplacedReservedChars(fileName)
+	fileName = data.FilePathWithReplacedReservedChars(fileName)
 
 	return fileName
 }

data/capture_buffer.go

@@ -3,7 +3,7 @@
 import (
 	"sync"
 
 	v1 "go.viam.com/api/app/datasync/v1"
 )
 
 // CaptureBufferedWriter is a buffered, persistent queue of SensorData.
@@ -15,18 +15,18 @@
 
 // CaptureBuffer is a persistent queue of SensorData backed by a series of *data.CaptureFile.
 type CaptureBuffer struct {
-	Directory          string
-	MetaData           *v1.DataCaptureMetadata
-	nextFile           *CaptureFile
+	directory          string
+	metaData           *v1.DataCaptureMetadata
+	nextFile           *ProgFile
 	lock               sync.Mutex
 	maxCaptureFileSize int64
 }
 
 // NewCaptureBuffer returns a new Buffer.
 func NewCaptureBuffer(dir string, md *v1.DataCaptureMetadata, maxCaptureFileSize int64) *CaptureBuffer {
 	return &CaptureBuffer{
-		Directory:          dir,
-		MetaData:           md,
+		directory:          dir,
+		metaData:           md,
 		maxCaptureFileSize: maxCaptureFileSize,
 	}
 }
@@ -36,12 +36,24 @@
 // are still being written to are indicated with the extension
 // InProgressFileExt. Files that have finished being written to are indicated by
 // FileExt.
+func isBinary(item *v1.SensorData) bool {
+	if item == nil {
+		return false
+	}
+	switch item.Data.(type) {
+	case *v1.SensorData_Binary:
+		return true
+	default:
+		return false
+	}
+}
+
 func (b *CaptureBuffer) Write(item *v1.SensorData) error {
 	b.lock.Lock()
 	defer b.lock.Unlock()
 
-	if item.GetBinary() != nil {
-		binFile, err := NewCaptureFile(b.Directory, b.MetaData)
+	if isBinary(item) {
+		binFile, err := NewProgFile(b.directory, b.metaData)
 		if err != nil {
 			return err
 		}
@@ -55,19 +67,19 @@
 	}
 
 	if b.nextFile == nil {
-		nextFile, err := NewCaptureFile(b.Directory, b.MetaData)
+		nextFile, err := NewProgFile(b.directory, b.metaData)
 		if err != nil {
 			return err
 		}
 		b.nextFile = nextFile
 		// We want to special case on "CaptureAllFromCamera" because it is sensor data that contains images
 		// and their corresponding annotations. We want each image and its annotations to be stored in a
 		// separate file.
-	} else if b.nextFile.Size() > b.maxCaptureFileSize || b.MetaData.MethodName == "CaptureAllFromCamera" {
+	} else if b.nextFile.Size() > b.maxCaptureFileSize || b.metaData.MethodName == "CaptureAllFromCamera" {
 		if err := b.nextFile.Close(); err != nil {
 			return err
 		}
-		nextFile, err := NewCaptureFile(b.Directory, b.MetaData)
+		nextFile, err := NewProgFile(b.directory, b.metaData)
 		if err != nil {
 			return err
 		}
@@ -93,5 +105,5 @@
 
 // Path returns the path to the directory containing the backing data capture files.
 func (b *CaptureBuffer) Path() string {
-	return b.Directory
+	return b.directory
 }

data/capture_buffer_test.go

@@ -1,6 +1,7 @@
 package data
 
 import (
+	"crypto/sha1"
 	"errors"
 	"io"
 	"os"
@@ -267,7 +268,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f.Close()) }()
 
-				cf, err := ReadCaptureFile(f)
+				cf, err := NewCaptureFile(f)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 
@@ -335,7 +336,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f2.Close()) }()
 
-				cf2, err := ReadCaptureFile(f2)
+				cf2, err := NewCaptureFile(f2)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf2.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 
@@ -470,7 +471,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f.Close()) }()
 
-				cf, err := ReadCaptureFile(f)
+				cf, err := NewCaptureFile(f)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 
@@ -508,7 +509,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f2.Close()) }()
 
-				cf2, err := ReadCaptureFile(f2)
+				cf2, err := NewCaptureFile(f2)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf2.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 
@@ -563,7 +564,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				f3, err := os.Open(filepath.Join(b.Path(), newFileNames[0]))
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f3.Close()) }()
-				cf3, err := ReadCaptureFile(f3)
+				cf3, err := NewCaptureFile(f3)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf3.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 				sd3, err := cf3.ReadNext()
@@ -576,7 +577,7 @@ func TestCaptureBufferReader(t *testing.T) {
 				f4, err := os.Open(filepath.Join(b.Path(), newFileNames[1]))
 				test.That(t, err, test.ShouldBeNil)
 				defer func() { utils.UncheckedError(f4.Close()) }()
-				cf4, err := ReadCaptureFile(f4)
+				cf4, err := NewCaptureFile(f4)
 				test.That(t, err, test.ShouldBeNil)
 				test.That(t, cf4.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 				sd4, err := cf4.ReadNext()
@@ -620,7 +621,7 @@ func TestCaptureBufferReader(t *testing.T) {
 
 		// Path() is the same as the first paramenter passed to NewCaptureBuffer
 		test.That(t, b.Path(), test.ShouldResemble, tmpDir)
-		test.That(t, b.MetaData, test.ShouldResemble, readImageCaptureMetadata)
+		test.That(t, b.metaData, test.ShouldResemble, readImageCaptureMetadata)
 
 		now := time.Now()
 		timeRequested := timestamppb.New(now.UTC())
@@ -644,7 +645,7 @@ func TestCaptureBufferReader(t *testing.T) {
 		test.That(t, err, test.ShouldBeNil)
 		defer func() { utils.UncheckedError(f.Close()) }()
 
-		cf2, err := ReadCaptureFile(f)
+		cf2, err := NewCaptureFile(f)
 		test.That(t, err, test.ShouldBeNil)
 		test.That(t, cf2.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
 
@@ -657,7 +658,105 @@ func TestCaptureBufferReader(t *testing.T) {
 	})
 }
 
-//nolint
+func TestWriteReadBinary(t *testing.T) {
+	type testCase struct {
+		name string
+		data []byte
+	}
+	eightKBFilled := make([]byte, 1024*8)
+	for i := range eightKBFilled {
+		eightKBFilled[i] = uint8(i % 256)
+	}
+
+	oneMbFilled := make([]byte, 1024*1000)
+	for i := range eightKBFilled {
+		oneMbFilled[i] = uint8(i % 256)
+	}
+
+	eightMbFilled := make([]byte, 1024*1000*8)
+	for i := range eightMbFilled {
+		eightMbFilled[i] = uint8(i % 256)
+	}
+
+	tcs := []testCase{
+		{"empty data", []byte{}},
+		{"small data", []byte("this is a fake image")},
+		{"8kb empty", make([]byte, 1024*8)},
+		{"8kb filled", eightKBFilled},
+		{"1mb empty", make([]byte, 1024*1000)},
+		{"1mb filled", oneMbFilled},
+		{"8mb empty", make([]byte, 1024*1000*8)},
+		{"8mb filled", eightMbFilled},
+	}
+
+	for _, tc := range tcs {
+		s := sha1.New()
+		_, err := s.Write(tc.data)
+		test.That(t, err, test.ShouldBeNil)
+		expectedHash := s.Sum(nil)
+		tmpDir := t.TempDir()
+		name := resource.NewName(resource.APINamespaceRDK.WithComponentType("camera"), "my-cam")
+		additionalParams := map[string]string{"mime_type": rutils.MimeTypeJPEG, "test": "1"}
+		methodParams, err := rprotoutils.ConvertStringMapToAnyPBMap(additionalParams)
+		test.That(t, err, test.ShouldBeNil)
+
+		readImageCaptureMetadata := BuildCaptureMetadata(
+			name.API,
+			name.ShortName(),
+			readImage,
+			additionalParams,
+			methodParams,
+			[]string{"my", "tags"},
+		)
+
+		now := time.Now()
+		timeRequested := timestamppb.New(now.UTC())
+		timeReceived := timestamppb.New(now.Add(time.Millisecond).UTC())
+		msg := &v1.SensorData{
+			Metadata: &v1.SensorMetadata{
+				TimeRequested: timeRequested,
+				TimeReceived:  timeReceived,
+			},
+			Data: &v1.SensorData_Binary{
+				Binary: tc.data,
+			},
+		}
+
+		buf := NewCaptureBuffer(tmpDir, readImageCaptureMetadata, int64(4*1024))
+
+		test.That(t, buf.Path(), test.ShouldResemble, tmpDir)
+		test.That(t, buf.metaData, test.ShouldResemble, readImageCaptureMetadata)
+
+		test.That(t, buf.Write(msg), test.ShouldBeNil)
+		test.That(t, buf.Flush(), test.ShouldBeNil)
+		dirEntries, err := os.ReadDir(buf.Path())
+		test.That(t, err, test.ShouldBeNil)
+		test.That(t, len(dirEntries), test.ShouldEqual, 1)
+		test.That(t, filepath.Ext(dirEntries[0].Name()), test.ShouldResemble, CompletedCaptureFileExt)
+		f, err := os.Open(filepath.Join(buf.Path(), dirEntries[0].Name()))
+		test.That(t, err, test.ShouldBeNil)
+		t.Cleanup(func() { test.That(t, f.Close(), test.ShouldBeNil) })
+		t.Run(tc.name, func(t *testing.T) {
+			ret, err := f.Seek(0, io.SeekStart)
+			test.That(t, err, test.ShouldBeNil)
+			test.That(t, ret, test.ShouldEqual, 0)
+			cf2, err := NewCaptureFile(f)
+			test.That(t, err, test.ShouldBeNil)
+			test.That(t, cf2.ReadMetadata(), test.ShouldResemble, readImageCaptureMetadata)
+
+			next, err := cf2.ReadNext()
+			test.That(t, err, test.ShouldBeNil)
+			test.That(t, next.GetMetadata(), test.ShouldResemble, msg.GetMetadata())
+			h := sha1.New()
+			_, err = h.Write(next.GetBinary())
+			test.That(t, err, test.ShouldBeNil)
+			actualHash := h.Sum(nil)
+			test.That(t, actualHash, test.ShouldResemble, expectedHash)
+		})
+	}
+}
+
+// nolint
 func getCaptureFiles(dir string) (dcFiles, progFiles []string) {
 	_ = filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
 		if err != nil {