Merge pull request #172 from bcc-code/feat/import-audio-form-mu1-mu2

Import audo form MU1 and MU2

activities/analyze.go

@@ -2,6 +2,7 @@ package activities
 
 import (
 	"context"
+	"github.com/bcc-code/bcc-media-flows/utils"
 
 	"github.com/bcc-code/bcc-media-flows/paths"
 	"github.com/bcc-code/bcc-media-flows/services/ffmpeg"
@@ -22,3 +23,39 @@ func AnalyzeFile(ctx context.Context, input AnalyzeFileParams) (*ffmpeg.StreamIn
 	}
 	return &info, nil
 }
+
+type GetVideoOffsetInput struct {
+	VideoPath1      paths.Path
+	VideoPath2      paths.Path
+	VideoFPS        int
+	AudioSampleRate int
+}
+
+func GetVideoOffset(ctx context.Context, input GetVideoOffsetInput) (int, error) {
+	log := activity.GetLogger(ctx)
+	activity.RecordHeartbeat(ctx, "GetVideoOffset")
+	log.Info("Starting GetVideoOffsetActivity")
+
+	video1TC, err := ffmpeg.GetTimeCode(input.VideoPath1.Local())
+	if err != nil {
+		return 0, err
+	}
+
+	video2TC, err := ffmpeg.GetTimeCode(input.VideoPath2.Local())
+	if err != nil {
+		return 0, err
+	}
+
+	// Video from YouPlay is always 25fps and 48000Hz
+	videoSamples1, err := utils.TCToSamples(video1TC, input.VideoFPS, input.AudioSampleRate)
+	if err != nil {
+		return 0, err
+	}
+
+	videoSamples2, err := utils.TCToSamples(video2TC, input.VideoFPS, input.AudioSampleRate)
+	if err != nil {
+		return 0, err
+	}
+
+	return videoSamples2 - videoSamples1, nil
+}

activities/audio.go

@@ -2,13 +2,15 @@ package activities
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/bcc-code/bcc-media-flows/common"
 	"github.com/bcc-code/bcc-media-flows/paths"
 	"github.com/bcc-code/bcc-media-flows/services/ffmpeg"
 	"github.com/bcc-code/bcc-media-flows/services/transcode"
 	"github.com/bcc-code/bcc-media-flows/utils"
 	"github.com/go-errors/errors"
+	"github.com/samber/lo"
 	"go.temporal.io/sdk/activity"
 )
 
@@ -84,3 +86,63 @@ func DetectSilence(ctx context.Context, input common.AudioInput) (bool, error) {
 
 	return transcode.AudioIsSilent(input.Path)
 }
+
+type ExtractAudioInput struct {
+	VideoPath       paths.Path
+	OutputFolder    paths.Path
+	FileNamePattern string
+	Channels        []int
+}
+
+type ExtractAudioOutput struct {
+	AudioFiles map[int]paths.Path
+}
+
+// ExtractAudio extracts audio from a video file.
+//   - VideoPath: the path to the video file
+//   - OutputFolder: the folder where the audio files will be saved
+//   - FileNamePattern: the pattern for the audio files. The pattern should contain one %d which will be replaced by the channel number
+//   - Channels: the channels to extract. If empty, all channels will be extracted
+func ExtractAudio(ctx context.Context, input ExtractAudioInput) (*ExtractAudioOutput, error) {
+	log := activity.GetLogger(ctx)
+	activity.RecordHeartbeat(ctx, "ExtractAudio")
+	log.Info("Starting ExtractAudioActivity")
+
+	availableChannels := map[int]ffmpeg.FFProbeStream{}
+
+	analyzed, err := AnalyzeFile(ctx, AnalyzeFileParams{
+		FilePath: input.VideoPath,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	for _, stream := range analyzed.AudioStreams {
+		availableChannels[stream.Index] = stream
+	}
+
+	if len(input.Channels) == 0 {
+		input.Channels = lo.Keys(availableChannels)
+	}
+
+	extractedChannels := map[int]paths.Path{}
+
+	for _, channel := range input.Channels {
+		if _, ok := availableChannels[channel]; !ok {
+			return nil, errors.Errorf("Channel %d not found in video", channel)
+		}
+
+		extractedChannels[channel] = input.OutputFolder.Append(fmt.Sprintf(input.FileNamePattern, channel))
+	}
+
+	stopChan, progressCallback := registerProgressCallback(ctx)
+	defer close(stopChan)
+
+	_, err = transcode.ExtractAudioChannels(input.VideoPath, extractedChannels, progressCallback)
+
+	return &ExtractAudioOutput{
+		AudioFiles: extractedChannels,
+	}, err
+
+}

activities/queues.go

@@ -27,6 +27,7 @@ func GetAudioTranscodeActivities() []any {
 		PrependSilence,
 		DetectSilence,
 		AdjustAudioToVideoStart,
+		ExtractAudio,
 	}
 }
 
@@ -46,6 +47,7 @@ func GetVideoTranscodeActivities() []any {
 		TranscodeMuxToSimpleMXF,
 		ExecuteFFmpeg,
 		MultitrackMux,
+		GetVideoOffset,
 	}
 }
 

cmd/trigger_ui/ingest-fix.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	ingestworkflows "github.com/bcc-code/bcc-media-flows/workflows/ingest"
+	"github.com/gin-gonic/gin"
+	"github.com/gin-gonic/gin/binding"
+	"github.com/google/uuid"
+	"go.temporal.io/sdk/client"
+	"net/http"
+)
+
+func (s *TriggerServer) ingestFixGET(c *gin.Context) {
+	c.HTML(http.StatusOK, "ingest-fix.gohtml", nil)
+}
+
+type mu1mu2ExtractForm struct {
+	VX1ID string `form:"vx1" binding:"required"`
+	VX2ID string `form:"vx2" binding:"required"`
+}
+
+func (s *TriggerServer) mu1mu2ExtractPOST(c *gin.Context) {
+	var form mu1mu2ExtractForm
+	err := c.ShouldBindWith(&form, binding.Form)
+	if err != nil {
+		renderErrorPage(c, http.StatusInternalServerError, err)
+		return
+	}
+
+	queue := getQueue()
+	workflowOptions := client.StartWorkflowOptions{
+		ID:        uuid.NewString(),
+		TaskQueue: queue,
+	}
+
+	_, err = s.wfClient.ExecuteWorkflow(c, workflowOptions, ingestworkflows.ExtractAudioFromMU1MU2, ingestworkflows.ExtractAudioFromMU1MU2Input{
+		MU1ID: form.VX1ID,
+		MU2ID: form.VX2ID,
+	})
+}

cmd/trigger_ui/main.go

@@ -347,6 +347,10 @@ func main() {
 		GET("/admin", server.uploadMasterAdminGET).
 		POST("/admin", server.uploadMasterAdminPOST)
 
+	router.Group("/ingest-fix").
+		GET("/", server.ingestFixGET).
+		POST("/ingest-fix/mu1mu2extract", server.mu1mu2ExtractPOST)
+
 	port := os.Getenv("PORT")
 	if port == "" {
 		port = "8083"

cmd/trigger_ui/templates/ingest-fix.gohtml

@@ -0,0 +1,34 @@
+<!DOCTYPE html>
+<html lang="en">
+
+<head>
+    <script src="https://cdn.tailwindcss.com?plugins=forms,typography"></script>
+</head>
+
+<body class="">
+<form method="POST" action="/ingest-fix/ingest-fix/mu1mu2extract" class="flex flex-col min-h-screen gap-2 mx-auto p-14 max-w-screen-lg">
+    <div class="max-w-none flex flex-col gap-2">
+        <h2 class="text-2xl font-bold">Extract Audio from MU1 and MU2</h2>
+    </div>
+
+    <div class="flex flex-col gap-2 w-full mt-8 w-20">
+        <div class="flex flex-col">
+            <label for="filename">MU1 VX-ID:</label>
+            <input id="filename" class="border" type="text" name="vx1" pattern="VX-[0-9]+" required>
+        </div>
+
+        <div class="flex flex-col">
+            <label for="filename">MU2 VX-ID:</label>
+            <input id="filename" class="border" type="text" name="vx2" pattern="VX-[0-9]+" required>
+        </div>
+
+        <input id="submit"
+               class="cursor-pointer rounded-md bg-[#6A64F1] py-3 px-8 text-center text-base font-semibold text-white outline-none"
+               type="submit" value="Start">
+    </div>
+</form>
+
+<script></script>
+</body>
+
+</html>

cmd/worker/main.go

@@ -91,6 +91,7 @@ var workerWorkflows = []any{
 	ingestworkflows.Incremental,
 	ingestworkflows.MoveUploadedFiles,
 	ingestworkflows.ImportAudioFileFromReaper,
+	ingestworkflows.ExtractAudioFromMU1MU2,
 	workflows.NormalizeAudioLevelWorkflow,
 	vb_export.VBExport,
 	vb_export.VBExportToAbekas,

services/ffmpeg/progress.go

@@ -26,6 +26,10 @@ type StreamInfo struct {
 	HasAudio     bool
 	HasVideo     bool
 	HasAlpha     bool
+	VideoStreams []FFProbeStream
+	AudioStreams []FFProbeStream
+	SubSteams    []FFProbeStream
+	OtherStreams []FFProbeStream
 	Progressive  bool
 	TotalFrames  int
 	TotalSeconds float64
@@ -47,6 +51,20 @@ func ProbeResultToInfo(info *FFProbeResult) StreamInfo {
 		}),
 	}
 
+	for _, stream := range info.Streams {
+		switch stream.CodecType {
+		case "audio":
+			streamInfo.AudioStreams = append(streamInfo.AudioStreams, stream)
+		case "video":
+			streamInfo.VideoStreams = append(streamInfo.VideoStreams, stream)
+		case "subtitle":
+			streamInfo.SubSteams = append(streamInfo.SubSteams, stream)
+		default:
+			streamInfo.OtherStreams = append(streamInfo.OtherStreams, stream)
+		}
+
+	}
+
 	stream, found := lo.Find(info.Streams, func(i FFProbeStream) bool {
 		return i.CodecType == "video"
 	})

services/transcode/audio.go

@@ -312,3 +312,25 @@ func SplitAudioChannels(filePath, outputDir paths.Path, cb ffmpeg.ProgressCallba
 	}
 	return files, nil
 }
+
+func ExtractAudioChannels(filePath paths.Path, output map[int]paths.Path, cb ffmpeg.ProgressCallback) (map[int]paths.Path, error) {
+	if len(output) == 0 {
+		return nil, fmt.Errorf("no output channels specified")
+	}
+
+	params := []string{
+		"-i", filePath.Local(),
+	}
+
+	out := make(map[int]paths.Path)
+	for channel, file := range output {
+		params = append(params, "-map", fmt.Sprintf("0:%d", channel), "-c", "copy", file.Local())
+	}
+
+	_, err := ffmpeg.Do(params, ffmpeg.StreamInfo{}, cb)
+	if err != nil {
+		return nil, err
+	}
+
+	return out, nil
+}