• Redesigned FFprobeCommand to retrieve media file entries in JSON format for easier extraction and processing.
• Changed default of --input-prefix to “.”.
• Added check that the input prefix actually exists.
• Ensured that the segments list variable actually exists when cleanup happens.
1 parent 489a7fe commit 923c1e97458934f9bd20adf5705e3694d8dbb1f1
Nigel Stanger authored on 20 Sep 2016
Showing 3 changed files
View
36
process_podcast.py
"See config_help.md details on the file "
"format.".format(p=globals.PROGRAM))
parser.add_argument(
"--input-prefix", "-p", dest="prefix", metavar="PATH", default="",
"--input-prefix", "-p", dest="prefix", metavar="PATH", default=".",
help="Path to be prefixed to all INPUT files. This includes the "
"configuration file, if applicable, and any files specified "
"within the configuration file.")
 
def check_arguments(args):
"""Sanity check the command line arguments."""
fn = "check_arguments"
if (args.quiet):
if args.quiet:
globals.log.setLevel(logging.WARNING)
# --debug overrides --quiet.
if (args.debug):
if args.debug:
globals.log.setLevel(logging.DEBUG)
globals.log.debug("{fn}(): args = {a}".format(fn=fn, a=args))
# Must specify at least one of --audio, --video, --config.
if (not any([args.audio, args.video, args.config])):
if not any([args.audio, args.video, args.config]):
globals.log.error("must specify at least one of --audio, --video, "
"or --config")
sys.exit(1)
if not os.path.exists(args.prefix):
globals.log.error('input prefix "{p}" does not '
"exist".format(p=args.prefix))
sys.exit(1)
# Prepend input files with --input-prefix where applicable.
# Handily, if prefix is "", os.path.join() leaves the original
# path unchanged.
args.audio, args.video, args.config = map(
lambda f: os.path.join(args.prefix, f) if f else f,
[args.audio, args.video, args.config])
 
 
 
def get_file_duration(file):
"""Calculate the duration a media file as a timedelta object."""
command = FFprobeCommand(
["-show_entries", "format=duration",
"-print_format", "default=noprint_wrappers=1:nokey=1",
file])
ss, ms = command.get_output().strip().split(".")
fn = "get_file_duration"
command = FFprobeCommand([file])
globals.log.debug("{fn}(): {cmd}".format(fn=fn, cmd=command))
# Only consider the first stream. If it's the only stream in the
# file, great; otherwise it seems reasonable to assume that all
# streams in the same file will have the same duration.
ss, ms = command.get_entries(
section="format", find_list=["duration"])[0].split(".")
ms = ms[:3].ljust(3, "0")
globals.log.debug("{fn}(): ss = {ss}, ms = {ms}".format(fn=fn, ss=ss,
ms=ms))
return datetime.timedelta(seconds=int(ss), milliseconds=int(ms))
 
 
def make_new_segment(type, filename, punch_in, punch_out, num):
fn = "main"
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s: {p}: %(message)s".format(p=globals.PROGRAM))
segments = None
try:
args = parse_command_line()
check_arguments(args)
 
except (KeyboardInterrupt):
pass
finally:
if (not args.keep):
if segments and not args.keep:
cleanup(segments)
 
 
if (__name__ == "__main__"):
View
16
segment.py
globals.log.debug("{cls}.{fn}(): {cmd}".format(
cls=self.__class__.__name__, fn=fn, cmd=command))
if (command.run() == 0):
self._temp_files_list.append(self._temp_frame_file)
command = FFprobeCommand(
input_options=[
"-select_streams", "v",
"-show_entries", "stream=nb_frames",
"-print_format", "default=noprint_wrappers=1:nokey=1",
self._temp_frame_file])
command = FFprobeCommand([self._temp_frame_file])
globals.log.debug("{cls}.{fn}(): {cmd}".format(
cls=self.__class__.__name__, fn=fn, cmd=command))
return int(command.get_output().strip()) - 1
return int(command.get_entries(
section="stream", find_list=["nb_frames"])[0]) - 1
else:
raise SegmentException(
"Failed to generate temporary file to get last frame "
"number for {s}".format(s=self))
View
42
shell_command.py
#!/usr/bin/env python
 
import datetime
import distutils.spawn
import json
import os.path
import re
 
import pexpect
 
 
class FFprobeCommand(ShellCommand):
"""An ffprobe shell command."""
_executable = distutils.spawn.find_executable("ffprobe")
_base_options = ["-loglevel", "error"]
_base_options = ["-loglevel", "error",
"-show_entries", "format:stream",
"-print_format", "json"]
def __init__(self, input_options=[], output_options=[]):
super(FFprobeCommand, self).__init__(input_options, output_options)
self.entries = None
# The input file should be the last input option.
assert(os.path.exists(self.input_options[-1]))
self.last_modified = os.path.getmtime(self.input_options[-1])
def get_entries(self, section="stream", find_list=[]):
"""Fetch specified attributes from the input file."""
# Re-fetch if the file's changed since we last looked.
modified = os.path.getmtime(self.input_options[-1])
if (not self.entries) or (modified > self.last_modified):
js = json.loads(self.get_output())
print js
self.entries = {"format": js["format"], "stream": js["streams"][0]}
return [self.entries[section][f] for f in find_list]
 
class FFmpegCommand(ShellCommand):
"""A "simple" ffmpeg shell command."""