RSDK-8647 - Add save do_command

.gitignore

@@ -2,4 +2,5 @@
 FFmpeg
 *.mp4
 .env
-*.DS_Store
+*.DS_Store
+bin

Makefile

@@ -19,15 +19,20 @@ FFMPEG_OPTS ?= --prefix=$(FFMPEG_BUILD) \
                --disable-doc \
                --disable-everything \
                --enable-static \
-			   --enable-libx264 \
-			   --enable-gpl \
+               --enable-libx264 \
+               --enable-gpl \
                --enable-encoder=libx264 \
                --enable-muxer=segment \
                --enable-muxer=mp4 \
                --enable-demuxer=segment \
+               --enable-demuxer=concat \
+               --enable-demuxer=mov \
+               --enable-demuxer=mp4 \
+               --enable-parser=h264 \
                --enable-protocol=file \
                --enable-protocol=concat \
-               --enable-protocol=crypto
+               --enable-protocol=crypto \
+               --enable-bsf=h264_mp4toannexb
 
 # TODO: cleanup libx264 static link
 CGO_LDFLAGS := "-L$(FFMPEG_BUILD)/lib -lavcodec -lavutil -lavformat -l:libjpeg.a /usr/lib/aarch64-linux-gnu/libx264.a -lz"

README.md

@@ -1,5 +1,5 @@
 # Video Storage
-The `video-store` module brings security camera functionality to your smart machine! The module consumes a source [Camera](https://docs.viam.com/components/camera/) and a [Vision Service](https://docs.viam.com/services/vision/), saves the camera output as video files to disk, and filters which video clips are uploaded to the cloud based on triggers from the vision service.
+The `video-store` module brings security camera functionality to your smart machine! The module consumes a source [Camera](https://docs.viam.com/components/camera/) and saves the camera output as video files on disk. You can then later request to upload video slices to the cloud using the [save](#save) command, or request the video bytes directly using the [fetch](#fetch) command.
 
 > **Note:** This component is a work in progress and is not yet fully implemented.
 
@@ -9,15 +9,15 @@ Fill in the attributes as applicable to the component, according to the example
 
 ```json
     {
-      "name": "fv-cam",
+      "name": "video-store-1",
       "namespace": "rdk",
       "type": "camera",
-      "model": "viam:camera:video-store",
+      "model": "viam:video:storage",
       "attributes": {
-        "camera": "webcam-1", // name of the camera to use
-        "vision": "vision-service-1", // name of the vision service dependency
+        "camera": "webcam-1", 
+        "vision": "vision-service-1", 
         "storage": {
-            "clip_seconds": 30,
+            "segment_seconds": 30,
             "size_gb": 100
         },
         "video": {
@@ -26,18 +26,14 @@ Fill in the attributes as applicable to the component, according to the example
             "bitrate": 1000000,
             "preset": "medium"
         },
-        "objects": {
-            "Person": 0.8 // label key and threshold value
-        },
-        "cam_props": { // camera properties of the source camera
+        "cam_props": { 
             "width": 640,
             "height": 480,
             "framerate": 30
         },
       },
       "depends_on": [
         "webcam-1",
-        "vision-service-1"
       ]
     }
 ```
@@ -51,12 +47,31 @@ Make sure to configure a [Data Manager Service](https://docs.viam.com/services/d
       "type": "data_manager",
       "attributes": {
         "tags": [],
-        "additional_sync_paths": [
-          "/home/viam/.viam/video-upload/"
-        ],
+        "additional_sync_paths": [],
         "capture_disabled": true,
         "sync_interval_mins": 1,
         "capture_dir": ""
       }
     }
 ```
+
+## Commands
+
+### `save`
+```json
+{
+  "command": "save",
+  "from": <start_timestamp>, [required]
+  "to": <end_timestamp>, [required]
+  "metadata": <arbitrary_metadata_string> [optional]
+}
+```
+
+### `fetch`
+```json
+{
+  "command": "fetch",
+  "from": <start_timestamp>, [required]
+  "to": <end_timestamp>, [required]
+}
+```

cam/cam.go

@@ -19,37 +19,43 @@ import (
 )
 
 // Model is the model for the video storage camera component.
-// TODO(seanp): Personal module for now, should be movied to viam module in prod.
+// TODO(seanp): Personal module for now, should be moved to viam module in prod.
 var Model = resource.ModelNamespace("seanavery").WithFamily("video").WithModel("storage")
 
 const (
+	// Default values for the video storage camera component.
 	defaultSegmentSeconds = 30 // seconds
 	defaultStorageSize    = 10 // GB
 	defaultVideoCodec     = "h264"
 	defaultVideoBitrate   = 1000000
 	defaultVideoPreset    = "medium"
 	defaultVideoFormat    = "mp4"
-	defaultLogLevel       = "error"
-	defaultUploadPath     = ".viam/video-upload"
+	defaultUploadPath     = ".viam/capture/video-upload"
 	defaultStoragePath    = ".viam/video-storage"
+
+	defaultLogLevel = "info"
+	deleterInterval = 60 // seconds
 )
 
 type videostore struct {
 	resource.AlwaysRebuild
 	resource.TriviallyCloseable
 
-	name       resource.Name
-	conf       *Config
-	logger     logging.Logger
-	uploadPath string
+	name   resource.Name
+	conf   *Config
+	logger logging.Logger
 
 	cam    camera.Camera
 	stream gostream.VideoStream
 
 	workers rdkutils.StoppableWorkers
 
-	enc *encoder
-	seg *segmenter
+	enc  *encoder
+	seg  *segmenter
+	conc *concater
+
+	storagePath string
+	uploadPath  string
 }
 
 type storage struct {
@@ -121,10 +127,10 @@ func newvideostore(
 	}
 
 	// TODO(seanp): make this configurable
-	// logLevel := lookupLogID(defaultLogLevel)
-	logLevel := lookupLogID("debug")
+	logLevel := lookupLogID(defaultLogLevel)
 	ffmppegLogLevel(logLevel)
 
+	// Create encoder to handle encoding of frames.
 	// TODO(seanp): Forcing h264 for now until h265 is supported.
 	if newConf.Video.Codec != "h264" {
 		newConf.Video.Codec = defaultVideoCodec
@@ -138,7 +144,6 @@ func newvideostore(
 	if newConf.Video.Format == "" {
 		newConf.Video.Format = defaultVideoFormat
 	}
-
 	vs.enc, err = newEncoder(
 		logger,
 		newConf.Video.Codec,
@@ -152,29 +157,40 @@ func newvideostore(
 		return nil, err
 	}
 
+	// Create segmenter to handle segmentation of video stream into clips.
 	if newConf.Storage.SegmentSeconds == 0 {
 		newConf.Storage.SegmentSeconds = defaultSegmentSeconds
 	}
-	if newConf.Storage.SizeGB == 0 {
-		newConf.Storage.SizeGB = defaultStorageSize
-	}
 	if newConf.Storage.UploadPath == "" {
 		newConf.Storage.UploadPath = filepath.Join(getHomeDir(), defaultUploadPath, vs.name.Name)
 	}
 	if newConf.Storage.StoragePath == "" {
 		newConf.Storage.StoragePath = filepath.Join(getHomeDir(), defaultStoragePath, vs.name.Name)
 	}
-	vs.seg, err = newSegmenter(logger, vs.enc, newConf.Storage.SizeGB, newConf.Storage.SegmentSeconds, newConf.Storage.StoragePath)
+	vs.storagePath = newConf.Storage.StoragePath
+	vs.seg, err = newSegmenter(
+		logger,
+		vs.enc,
+		newConf.Storage.SizeGB,
+		newConf.Storage.SegmentSeconds,
+		newConf.Storage.StoragePath,
+	)
 	if err != nil {
 		return nil, err
 	}
 
+	// Create concater to handle concatenation of video clips when requested.
 	vs.uploadPath = newConf.Storage.UploadPath
 	err = createDir(vs.uploadPath)
 	if err != nil {
 		return nil, err
 	}
+	vs.conc, err = newConcater(logger, vs.storagePath, vs.uploadPath, vs.name.Name)
+	if err != nil {
+		return nil, err
+	}
 
+	// Start workers to process frames and clean up storage.
 	vs.workers = rdkutils.NewStoppableWorkers(vs.processFrames, vs.deleter)
 
 	return vs, nil
@@ -185,7 +201,13 @@ func (cfg *Config) Validate(path string) ([]string, error) {
 	if cfg.Camera == "" {
 		return nil, utils.NewConfigValidationFieldRequiredError(path, "camera")
 	}
-
+	// Check Storage
+	if cfg.Storage == (storage{}) {
+		return nil, utils.NewConfigValidationFieldRequiredError(path, "storage")
+	}
+	if cfg.Storage.SizeGB == 0 {
+		return nil, utils.NewConfigValidationFieldRequiredError(path, "size_gb")
+	}
 	// TODO(seanp): Remove once camera properties are returned from camera component.
 	if cfg.Properties == (cameraProperties{}) {
 		return nil, utils.NewConfigValidationFieldRequiredError(path, "cam_props")
@@ -198,8 +220,39 @@ func (vs *videostore) Name() resource.Name {
 	return vs.name
 }
 
-func (vs *videostore) DoCommand(_ context.Context, _ map[string]interface{}) (map[string]interface{}, error) {
-	return nil, resource.ErrDoUnimplemented
+// DoCommand processes the commands for the video storage camera component.
+func (vs *videostore) DoCommand(_ context.Context, command map[string]interface{}) (map[string]interface{}, error) {
+	cmd, ok := command["command"].(string)
+	if !ok {
+		return nil, errors.New("invalid command type")
+	}
+
+	switch cmd {
+	// Save command is used to concatenate video clips between the given timestamps.
+	// The concatenated video file is then uploaded to the cloud the upload path.
+	// The response contains the name of the uploaded file.
+	case "save":
+		vs.logger.Debug("save command received")
+		from, to, metadata, err := validateSaveCommand(command)
+		if err != nil {
+			return nil, err
+		}
+		uploadFilePath, err := vs.conc.concat(from, to, metadata)
+		if err != nil {
+			vs.logger.Error("failed to concat files ", err)
+			return nil, err
+		}
+		uploadFileName := filepath.Base(uploadFilePath)
+		return map[string]interface{}{
+			"command": "save",
+			"file":    uploadFileName,
+		}, nil
+	case "fetch":
+		vs.logger.Debug("fetch command received")
+		return nil, resource.ErrDoUnimplemented
+	default:
+		return nil, errors.New("invalid command")
+	}
 }
 
 func (vs *videostore) Images(_ context.Context) ([]camera.NamedImage, resource.ResponseMetadata, error) {
@@ -227,9 +280,7 @@ func (vs *videostore) Stream(_ context.Context, _ ...gostream.ErrorHandler) (gos
 }
 
 // processFrames reads frames from the camera, encodes, and writes to the segmenter
-// which chuncks video stream into clip files inside the storage directory. This is
-// meant for long term storage of video clips that are not necessarily triggered by
-// detections.
+// which chunks video stream into clip files inside the storage directory.
 // TODO(seanp): Should this be throttled to a certain FPS?
 func (vs *videostore) processFrames(ctx context.Context) {
 	for {
@@ -263,11 +314,11 @@ func (vs *videostore) processFrames(ctx context.Context) {
 	}
 }
 
-// deleter is a go routine that cleans up old clips if storage is full. It runs every
-// minute and deletes the oldest clip until the storage size is below the max.
+// deleter is a go routine that cleans up old clips if storage is full. Runs on interval
+// and deletes the oldest clip until the storage size is below the configured max.
 func (vs *videostore) deleter(ctx context.Context) {
 	// TODO(seanp): Using seconds for now, but should be minutes in prod.
-	ticker := time.NewTicker(60 * time.Second)
+	ticker := time.NewTicker(deleterInterval * time.Second)
 	defer ticker.Stop()
 	for {
 		select {
@@ -285,14 +336,14 @@ func (vs *videostore) deleter(ctx context.Context) {
 }
 
 // Close closes the video storage camera component.
-// It closes the stream, workers, encoder, segmenter, and watcher.
 func (vs *videostore) Close(ctx context.Context) error {
 	err := vs.stream.Close(ctx)
 	if err != nil {
 		return err
 	}
 	vs.workers.Stop()
-	vs.enc.Close()
-	vs.seg.Close()
+	vs.enc.close()
+	vs.seg.close()
+	vs.conc.close()
 	return nil
 }