Implement frames-to-video command

This finishes the "Simple Blender Render" job type, as it can now be run
fully by a Worker!

FEATURES.md

@@ -44,8 +44,9 @@ Note that list is **not** in any specific order.
 - [x] Command runner
 - [x] Log uploader
 - [ ] POSIX signal handling (sleep/wake up)
-- [ ] Implement `create-video` command.
+- [x] Implement `frames-to-video` command.
-- [ ] Implement `move-to-final` command.
+- [x] Implement `move-directory` command.
+- [ ] Refactor CLI-running code by analyzing the current Blender and FFmpeg commands.
 
 ## Both
 

debug-job-render.sh

@@ -14,14 +14,14 @@ curl -v -X 'POST' \
   "settings": {
     "add_path_components": 0,
     "blender_cmd": "{blender}",
-    "blendfile": "/render/_flamenco/tests/jobs/2022-03-15-191442.471681-Demo for Peoples/flamenco-test.flamenco.blend",
+    "blendfile": "flamenco-test.blend",
     "chunk_size": 30,
     "format": "PNG",
     "fps": 24,
     "frames": "1-60",
     "image_file_extension": ".png",
     "images_or_video": "images",
-    "render_output_path": "/render/_flamenco/tests/renders/sybren/Demo for Peoples/2022-03-15_191444/######",
+    "render_output_path": "/render/_flamenco/tests/renders/sybren/Demo for Peoples/######",
     "render_output_root": "/render/_flamenco/tests/renders/sybren/",
     "video_container_format": "MPEG1"
   },

internal/manager/job_compilers/job_compilers_test.go

@@ -15,6 +15,11 @@ import (
 	"git.blender.org/flamenco/pkg/api"
 )
 
+// The example job is expected to result in these arguments for FFmpeg.
+var expectedFramesToVideoArgs = []interface{}{
+	"-c:v", "h264", "-crf", "20", "-g", "18", "-vf", "pad=ceil(iw/2)*2:ceil(ih/2)*2", "-pix_fmt", "yuv420p", "-r", int64(24), "-y",
+}
+
 func exampleSubmittedJob() api.SubmittedJob {
 	settings := api.JobSettings{
 		AdditionalProperties: map[string]interface{}{
@@ -109,9 +114,10 @@ func TestSimpleBlenderRenderHappy(t *testing.T) {
 	assert.Equal(t, 1, len(tVideo.Commands))
 	assert.Equal(t, "create-video", tVideo.Commands[0].Name)
 	assert.EqualValues(t, AuthoredCommandParameters{
-		"input_files": "/render/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/*.png",
+		"exe":        "{ffmpeg}",
-		"output_file": "/render/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"inputGlob":  "/render/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/*.png",
-		"fps":         int64(24),
+		"outputFile": "/render/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"args":       expectedFramesToVideoArgs,
 	}, tVideo.Commands[0].Parameters)
 
 	for index, task := range aj.Tasks {
@@ -191,9 +197,10 @@ func TestSimpleBlenderRenderWindowsPaths(t *testing.T) {
 	assert.Equal(t, 1, len(tVideo.Commands))
 	assert.Equal(t, "create-video", tVideo.Commands[0].Name)
 	assert.EqualValues(t, AuthoredCommandParameters{
-		"input_files": "R:/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/*.png",
+		"exe":        "{ffmpeg}",
-		"output_file": "R:/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"inputGlob":  "R:/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/*.png",
-		"fps":         int64(24),
+		"outputFile": "R:/sprites/farm_output/promo/square_ellie/square_ellie.lighting_light_breakdown2__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"args":       expectedFramesToVideoArgs,
 	}, tVideo.Commands[0].Parameters)
 }
 
@@ -238,9 +245,10 @@ func TestSimpleBlenderRenderOutputPathFieldReplacement(t *testing.T) {
 
 	tVideo := aj.Tasks[4] // This should be a video encoding task
 	assert.EqualValues(t, AuthoredCommandParameters{
-		"input_files": "/root/2006-01-02_090405/jobname__intermediate-2006-01-02_090405/*.png",
+		"exe":        "{ffmpeg}",
-		"output_file": "/root/2006-01-02_090405/jobname__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"inputGlob":  "/root/2006-01-02_090405/jobname__intermediate-2006-01-02_090405/*.png",
-		"fps":         int64(24),
+		"outputFile": "/root/2006-01-02_090405/jobname__intermediate-2006-01-02_090405/scene123-1-10.mp4",
+		"args":       expectedFramesToVideoArgs,
 	}, tVideo.Commands[0].Parameters)
 
 }

internal/manager/job_compilers/scripts/simple_blender_render.js

@@ -148,11 +148,20 @@ function authorCreateVideoTask(settings, renderDir) {
     const outfile = path.join(renderDir, `${stem}-${settings.frames}.mp4`);
     const outfileExt = guessOutputFileExtension(settings);
 
-    const task = author.Task('create-video', 'ffmpeg');
+    const task = author.Task('preview-video', 'ffmpeg');
-    const command = author.Command("create-video", {
+    const command = author.Command("frames-to-video", {
-        input_files: path.join(renderDir, `*${outfileExt}`),
+        exe: "{ffmpeg}",
-        output_file: outfile,
+        inputGlob: path.join(renderDir, `*${outfileExt}`),
-        fps: settings.fps,
+        outputFile: outfile,
+        args: [
+            "-c:v", "h264",
+            "-crf", "20",
+            "-g", "18",
+            "-vf", "pad=ceil(iw/2)*2:ceil(ih/2)*2",
+            "-pix_fmt", "yuv420p",
+            "-r", settings.fps,
+            "-y", // Be sure to always pass either "-n" or "-y".
+        ],
     });
     task.addCommand(command);
 

internal/worker/command_exe.go

@@ -76,6 +76,9 @@ func NewCommandExecutor(cli CommandLineRunner, listener CommandListener, timeSer
 		// blender
 		"blender-render": ce.cmdBlenderRender,
 
+		// ffmpeg
+		"frames-to-video": ce.cmdFramesToVideo,
+
 		// file-management
 		"move-directory": ce.cmdMoveDirectory,
 	}

internal/worker/command_ffmpeg.go

@@ -0,0 +1,205 @@
+package worker
+
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+/* This file contains the commands in the "ffmpeg" type group. */
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"os/exec"
+	"runtime"
+
+	"github.com/google/shlex"
+	"github.com/rs/zerolog"
+
+	"git.blender.org/flamenco/pkg/api"
+	"git.blender.org/flamenco/pkg/crosspath"
+)
+
+type CreateVideoParams struct {
+	exe        string   // Expansion of `{ffmpeg}`: executable path + its CLI parameters defined by the Manager.
+	inputGlob  string   // Glob of input files.
+	outputFile string   // File to save the video to.
+	argsBefore []string // Additional CLI arguments from `exe`.
+	args       []string // Additional CLI arguments defined by the job compiler script, to between the input and output filenames.
+}
+
+// cmdFramesToVideo uses ffmpeg to concatenate image frames to a video file.
+func (ce *CommandExecutor) cmdFramesToVideo(ctx context.Context, logger zerolog.Logger, taskID string, cmd api.Command) error {
+	cmdCtx, cmdCtxCancel := context.WithCancel(ctx)
+	defer cmdCtxCancel()
+
+	execCmd, cleanup, err := ce.cmdFramesToVideoExeCommand(cmdCtx, logger, taskID, cmd)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	execCmd.Stderr = execCmd.Stdout // Redirect stderr to stdout.
+	outPipe, err := execCmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+
+	if err := execCmd.Start(); err != nil {
+		logger.Error().Err(err).Msg("error starting CLI execution")
+		return err
+	}
+
+	ffmpegPID := execCmd.Process.Pid
+	logger = logger.With().Int("pid", ffmpegPID).Logger()
+
+	reader := bufio.NewReaderSize(outPipe, StdoutBufferSize)
+	logChunker := NewLogChunker(taskID, ce.listener, ce.timeService)
+
+	for {
+		lineBytes, isPrefix, readErr := reader.ReadLine()
+		if readErr == io.EOF {
+			break
+		}
+		if readErr != nil {
+			logger.Error().Err(err).Msg("error reading stdout/err")
+			return err
+		}
+		line := string(lineBytes)
+		if isPrefix {
+			logger.Warn().
+				Str("line", fmt.Sprintf("%s...", line[:256])).
+				Int("lineLength", len(line)).
+				Msg("unexpectedly long line read, truncating")
+		}
+
+		logger.Debug().Msg(line)
+		logChunker.Append(ctx, fmt.Sprintf("pid=%d > %s", ffmpegPID, line))
+	}
+	logChunker.Flush(ctx)
+
+	if err := execCmd.Wait(); err != nil {
+		logger.Error().Err(err).Msg("error in CLI execution")
+		return err
+	}
+
+	if execCmd.ProcessState.Success() {
+		logger.Info().Msg("command exited succesfully")
+	} else {
+		logger.Error().
+			Int("exitCode", execCmd.ProcessState.ExitCode()).
+			Msg("command exited abnormally")
+		return fmt.Errorf("command exited abnormally with code %d", execCmd.ProcessState.ExitCode())
+	}
+
+	return nil
+}
+
+func (ce *CommandExecutor) cmdFramesToVideoExeCommand(
+	ctx context.Context,
+	logger zerolog.Logger,
+	taskID string,
+	cmd api.Command,
+) (*exec.Cmd, func(), error) {
+	parameters, err := cmdFramesToVideoParams(logger, cmd)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	inputGlobArgs, cleanup := parameters.getInputGlob()
+
+	cliArgs := make([]string, 0)
+	cliArgs = append(cliArgs, parameters.argsBefore...)
+	cliArgs = append(cliArgs, inputGlobArgs...)
+	cliArgs = append(cliArgs, parameters.args...)
+	cliArgs = append(cliArgs, parameters.outputFile)
+
+	execCmd := ce.cli.CommandContext(ctx, parameters.exe, cliArgs...)
+	if execCmd == nil {
+		logger.Error().
+			Str("cmdName", cmd.Name).
+			Msg("unable to create command executor")
+		return nil, nil, ErrNoExecCmd
+	}
+	logger.Info().
+		Str("cmdName", cmd.Name).
+		Str("execCmd", execCmd.String()).
+		Msg("going to execute FFmpeg")
+
+	if err := ce.listener.LogProduced(ctx, taskID, fmt.Sprintf("going to run: %s %q", parameters.exe, cliArgs)); err != nil {
+		return nil, nil, err
+	}
+
+	return execCmd, cleanup, nil
+}
+
+func cmdFramesToVideoParams(logger zerolog.Logger, cmd api.Command) (CreateVideoParams, error) {
+	var (
+		parameters CreateVideoParams
+		ok         bool
+	)
+
+	if parameters.exe, ok = cmdParameter[string](cmd, "exe"); !ok || parameters.exe == "" {
+		logger.Warn().Interface("command", cmd).Msg("missing 'exe' parameter")
+		return parameters, fmt.Errorf("missing 'exe' parameter: %+v", cmd.Parameters)
+	}
+	if parameters.inputGlob, ok = cmdParameter[string](cmd, "inputGlob"); !ok || parameters.inputGlob == "" {
+		logger.Warn().Interface("command", cmd).Msg("missing 'inputGlob' parameter")
+		return parameters, fmt.Errorf("missing 'inputGlob' parameter: %+v", cmd.Parameters)
+	}
+	if parameters.outputFile, ok = cmdParameter[string](cmd, "outputFile"); !ok || parameters.outputFile == "" {
+		logger.Warn().Interface("command", cmd).Msg("missing 'outputFile' parameter")
+		return parameters, fmt.Errorf("missing 'outputFile' parameter: %+v", cmd.Parameters)
+	}
+	if parameters.argsBefore, ok = cmdParameterAsStrings(cmd, "argsBefore"); !ok {
+		logger.Warn().Interface("command", cmd).Msg("invalid 'argsBefore' parameter")
+		return parameters, fmt.Errorf("invalid 'argsBefore' parameter: %+v", cmd.Parameters)
+	}
+	if parameters.args, ok = cmdParameterAsStrings(cmd, "args"); !ok {
+		logger.Warn().Interface("command", cmd).Msg("invalid 'args' parameter")
+		return parameters, fmt.Errorf("invalid 'args' parameter: %+v", cmd.Parameters)
+	}
+
+	// Move any CLI args from 'exe' to 'argsBefore'.
+	exeArgs, err := shlex.Split(parameters.exe)
+	if err != nil {
+		logger.Warn().Err(err).Interface("command", cmd).Msg("error parsing 'exe' parameter with shlex")
+		return parameters, fmt.Errorf("parsing 'exe' parameter %q: %w", parameters.exe, err)
+	}
+	if len(exeArgs) > 1 {
+		allArgsBefore := []string{}
+		allArgsBefore = append(allArgsBefore, exeArgs[1:]...)
+		allArgsBefore = append(allArgsBefore, parameters.argsBefore...)
+		parameters.exe = exeArgs[0]
+		parameters.argsBefore = allArgsBefore
+	}
+
+	return parameters, nil
+}
+
+// getInputGlob constructs CLI arguments for FFmpeg input file globbing.
+// The 2nd return value is a cleanup function.
+func (p *CreateVideoParams) getInputGlob() ([]string, func()) {
+	if runtime.GOOS == "windows" {
+		// FFMpeg on Windows doesn't support globbing, so we have to do that in Go
+		// instead.
+		// TODO: implement this!
+		// 	index_file = self.create_index_file(input_files)
+		// 	args += [
+		// 			"-f",
+		// 			"concat",
+		// 			"-i",
+		// 			index_file.as_posix(),
+		// 	]
+		// TODO: the returned cleanup function should delete the index file.
+		panic("not implemented yet")
+	}
+
+	cliArgs := []string{
+		"-pattern_type",
+		"glob",
+		"-i",
+		crosspath.ToSlash(p.inputGlob),
+	}
+	cleanup := func() {}
+	return cliArgs, cleanup
+}

internal/worker/command_ffmpeg_test.go

@@ -0,0 +1,49 @@
+package worker
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+
+	"git.blender.org/flamenco/pkg/api"
+)
+
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+func TestCmdFramesToVideoSimple(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	ce, mocks := testCommandExecutor(t, mockCtrl)
+
+	taskID := "1d54c6fe-1242-4c8f-bd63-5a09e358d7b6"
+	cmd := api.Command{
+		Name: "blender",
+		Parameters: map[string]interface{}{
+			"exe":        "/path/to/ffmpeg -v quiet",
+			"argsBefore": []string{"-report"},
+			"inputGlob":  "path/to/renders/*.png",
+			"args": []string{
+				"-c:v", "hevc",
+				"-crf", "31",
+				"-vf", "pad=ceil(iw/2)*2:ceil(ih/2)*2",
+			},
+			"outputFile": "path/to/renders/preview.mkv",
+		},
+	}
+
+	cliArgs := []string{
+		"-v", "quiet", // exe
+		"-report",                                              // argsBefore
+		"-pattern_type", "glob", "-i", "path/to/renders/*.png", // inputGlob
+		"-c:v", "hevc", "-crf", "31", "-vf", "pad=ceil(iw/2)*2:ceil(ih/2)*2", // args
+		"path/to/renders/preview.mkv", // outputFile
+	}
+	mocks.cli.EXPECT().CommandContext(gomock.Any(), "/path/to/ffmpeg", cliArgs).Return(nil)
+
+	err := ce.cmdFramesToVideo(context.Background(), zerolog.Nop(), taskID, cmd)
+	assert.Equal(t, ErrNoExecCmd, err, "nil *exec.Cmd should result in ErrNoExecCmd")
+}

internal/worker/config.go

@@ -24,6 +24,11 @@ const (
 	configFilename      = "flamenco-worker.yaml"
 )
 
+var defaultConfig = WorkerConfig{
+	ConfiguredManager: "", // Auto-detect by default.
+	TaskTypes:         []string{"blender", "ffmpeg", "file-management", "misc"},
+}
+
 // WorkerConfig represents the configuration of a single worker.
 // It does not include authentication credentials.
 type WorkerConfig struct {
@@ -125,10 +130,7 @@ func (fcw *FileConfigWrangler) SetManagerURL(managerURL string) WorkerConfig {
 
 // DefaultConfig returns a fairly sane default configuration.
 func (fcw FileConfigWrangler) DefaultConfig() WorkerConfig {
-	return WorkerConfig{
+	return defaultConfig
-		ConfiguredManager: "", // Auto-detect by default.
-		TaskTypes:         []string{"blender", "file-management", "exr-merge", "misc"},
-	}
 }
 
 // WriteConfig stores a struct as YAML file.