Add capture-pipeline util from can-i-trust-you

Signed-off-by: Paul Kocialkowski <contact@paulk.fr>

1 files changed, 523 insertions, 0 deletions

diff --git a/capture-pipeline b/capture-pipeline
new file mode 100755
index 0000000..85e663f
--- /dev/null
+++ b/capture-pipeline
@@ -0,0 +1,523 @@
+#!/usr/bin/python
+
+import sys
+import os
+import getopt
+import datetime
+import subprocess
+import yaml
+
+# settings
+
+settings = {}
+
+def settings_entry(shot, entry):
+	global settings
+
+	if entry in shot:
+		return shot[entry]
+	elif entry in settings:
+		return settings[entry]
+	else:
+		return None
+
+# timecode
+
+def timecode_seconds(timecode):
+	values = timecode.split(':')
+	seconds = 0.
+
+	for i in range(len(values)):
+		index = len(values) - i - 1
+		seconds += float(values[i]) * pow(60, index)
+
+	return seconds
+
+def timecode_string(seconds):
+	remainder = seconds
+	string = ""
+	pow_max = 3
+
+	for i in range(pow_max):
+		index = pow_max - i - 1
+		if remainder / pow(60, index) >= 1.:
+			if index == 0:
+				if remainder < 10.:
+					string += "0"
+
+				string += "%.3f" % remainder
+			else:
+				part = int(remainder / pow(60, index))
+				remainder -= float(part * pow(60, index))
+
+				string += "%02d:" % part
+		else:
+			if index == 0:
+				string += "00.0"
+			else:
+				string += "00:"
+
+	return string
+
+# audio
+
+def audio_suffix():
+	return "." + settings["audio-extension"]
+
+def audio_source_base(sequence, shot):
+	if "audio-source" in shot:
+		prefix = shot["audio-source"]
+	else:
+		prefix = settings["audio-source"]
+
+	return os.path.join(prefix, sequence["label"])
+
+def audio_source_name(sequence, shot, take):
+	base = audio_source_base(sequence, shot)
+	suffix = audio_suffix()
+	name = shot_name(shot)
+	index = 1
+
+	path = os.path.join(base, name + suffix)
+	if os.path.isfile(path):
+		return name
+
+	if "index" in take:
+		index = take["index"]
+
+	name += "-" + str(index)
+	path = os.path.join(base, name + suffix)
+	if os.path.isfile(path):
+		return name
+
+	return None
+
+def audio_source_path(sequence, shot, take):
+	base = audio_source_base(sequence, shot)
+	name = audio_source_name(sequence, shot, take)
+	suffix = audio_suffix()
+
+	if name:
+		return os.path.join(base, name + suffix)
+
+	return None
+
+def audio_output_name(sequence, shot, take, part = None):
+	base = os.path.join(settings["audio-output"], sequence["label"])
+	label = shot_label(shot)
+	prefix = shot_prefix(shot)
+	suffix = audio_suffix()
+	name = ""
+
+	if part and "actions" in part:
+		actions = part["actions"]
+
+		if len(actions) > 4:
+			actions = actions[:2] + actions[-2:]
+
+		for action in actions:
+			name += action + "-"
+
+		# remove separator
+		name = name[:-1]
+	elif label:
+		name = label
+	else:
+		return None
+
+	if prefix:
+		name += "-" + prefix
+
+	return name
+
+def audio_output_step(sequence, shot, take, part = None):
+	name = audio_output_name(sequence, shot, take, part)
+
+	if name:
+		if name in sequence["audio-registry"]:
+			if sequence["audio-registry"][name]["count"] > 1:
+				sequence["audio-registry"][name]["index"] += 1
+
+def audio_output_path(sequence, shot, take, part = None):
+	base = os.path.join(settings["audio-output"], sequence["label"])
+	name = audio_output_name(sequence, shot, take, part)
+	suffix = audio_suffix()
+
+	if name:
+		if name in sequence["audio-registry"]:
+			if sequence["audio-registry"][name]["count"] > 1:
+				name += "-" + str(sequence["audio-registry"][name]["index"])
+
+		return os.path.join(base, name + suffix)
+
+	return None
+
+def audio_output_prepare(sequence):
+	base = os.path.join(settings["audio-output"], sequence["label"])
+
+	try:
+		os.stat(base)
+	except:
+		os.makedirs(base, exist_ok = True)
+
+def audio_ffmpeg_command(sequence, shot, take, part):
+	source_path = audio_source_path(sequence, shot, take)
+	output_path = audio_output_path(sequence, shot, take, part)
+
+	ffmpeg_command = [ "ffmpeg", "-i", source_path ]
+
+	encoder = settings_entry(shot, "audio-encoder")
+	if encoder:
+		ffmpeg_command += [ "-c:a", encoder ]
+
+	filters = settings_entry(shot, "audio-filters")
+	if filters:
+		ffmpeg_command += [ "-filter:a", ",".join(filters) ]
+
+	if "audio-start" in part:
+		ffmpeg_command += [ "-ss", part["audio-start"] ]
+	elif "video-start" in part:
+		video_start_seconds = timecode_seconds(part["video-start"])
+		video_sync_seconds = timecode_seconds(take["video-sync"])
+		audio_sync_seconds = timecode_seconds(take["audio-sync"])
+		audio_start_seconds = audio_sync_seconds + video_start_seconds - video_sync_seconds
+
+		ffmpeg_command += [ "-ss", timecode_string(audio_start_seconds) ]
+
+	if "audio-stop" in part:
+		ffmpeg_command += [ "-to", part["audio-stop"] ]
+	elif "video-stop" in part:
+		video_stop_seconds = timecode_seconds(part["video-stop"])
+		video_sync_seconds = timecode_seconds(take["video-sync"])
+		audio_sync_seconds = timecode_seconds(take["audio-sync"])
+		audio_stop_seconds = audio_sync_seconds + video_stop_seconds - video_sync_seconds
+
+		ffmpeg_command += [ "-to", timecode_string(audio_stop_seconds) ]
+
+	ffmpeg_command += [ output_path ]
+
+	return ffmpeg_command
+
+def audio_register(sequence, shot, take, part):
+	if "audio-registry" not in sequence:
+		sequence["audio-registry"] = {}
+
+	name = audio_output_name(sequence, shot, take, part)
+	if name in sequence["audio-registry"]:
+		sequence["audio-registry"][name]["count"] += 1
+	else:
+		sequence["audio-registry"][name] = { "count": 1 , "index": 1}
+
+def audio_process(sequence, shot, take, part):
+	source_path = audio_source_path(sequence, shot, take)
+	output_path = audio_output_path(sequence, shot, take, part)
+
+	if not source_path or not output_path:
+		return
+
+	command = audio_ffmpeg_command(sequence, shot, take, part)
+
+	audio_output_prepare(sequence)
+	audio_output_step(sequence, shot, take, part)
+
+	print(" ".join(command))
+
+	if os.path.isfile(output_path):
+		return
+
+	subprocess.call(command)
+
+# video
+
+def video_suffix(shot):
+	return "." + settings_entry(shot, "video-extension")
+
+def video_source_base(sequence, shot):
+	return os.path.join(settings_entry(shot, "video-source"), sequence["label"])
+
+def video_source_name(sequence, shot, take):
+	base = video_source_base(sequence, shot)
+	suffix = video_suffix(shot)
+	name = shot_name(shot)
+	index = 1
+
+	path = os.path.join(base, name + suffix)
+	if os.path.isfile(path):
+		return name
+
+	if "index" in take:
+		index = take["index"]
+
+	name += "-" + str(index)
+	path = os.path.join(base, name + suffix)
+	if os.path.isfile(path):
+		return name
+
+	return None
+
+def video_source_path(sequence, shot, take):
+	base = video_source_base(sequence, shot)
+	name = video_source_name(sequence, shot, take)
+	suffix = video_suffix(shot)
+
+	if name:
+		return os.path.join(base, name + suffix)
+
+	return None
+
+def video_output_name(sequence, shot, take, part = None):
+	base = os.path.join(settings["video-output"], sequence["label"])
+	label = shot_label(shot)
+	prefix = shot_prefix(shot)
+	suffix = video_suffix(shot)
+	name = ""
+
+	if part and "actions" in part:
+		actions = part["actions"]
+
+		if len(actions) > 4:
+			actions = actions[:2] + actions[-2:]
+
+		for action in actions:
+			name += action + "-"
+
+		# remove separator
+		name = name[:-1]
+	elif label:
+		name = label
+	else:
+		return None
+
+	if prefix:
+		name += "-" + prefix
+
+	return name
+
+def video_output_step(sequence, shot, take, part = None):
+	name = video_output_name(sequence, shot, take, part)
+
+	if name:
+		if name in sequence["video-registry"]:
+			if sequence["video-registry"][name]["count"] > 1:
+				sequence["video-registry"][name]["index"] += 1
+
+def video_output_path(sequence, shot, take, part = None):
+	base = os.path.join(settings["video-output"], sequence["label"])
+	name = video_output_name(sequence, shot, take, part)
+	suffix = video_suffix(shot)
+
+	if name:
+		if name in sequence["video-registry"]:
+			if sequence["video-registry"][name]["count"] > 1:
+				name += "-" + str(sequence["video-registry"][name]["index"])
+
+		return os.path.join(base, name + suffix)
+
+	return None
+
+def video_output_prepare(sequence):
+	base = os.path.join(settings["video-output"], sequence["label"])
+
+	try:
+		os.stat(base)
+	except:
+		os.makedirs(base, exist_ok = True)
+
+def video_ffmpeg_command(sequence, shot, take, part):
+	source_path = video_source_path(sequence, shot, take)
+	output_path = video_output_path(sequence, shot, take, part)
+
+	ffmpeg_command = [ "ffmpeg", "-i", source_path ]
+
+	width = settings_entry(shot, "video-width")
+	height = settings_entry(shot, "video-height")
+
+	if width and height:
+		ffmpeg_command += [ "-s", str(width)+"x"+str(height) ]
+
+	frame_rate = settings_entry(shot, "video-frame-rate")
+	if frame_rate:
+		ffmpeg_command += [ "-r", str(frame_rate) ]
+
+	encoder = settings_entry(shot, "video-encoder")
+	if encoder:
+		ffmpeg_command += [ "-c:v", encoder ]
+
+	crf = settings_entry(shot, "video-crf")
+	if crf:
+		ffmpeg_command += [ "-crf", str(crf) ]
+
+	quality = settings_entry(shot, "video-quality")
+	if quality:
+		ffmpeg_command += [ "-q:v", str(quality) ]
+
+	rate_max = settings_entry(shot, "video-rate-max")
+	if rate_max:
+		ffmpeg_command += [ "-maxrate", rate_max ]
+		ffmpeg_command += [ "-bufsize", "2M" ]
+
+	filters = settings_entry(shot, "video-filters")
+	if filters:
+		ffmpeg_command += [ "-filter:v", ",".join(filters) ]
+
+	audio = settings_entry(shot, "video-audio")
+	if audio == "remove":
+		ffmpeg_command += [ "-an" ]
+
+	if "video-start" in part:
+		ffmpeg_command += [ "-ss", part["video-start"] ]
+
+	if "video-stop" in part:
+		ffmpeg_command += [ "-to", part["video-stop"] ]
+
+	ffmpeg_command += [ output_path ]
+
+	return ffmpeg_command
+
+def video_register(sequence, shot, take, part):
+	if "video-registry" not in sequence:
+		sequence["video-registry"] = {}
+
+	name = video_output_name(sequence, shot, take, part)
+	if name in sequence["video-registry"]:
+		sequence["video-registry"][name]["count"] += 1
+	else:
+		sequence["video-registry"][name] = { "count": 1 , "index": 1}
+
+def video_process(sequence, shot, take, part):
+	source_path = video_source_path(sequence, shot, take)
+	output_path = video_output_path(sequence, shot, take, part)
+
+	if not source_path or not output_path:
+		return
+
+	command = video_ffmpeg_command(sequence, shot, take, part)
+
+	video_output_prepare(sequence)
+	video_output_step(sequence, shot, take, part)
+
+	print(" ".join(command))
+
+	if os.path.isfile(output_path):
+		return
+
+	subprocess.call(command)
+
+# media
+
+def media_register(sequence, shot, take, part = None):
+	if not part:
+		part = {}
+
+	audio_register(sequence, shot, take, part)
+	video_register(sequence, shot, take, part)
+
+def media_process(sequence, shot, take, part = None):
+	if not part:
+		part = {}
+
+	audio_process(sequence, shot, take, part)
+	video_process(sequence, shot, take, part)
+
+# take
+
+def take_register(sequence, shot, take):
+	if "parts" in take:
+		for part in take["parts"]:
+			media_register(sequence, shot, take, part)
+	else:
+		media_register(sequence, shot, take)
+
+def take_process(sequence, shot, take):
+	print("  - source: " + video_source_name(sequence, shot, take))
+
+	if "parts" in take:
+		for part in take["parts"]:
+			print("  - output: " + video_output_name(sequence, shot, take, part))
+			media_process(sequence, shot, take, part)
+	else:
+		print("  - output: " + video_output_name(sequence, shot, take))
+		media_process(sequence, shot, take)
+
+# shot
+
+def shot_prefix(shot):
+	label = ""
+	separator = ""
+
+	if "value" in shot:
+		label += separator + shot["value"]
+		separator = "-"
+
+	if "position" in shot:
+		label += separator + shot["position"]
+		separator = "-"
+
+	return label
+
+def shot_label(shot):
+	if "label" in shot:
+		return shot["label"]
+
+	return None
+
+def shot_name(shot):
+	prefix = shot_prefix(shot)
+	label = shot_label(shot)
+
+	if label:
+		return prefix + "-" + label
+	else:
+		return prefix
+
+def shot_process(sequence, shot):
+	print("- shot: " + shot_name(shot))
+
+	if "takes" in shot:
+		for take in shot["takes"]:
+			take_process(sequence, shot, take)
+	else:
+		take = {}
+		take_process(sequence, shot, take)
+
+def shot_register(sequence, shot):
+	if "takes" in shot:
+		for take in shot["takes"]:
+			take_register(sequence, shot, take)
+	else:
+		take = {}
+		take_register(sequence, shot, take)
+
+def sequence_process(sequence):
+	label = sequence["label"]
+
+	print("# sequence: " + label + "\n")
+
+	shots = sequence["shots"]
+
+	for shot in shots:
+		shot_register(sequence, shot)
+
+	for shot in shots:
+		shot_process(sequence, shot)
+
+	print("")
+
+def main():
+	global settings
+
+	if len(sys.argv) < 2:
+		return 1
+
+	path = sys.argv[1]
+
+	stream = open(path, 'r')
+	data = yaml.load(stream, Loader = yaml.SafeLoader)
+
+	settings = data["settings"]
+
+	for sequence in data["sequences"]:
+		sequence_process(sequence)
+
+if __name__ == "__main__":
+	main()