Austin Mroz
75fc10e0ff
While I have previously shied away from using imageio-ffmpeg as the default version because the included version is 4 years old and specifically does not work with the av1 format, there have been specific issues from people with a system version of ffmpeg that is even older. There has been a longstanding desire to create a system for determining the best version of ffmpeg, but this has proved a reasonably involved task. Some version information uses git commit hashes instead of a potentially sortable semvar and there are a large number of compile time flags that could result in a newer binary having less functionality. Thus a sysstem has been added that parses compile flags for desired encoders as well as the copyright year to determine which ffmpeg version is more feature full.
108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
import hashlib
|
|
import os
|
|
from typing import Iterable
|
|
import shutil
|
|
import subprocess
|
|
|
|
from .logger import logger
|
|
|
|
def ffmpeg_suitability(path):
|
|
try:
|
|
version = subprocess.run([path, "-version"], check=True,
|
|
capture_output=True).stdout.decode("utf-8")
|
|
except:
|
|
return 0
|
|
score = 0
|
|
#rough layout of the importance of various features
|
|
simple_criterion = [("libvpx", 20),("264",10), ("265",3),
|
|
("svtav1",5),("libopus", 1)]
|
|
for criterion in simple_criterion:
|
|
if version.find(criterion[0]) >= 0:
|
|
score += criterion[1]
|
|
#obtain rough compile year from copyright information
|
|
copyright_index = version.find('2000-2')
|
|
if copyright_index >= 0:
|
|
copyright_year = version[copyright_index+6:copyright_index+9]
|
|
if copyright_year.isnumeric():
|
|
score += int(copyright_year)
|
|
return score
|
|
|
|
if "VHS_FORCE_FFMPEG_PATH" in os.environ:
|
|
ffmpeg_path = os.env["VHS_FORCE_FFMPEG_PATH"]
|
|
else:
|
|
ffmpeg_paths = []
|
|
try:
|
|
from imageio_ffmpeg import get_ffmpeg_exe
|
|
imageio_ffmpeg_path = get_ffmpeg_exe()
|
|
ffmpeg_paths.append(imageio_ffmpeg_path)
|
|
except:
|
|
if "VHS_USE_IMAGEIO_FFMPEG" in os.environ:
|
|
raise
|
|
logger.warn("Failed to import imageio_ffmpeg")
|
|
if "VHS_USE_IMAGEIO_FFMPEG" in os.environ:
|
|
ffmpeg_path = imageio_ffmpeg_path
|
|
else:
|
|
system_ffmpeg = shutil.which("ffmpeg")
|
|
if system_ffmpeg is not None:
|
|
ffmpeg_paths.append(system_ffmpeg)
|
|
if len(ffmpeg_paths) == 0:
|
|
logger.error("No valid ffmpeg found.")
|
|
ffmpeg_path = max(ffmpeg_paths, key=ffmpeg_suitability)
|
|
|
|
def get_sorted_dir_files_from_directory(directory: str, skip_first_images: int=0, select_every_nth: int=1, extensions: Iterable=None):
|
|
directory = directory.strip()
|
|
dir_files = os.listdir(directory)
|
|
dir_files = sorted(dir_files)
|
|
dir_files = [os.path.join(directory, x) for x in dir_files]
|
|
dir_files = list(filter(lambda filepath: os.path.isfile(filepath), dir_files))
|
|
# filter by extension, if needed
|
|
if extensions is not None:
|
|
extensions = list(extensions)
|
|
new_dir_files = []
|
|
for filepath in dir_files:
|
|
ext = "." + filepath.split(".")[-1]
|
|
if ext.lower() in extensions:
|
|
new_dir_files.append(filepath)
|
|
dir_files = new_dir_files
|
|
# start at skip_first_images
|
|
dir_files = dir_files[skip_first_images:]
|
|
dir_files = dir_files[0::select_every_nth]
|
|
return dir_files
|
|
|
|
# modified from https://stackoverflow.com/questions/22058048/hashing-a-file-in-python
|
|
def calculate_file_hash(filename: str, hash_every_n: int = 1):
|
|
h = hashlib.sha256()
|
|
b = bytearray(10*1024*1024) # read 10 megabytes at a time
|
|
mv = memoryview(b)
|
|
with open(filename, 'rb', buffering=0) as f:
|
|
i = 0
|
|
# don't hash entire file, only portions of it if requested
|
|
while n := f.readinto(mv):
|
|
if i%hash_every_n == 0:
|
|
h.update(mv[:n])
|
|
i += 1
|
|
return h.hexdigest()
|
|
|
|
def get_audio(file, start_time=0, duration=0):
|
|
args = [ffmpeg_path, "-v", "error", "-i", file]
|
|
if start_time > 0:
|
|
args += ["-ss", str(start_time)]
|
|
if duration > 0:
|
|
args += ["-t", str(duration)]
|
|
return subprocess.run(args + ["-f", "wav", "-"],
|
|
stdout=subprocess.PIPE, check=True).stdout
|
|
def lazy_eval(func):
|
|
class Cache:
|
|
def __init__(self, func):
|
|
self.res = None
|
|
self.func = func
|
|
def get(self):
|
|
if self.res is None:
|
|
self.res = self.func()
|
|
return self.res
|
|
cache = Cache(func)
|
|
return lambda : cache.get()
|
|
|
|
def is_url(url):
|
|
return url.split("://")[0] in ["http", "https"]
|