We have been using FFmpeg to convert image sequences to ProRes files.
We are writing a custom script where N machines will render a section of the image sequence into a smaller movie file. Then, the final machine will concatenate all movies into the single ProRes file.
For example, a 1000 frame sequence rendered by 5 machines - each machine renders 200 frame ProRes files. Then, one machine quickly concats 5 movies into the final 1000-frame ProRes.
What we were hoping is that user would submit the job with N queued tasks + 1 pending task. Deadline would create N tasks which immediately render + 1 pending task.
When all N previous tasks complete, the N+1 pending task immediately releases and runs the concatenation.
The problem is that the final task correctly submits as pending. However, it will not release the others finish.
What I was hoping is that we could set the Job to be Dependent on ITSELF, and that as soon as each chunk completes, it would Release any pending tasks and complete.
Is this even possible? Or is this only possible with 2 jobs, where 1 job contains the tasks and 1 job runs the concat?
Current FFmpeg.py below
#!/usr/bin/env python3
from __future__ import absolute_import
import re
from System import *
from System.Diagnostics import *
from System.IO import *
from Deadline.Plugins import DeadlinePlugin, PluginType
from Deadline.Scripting import *
from six.moves import range
def GetDeadlinePlugin():
return FFmpegPlugin()
def CleanupDeadlinePlugin( deadlinePlugin ):
deadlinePlugin.Cleanup()
class FFmpegPlugin(DeadlinePlugin):
def __init__( self ):
self.InitializeProcessCallback += self.InitializeProcess
self.RenderExecutableCallback += self.RenderExecutable
self.RenderArgumentCallback += self.RenderArgument
self.PreRenderTasksCallback += self.PreRenderTasks
self.PostRenderTasksCallback += self.PostRenderTasks
def Cleanup(self):
for stdoutHandler in self.StdoutHandlers:
del stdoutHandler.HandleCallback
del self.InitializeProcessCallback
del self.RenderExecutableCallback
del self.RenderArgumentCallback
del self.PreRenderTasksCallback
del self.PostRenderTasksCallback
def InitializeProcess(self):
self.SingleFramesOnly=False
self.StdoutHandling=True
self.ProcessPriority = ProcessPriorityClass.RealTime
self.AddStdoutHandlerCallback(".*Error.*").HandleCallback += self.HandleStdoutError
# Frame-based progress: FFmpeg emits "frame=N" to stdout (zero-indexed from start of encode)
self.AddStdoutHandlerCallback( r"\bframe=\s*(\d+)\b" ).HandleCallback += self.HandleStdoutProgress
def RenderExecutable(self):
FFmpegExeList = self.GetConfigEntry("FFmpeg_RenderExecutable")
FFmpegExe = FileUtils.SearchFileList( FFmpegExeList )
if( FFmpegExe == "" ):
self.FailRender( "No file found in the semicolon separated list \"" + FFmpegExeList+ "\". The path to the render executable can be configured from the Plugin Configuration in the Deadline Monitor." )
return FFmpegExe
def RenderArgument(self):
# ── Concat task ─────────────────────────────────────────────────
concatTaskStr = self.GetPluginInfoEntryWithDefault("ConcatTask", "")
taskId = str(self.GetCurrentTaskId())
if concatTaskStr != "" and taskId == concatTaskStr:
return self.BuildConcatArgument()
# ── Chunk mode ──────────────────────────────────────────────────
chunkStart = self.GetPluginInfoEntryWithDefault("ChunkStartFrame_" + taskId, "")
chunkCount = self.GetPluginInfoEntryWithDefault("ChunkFrameCount_" + taskId, "")
chunkOutput = self.GetPluginInfoEntryWithDefault("ChunkOutputFile_" + taskId, "")
isChunk = (chunkStart != "" and chunkCount != "" and chunkOutput != "")
if isChunk:
self.LogInfo("Chunk mode: task %s startFrame=%s frames=%s" % (taskId, chunkStart, chunkCount))
self.LogInfo("Chunk output: %s" % chunkOutput)
outputFile = self.GetPluginInfoEntryWithDefault( "OutputFile", "" )
if isChunk:
outputFile = chunkOutput
outputFile = RepositoryUtils.CheckPathMapping( outputFile )
outputFile = self.ProcessPath( outputFile )
outputArgs = self.GetPluginInfoEntryWithDefault( "OutputArgs", "" )
if isChunk:
outputArgs = "-frames:v " + chunkCount + " " + outputArgs
additionalArgs = self.GetPluginInfoEntryWithDefault( "AdditionalArgs", "" )
useSameArgs = self.GetBooleanPluginInfoEntryWithDefault( "UseSameInputArgs", False )
videoPreset = self.GetPluginInfoEntryWithDefault( "VideoPreset", "" )
videoPreset = RepositoryUtils.CheckPathMapping( videoPreset )
videoPreset = self.ProcessPath( videoPreset )
audioPreset = self.GetPluginInfoEntryWithDefault( "AudioPreset", "" )
audioPreset = RepositoryUtils.CheckPathMapping( audioPreset )
audioPreset = self.ProcessPath( audioPreset )
subtitlePreset = self.GetPluginInfoEntryWithDefault( "SubtitlePreset", "" )
subtitlePreset = RepositoryUtils.CheckPathMapping( subtitlePreset )
subtitlePreset = self.ProcessPath( subtitlePreset )
if useSameArgs:
inputArgs0 = self.GetPluginInfoEntryWithDefault( "InputArgs0", "" )
if( outputFile == "" ):
self.FailRender( "No output file was specified." )
renderArgument = ""
if useSameArgs:
self.LogInfo( "UseSameInputArgs = True" )
else:
self.LogInfo( "UseSameInputArgs = False" )
for i in range(0,9):
inputFile = self.GetPluginInfoEntryWithDefault( "InputFile%d" % i, "" )
inputArgs = self.GetPluginInfoEntryWithDefault( "InputArgs%d" % i, "" )
replacePadding = self.GetBooleanPluginInfoEntryWithDefault( "ReplacePadding%d" % i, True )
if inputFile != "":
inputFile = RepositoryUtils.CheckPathMapping( inputFile )
inputFile = self.ProcessPath( inputFile )
if replacePadding:
currPadding = FrameUtils.GetFrameStringFromFilename( inputFile )
paddingSize = len( currPadding )
if '-' in currPadding:
front = "-%"
paddingSize = paddingSize - 1
else:
front = "%"
if paddingSize > 0:
padding = front + StringUtils.ToZeroPaddedString( paddingSize, 2, False ) + "d"
inputFile = FrameUtils.SubstituteFrameNumber( inputFile, padding )
if (useSameArgs and inputArgs0 != ""):
args = inputArgs0
if isChunk and i == 0:
args = re.sub(r'-start_number\s+\d+', '-start_number ' + chunkStart, args)
renderArgument += "%s " % args
elif (not useSameArgs and inputArgs != ""):
args = inputArgs
if isChunk and i == 0:
args = re.sub(r'-start_number\s+\d+', '-start_number ' + chunkStart, args)
renderArgument += "%s " % args
renderArgument += "-i \"%s\" " % inputFile
if outputArgs != "":
renderArgument += "%s " % outputArgs
renderArgument += "-y \"%s\"" % outputFile
if additionalArgs != "":
renderArgument += " %s" % additionalArgs
if videoPreset != "":
renderArgument += " -vpre \"%s\"" % videoPreset
if audioPreset != "":
renderArgument += " -apre \"%s\"" % audioPreset
if subtitlePreset != "":
renderArgument += " -spre \"%s\"" % subtitlePreset
return renderArgument
def BuildConcatArgument(self):
"""Build FFmpeg concat demuxer arguments. List file written by PreRenderTasks."""
outputFile = self.GetPluginInfoEntryWithDefault("OutputFile", "")
outputFile = RepositoryUtils.CheckPathMapping(outputFile)
outputFile = self.ProcessPath(outputFile)
if not outputFile:
self.FailRender("OutputFile not specified for concat task.")
concatListFile = Path.ChangeExtension(outputFile, ".concat.txt")
self.LogInfo("Concat mode: list=%s output=%s" % (concatListFile, outputFile))
return '-f concat -safe 0 -i "%s" -c copy -y "%s"' % (concatListFile, outputFile)
def ProcessPath( self, filepath ):
if SystemUtils.IsRunningOnWindows():
filepath = filepath.replace("/","\\")
if filepath.startswith( "\\" ) and not filepath.startswith( "\\\\" ):
filepath = "\\" + filepath
else:
filepath = filepath.replace("\\","/")
return filepath
def PreRenderTasks(self):
self.LogInfo( "FFmpeg job starting..." )
concatTaskStr = self.GetPluginInfoEntryWithDefault("ConcatTask", "")
taskId = str(self.GetCurrentTaskId())
# ── Chunk task: ensure _temp/ directory exists ──────────────────
if concatTaskStr == "" or taskId != concatTaskStr:
chunkOutput = self.GetPluginInfoEntryWithDefault("ChunkOutputFile_" + taskId, "")
if chunkOutput:
chunkOutput = RepositoryUtils.CheckPathMapping(chunkOutput)
chunkOutput = self.ProcessPath(chunkOutput)
outDir = Path.GetDirectoryName(chunkOutput)
if outDir and not Directory.Exists(outDir):
self.LogInfo("Creating output directory: %s" % outDir)
Directory.CreateDirectory(outDir)
# ── Concat task: generate list file from PluginInfo keys ────────
if concatTaskStr != "" and taskId == concatTaskStr:
chunkCount = int(self.GetPluginInfoEntryWithDefault("ConcatChunkCount", "0"))
if chunkCount == 0:
self.FailRender("ConcatChunkCount not set for concat task.")
chunkFiles = []
for i in range(chunkCount):
cf = self.GetPluginInfoEntryWithDefault("ChunkOutputFile_%d" % i, "")
cf = RepositoryUtils.CheckPathMapping(cf)
cf = self.ProcessPath(cf)
if not cf:
self.FailRender("ChunkOutputFile_%d not found in PluginInfo." % i)
chunkFiles.append(cf)
missing = [f for f in chunkFiles if not File.Exists(f)]
if missing:
self.FailRender("Chunk files missing: %s" % str(missing))
outputFile = self.GetPluginInfoEntryWithDefault("OutputFile", "")
outputFile = RepositoryUtils.CheckPathMapping(outputFile)
outputFile = self.ProcessPath(outputFile)
concatListFile = Path.ChangeExtension(outputFile, ".concat.txt")
self.LogInfo("Writing concat list: %s" % concatListFile)
with open(concatListFile, "w") as f:
for cf in chunkFiles:
f.write("file '%s'\n" % cf)
self.LogInfo("Concat list written with %d entries." % chunkCount)
def PostRenderTasks(self):
self.LogInfo( "FFmpeg job finished." )
concatTaskStr = self.GetPluginInfoEntryWithDefault("ConcatTask", "")
taskId = str(self.GetCurrentTaskId())
# ── Concat task completed: clean up temp files ────────────────
if concatTaskStr != "" and taskId == concatTaskStr:
chunkCount = int(self.GetPluginInfoEntryWithDefault("ConcatChunkCount", "0"))
if chunkCount > 0:
cf = self.GetPluginInfoEntryWithDefault("ChunkOutputFile_0", "")
cf = RepositoryUtils.CheckPathMapping(cf)
cf = self.ProcessPath(cf)
if cf:
tempDir = Path.GetDirectoryName(cf)
if tempDir and Directory.Exists(tempDir):
Directory.Delete(tempDir, True)
self.LogInfo("Cleanup: deleted temp folder %s" % tempDir)
outputFile = self.GetPluginInfoEntryWithDefault("OutputFile", "")
outputFile = RepositoryUtils.CheckPathMapping(outputFile)
outputFile = self.ProcessPath(outputFile)
concatListFile = Path.ChangeExtension(outputFile, ".concat.txt")
if File.Exists(concatListFile):
File.Delete(concatListFile)
self.LogInfo("Cleanup: deleted concat list %s" % concatListFile)
return
# ── Chunk task completed: release concat when all chunks done ──
if concatTaskStr == "":
return # Not a chunked job
concatTaskIndex = int(concatTaskStr)
chunkCount = int(self.GetPluginInfoEntryWithDefault("ConcatChunkCount", "0"))
if chunkCount == 0:
return
job = self.GetJob()
currentTaskId = int(self.GetCurrentTaskId())
tasks = RepositoryUtils.GetJobTasks(job, True)
concatTaskObj = None
completedChunks = 0
for task in tasks:
tid = int(task.TaskId)
if tid == concatTaskIndex:
concatTaskObj = task
if tid < chunkCount:
# Current task may not be marked Completed in repo yet,
# but we know it succeeded (we're in PostRenderTasks)
if tid == currentTaskId or str(task.TaskStatus) == "Completed":
completedChunks += 1
if concatTaskObj is None:
self.LogWarning("Could not find concat task %d" % concatTaskIndex)
return
self.LogInfo("Chunk progress: %d/%d complete" % (completedChunks, chunkCount))
# All chunks done: release the concat task (pended at submit time)
if completedChunks >= chunkCount:
self.LogInfo("All %d chunks complete -- releasing concat task %d" % (chunkCount, concatTaskIndex))
RepositoryUtils.ReleasePendingTasks(job, [concatTaskObj])
def HandleStdoutError(self):
self.FailRender( self.GetRegexMatch(0) )
def HandleStdoutProgress(self):
taskId = str(self.GetCurrentTaskId())
chunkCount = self.GetPluginInfoEntryWithDefault("ChunkFrameCount_" + taskId, "")
if chunkCount:
try:
frameCount = int(chunkCount)
currFrame = int(self.GetRegexMatch(1))
self.SetProgress( min(100, 100 * currFrame // frameCount) )
except (ValueError, IndexError):
pass
return
rangeStr = self.GetJob().JobExtraInfo0
if not rangeStr or "-" not in rangeStr:
return
try:
parts = rangeStr.split("-")
startFrame = int(parts[0])
endFrame = int(parts[1])
frameCount = abs(endFrame - startFrame) + 1
currFrame = int(self.GetRegexMatch(1))
self.SetProgress( min(100, 100 * currFrame // frameCount) )
except (ValueError, IndexError):
pass