No puede seleccionar más de 25 temas Los temas deben comenzar con una letra o número, pueden incluir guiones ('-') y pueden tener hasta 35 caracteres de largo.

85 líneas
3.2 KiB

  1. """Classes for producing videos"""
  2. from abc import ABC
  3. import logging
  4. import subprocess
  5. import tempfile
  6. class VideoProducer(ABC):
  7. """Video producer interface."""
  8. def __init__(self, features):
  9. pass
  10. def produce(self):
  11. pass
  12. class FfmpegVideoProducer(VideoProducer):
  13. """Produce videos using ffmpeg"""
  14. # TODO: consider output filename options
  15. def __init__(self, features):
  16. if not features:
  17. raise ValueError("No features provided")
  18. # TODO: consider if we want to permit empty features (producing no video)
  19. self.features = features
  20. def _ffmpeg_feature_to_clip(self, feature=None, output_filepath=None):
  21. """use ffmpeg to produve a video clip from a feature"""
  22. OVERWRITE = True # TODO: consider making this a config option
  23. if not feature or not feature.interval:
  24. raise ValueError("No feature provided")
  25. if not output_filepath:
  26. raise ValueError("No output filepath provided")
  27. ffmpeg_prefix = ["ffmpeg", "-y"] if OVERWRITE else ["ffmpeg"]
  28. ffmpeg_suffix = ["-r", "60", "-c:v", "libx264", "-crf", "26", "-c:a", "aac", "-preset", "ultrafast"]
  29. # TODO: match framerate of input video
  30. # TODO: adjustable encoding options
  31. seek = ["-ss", str(feature.interval.start)]
  32. duration = ["-t", str(feature.interval.duration)]
  33. ffmpeg_args = ffmpeg_prefix + seek + ["-i"] + [feature.path] + duration + ffmpeg_suffix + [output_filepath]
  34. logging.info(f"ffmpeg_args: {ffmpeg_args}")
  35. subprocess.run(ffmpeg_args, stdout=None, stderr=None)
  36. def _ffmpeg_concat_clips(self, clips=None, output_filepath=None):
  37. """use ffmpeg to concatenate clips into a single video"""
  38. OVERWRITE = True
  39. ffmpeg_prefix = ["ffmpeg"]
  40. ffmpeg_prefix += ["-y"] if OVERWRITE else []
  41. ffmpeg_prefix += ["-f", "concat", "-safe", "0", "-i"]
  42. # there is a method to do this via process substitution, but it's not portable
  43. # so we'll use the input file list method
  44. if not clips:
  45. raise ValueError("No clips provided")
  46. if not output_filepath:
  47. raise ValueError("No output filepath provided")
  48. # generate a temporary file with the list of clips
  49. join_file = tempfile.NamedTemporaryFile(mode="w")
  50. for clip in clips:
  51. join_file.write(f"file '{clip}'\n")
  52. join_file.flush()
  53. ffmpeg_args = ffmpeg_prefix + [join_file.name] + ["-c", "copy", output_filepath]
  54. logging.info(f"ffmpeg_args: {ffmpeg_args}")
  55. subprocess.run(ffmpeg_args, stdout=None, stderr=None)
  56. join_file.close()
  57. def produce(self):
  58. OUTPUT_DIR = "/tmp/" # TODO: make this a config option
  59. clips = []
  60. for num, feature in enumerate(self.features):
  61. output_filepath = f"{OUTPUT_DIR}/highlight_{num}.mp4"
  62. self._ffmpeg_feature_to_clip(feature, output_filepath)
  63. clips.append(output_filepath)
  64. # concatenate the clips
  65. output_filepath = f"{OUTPUT_DIR}/highlights.mp4"
  66. self._ffmpeg_concat_clips(clips, output_filepath)
  67. logging.info(f"Produced video: {output_filepath}")