- """Classes for producing videos"""
-
- from abc import ABC
- import json
- import logging
- import os
- import subprocess
- import tempfile
-
- # for visualisations:
- import matplotlib.pyplot as plt
-
- # for encoding as JSON
- from pipeline.utils import Feature
-
-
- class Producer(ABC):
- """Generic producer interface."""
- def __init__(self, features):
- """All producers should take a list of features as input"""
- def produce(self):
- """All Producers should produce something!"""
-
- class VideoProducer(Producer):
- """Video producer interface."""
-
- class FfmpegVideoProducer(VideoProducer):
- """Produce videos using ffmpeg"""
- _CONFIG_DEFAULT_OUTPUT_DIR = "/tmp/"
- _CONFIG_DEFAULT_OUTPUT_FILENAME = "highlights.mp4"
- _CONFIG_COMPILE_CLIPS = True
-
- def __init__(self, features, compile_clips=_CONFIG_COMPILE_CLIPS,
- output_dir=_CONFIG_DEFAULT_OUTPUT_DIR,
- output_filename=_CONFIG_DEFAULT_OUTPUT_FILENAME,
- ) -> None:
- if not features:
- raise ValueError("No features provided")
- # TODO: consider if we want to permit empty features (producing no video)
- self.features = features
- self._compile_clips = compile_clips
- self._output_dir = output_dir
- self._output_filename = output_filename
-
- def _run_no_output(self, cmd: list, cwd:str=".") -> None:
- """Run a command and return the output as a string
-
- Defined to be mocked out in tests via unittest.mock.patch
- """
- subprocess.run(cmd, stdout=None, stderr=None, cwd=cwd)
-
- def _ffmpeg_feature_to_clip(self, feature=None, output_filepath=None):
- """use ffmpeg to produve a video clip from a feature"""
- OVERWRITE = True # TODO: consider making this a config option
- if not feature or not feature.interval:
- raise ValueError("No feature provided")
-
- if not output_filepath:
- raise ValueError("No output filepath provided")
-
- ffmpeg_prefix = ["ffmpeg", "-y"] if OVERWRITE else ["ffmpeg"]
- ffmpeg_suffix = ["-r", "60", "-c:v", "libx264", "-crf", "26", "-c:a", "aac", "-preset", "ultrafast"]
- # TODO: match framerate of input video
- # TODO: adjustable encoding options
- seek = ["-ss", str(feature.interval.start)]
- duration = ["-t", str(feature.interval.duration)]
- ffmpeg_args = ffmpeg_prefix + seek + ["-i"] + [feature.source.path] +\
- duration + ffmpeg_suffix + [output_filepath]
- logging.info(f"ffmpeg_args: {ffmpeg_args}")
- self._run_no_output(ffmpeg_args)
-
- def _ffmpeg_concat_clips(self, clips=None, output_filepath=None):
- """use ffmpeg to concatenate clips into a single video"""
- OVERWRITE = True
- ffmpeg_prefix = ["ffmpeg"]
- ffmpeg_prefix += ["-y"] if OVERWRITE else []
- ffmpeg_prefix += ["-f", "concat", "-safe", "0", "-i"]
-
- # there is a method to do this via process substitution, but it's not portable
- # so we'll use the input file list method
-
- if not clips:
- raise ValueError("No clips provided")
-
- if not output_filepath:
- raise ValueError("No output filepath provided")
-
- # generate a temporary file with the list of clips
- join_file = tempfile.NamedTemporaryFile(mode="w")
- for clip in clips:
- join_file.write(f"file '{clip}'\n")
- join_file.flush()
-
- ffmpeg_args = ffmpeg_prefix + [join_file.name] + ["-c", "copy", output_filepath]
- logging.info(f"ffmpeg_args: {ffmpeg_args}")
- self._run_no_output(ffmpeg_args)
- join_file.close()
-
- def produce(self):
- """Produce clips or a video from the features"""
-
- clips = []
- for num, feature in enumerate(self.features):
- output_filepath = f"{self._output_dir}/highlight_{num}.mp4"
- self._ffmpeg_feature_to_clip(feature, output_filepath)
- clips.append(output_filepath)
-
- # concatenate the clips
- if self._compile_clips:
- output_filepath = f"{self._output_dir}/{self._output_filename}"
- self._ffmpeg_concat_clips(clips, output_filepath)
- logging.info(f"Produced video: {output_filepath}")
-
- class VisualisationProducer(Producer):
- """Visualisation producer -- illustrate the features we have extracted"""
- DEFAULT_OUTPUT_FILEPATH = "visualisation.png"
- def __init__(self, features, output_filepath=DEFAULT_OUTPUT_FILEPATH):
- if not features:
- raise ValueError("No features provided")
- self.features = features
- if not output_filepath:
- raise ValueError("No output filepath provided")
- self.output_filepath = output_filepath
-
- def _fe_colour(self, feature) -> str:
- """Return a colour for a feature
-
- laughter: red
- loudness: blue
- video activity: green
- words: purple
- default: pink
- """
- if feature.feature_extractor == "laughter":
- return "red"
- if feature.feature_extractor == "loudness":
- return "blue"
- if feature.feature_extractor == "videoactivity":
- return "green"
- if feature.feature_extractor == "words":
- return "purple"
- return "black"
-
- def produce(self):
- """Produce visualisation"""
- # basic idea: use matplotlib to plot:
- # - a wide line segment representing the source video[s]
- # - shorter line segments representing the features extracted where:
- # + width represents duration
- # + colour represents feature type
- # + position represents time
- # - save as image
- plotted_source_videos = []
- bar_labels = []
-
- fig, ax = plt.subplots()
- for feature in self.features:
- # plot source video line if not done already
- if feature.source not in plotted_source_videos:
- # use video duration as width
- # ax.plot([0, feature.source.duration()], [0, 0], color='black', linewidth=10)
- ax.broken_barh([(0, feature.source.duration())], (0, 5), facecolors='grey')
- plotted_source_videos.append(feature.source)
- bar_labels.append(os.path.basename(feature.source.path))
- # annotate the source video
- ax.text(0.25, 0.25, os.path.basename(feature.source.path), ha='left', va='bottom',
- fontsize=16)
-
- # plot feature line
- # ax.plot([feature.interval.start, feature.interval.end], [1, 1], color='red', linewidth=5)
- ax.broken_barh([(feature.interval.start, feature.interval.duration)],
- (10, 5), facecolors=f'{self._fe_colour(feature)}')
- if feature.feature_extractor not in bar_labels:
- bar_labels.append(feature.feature_extractor)
- # label bar with feature extractor
- # ax.text(0, 8, feature.feature_extractor, ha='left', va='bottom',
- # fontsize=16)
-
- # label the plot's axes
- ax.set_xlabel('Time')
- # ax.set_yticks([], labels=bar_labels)
- ax.set_yticks([])
- # ax.tick_params(axis='y', labelrotation=90, ha='right')
- # save the plot
- plt.savefig(self.output_filepath)
- plt.close()
-
- class PipelineJSONEncoder(json.JSONEncoder):
- def default(self, obj):
- if hasattr(obj, 'to_json'):
- return obj.to_json()
- else:
- return json.JSONEncoder.default(self, obj)
-
- class JSONProducer(Producer):
- """Produce JSON output"""
- DEFAULT_OUTPUT_FILEPATH = "features.json"
- def __init__(self, features, output_filepath=DEFAULT_OUTPUT_FILEPATH):
- if not features:
- raise ValueError("No features provided")
- self.features = features
- if not output_filepath:
- raise ValueError("No output filepath provided")
- self.output_filepath = output_filepath
-
- def produce(self):
- with open(self.output_filepath, "w") as jsonfile:
- jsonfile.write(json.dumps(self.features, cls=PipelineJSONEncoder, indent=4))
|