From ab77541f367a658e28901943e4d7eefec93a9813 Mon Sep 17 00:00:00 2001 From: Brannon King Date: Tue, 14 Jan 2020 10:43:28 -0700 Subject: [PATCH 1/9] in progress on video transcoding works Fixing lint tests remove eval --- lbry/conf.py | 15 +- lbry/extras/daemon/daemon.py | 18 +- lbry/file_analysis.py | 329 +++++++++++++++++++++++++++++++++++ scripts/check_video.py | 46 +++++ 4 files changed, 406 insertions(+), 2 deletions(-) create mode 100644 lbry/file_analysis.py create mode 100755 scripts/check_video.py diff --git a/lbry/conf.py b/lbry/conf.py index a72fd6510..09810d6a3 100644 --- a/lbry/conf.py +++ b/lbry/conf.py @@ -462,7 +462,20 @@ class BaseConfig: self.persisted.save() -class CLIConfig(BaseConfig): +class TranscodeConfig(BaseConfig): + + ffmpeg_folder = String('The path to ffmpeg and ffprobe', '') + video_encoder = String('FFmpeg codec and parameters for the video encoding. ' + 'Example: libaom-av1 -crf 25 -b:v 0 -strict experimental', + 'libx264 -crf 18 -vf "format=yuv420p"') + audio_encoder = String('FFmpeg codec and parameters for the audio encoding. ' + 'Example: libopus -b:a 128k', + 'aac -b:a 192k') + volume_filter = String('FFmpeg filter for audio normalization.', '-af loudnorm') + volume_analysis_time = Integer('Maximum seconds into the file that we examine audio volume (0 to disable).', '240') + + +class CLIConfig(TranscodeConfig): api = String('Host name and port for lbrynet daemon API.', 'localhost:5279', metavar='HOST:PORT') diff --git a/lbry/extras/daemon/daemon.py b/lbry/extras/daemon/daemon.py index 9d44cd3be..cd80d7698 100644 --- a/lbry/extras/daemon/daemon.py +++ b/lbry/extras/daemon/daemon.py @@ -44,6 +44,7 @@ from lbry.extras.daemon.componentmanager import ComponentManager from lbry.extras.daemon.json_response_encoder import JSONResponseEncoder from lbry.extras.daemon import comment_client from lbry.extras.daemon.undecorated import undecorated +from lbry.file_analysis import VideoFileAnalyzer from lbry.schema.claim import Claim from lbry.schema.url import URL @@ -296,6 +297,7 @@ class Daemon(metaclass=JSONRPCServerType): def __init__(self, conf: Config, component_manager: typing.Optional[ComponentManager] = None): self.conf = conf + self._video_file_analyzer = VideoFileAnalyzer(conf) self._node_id = None self._installation_id = None self.session_id = base58.b58encode(utils.generate_id()).decode() @@ -2808,6 +2810,7 @@ class Daemon(metaclass=JSONRPCServerType): Usage: publish ( | --name=) [--bid=] [--file_path=] + [--validate_file] [--optimize_file] [--fee_currency=] [--fee_amount=] [--fee_address=] [--title=] [--description=<description>] [--author=<author>] [--tags=<tags>...] [--languages=<languages>...] [--locations=<locations>...] @@ -2823,6 +2826,11 @@ class Daemon(metaclass=JSONRPCServerType): --name=<name> : (str) name of the content (can only consist of a-z A-Z 0-9 and -(dash)) --bid=<bid> : (decimal) amount to back the claim --file_path=<file_path> : (str) path to file to be associated with name. + --validate_file : (bool) validate that the video container and encodings match + common web browser support or that optimization succeeds if specified. + FFmpeg is required + --optimize_file : (bool) transcode the video & audio if necessary to ensure + common web browser support. FFmpeg is required --fee_currency=<fee_currency> : (string) specify fee currency --fee_amount=<fee_amount> : (decimal) content download fee --fee_address=<fee_address> : (str) address where to send fee payments, will use @@ -2994,12 +3002,13 @@ class Daemon(metaclass=JSONRPCServerType): self, name, bid, file_path, allow_duplicate_name=False, channel_id=None, channel_name=None, channel_account_id=None, account_id=None, wallet_id=None, claim_address=None, funding_account_ids=None, - preview=False, blocking=False, **kwargs): + preview=False, blocking=False, validate_file=False, optimize_file=False, **kwargs): """ Make a new stream claim and announce the associated file to lbrynet. Usage: stream_create (<name> | --name=<name>) (<bid> | --bid=<bid>) (<file_path> | --file_path=<file_path>) + [--validate_file] [--optimize_file] [--allow_duplicate_name=<allow_duplicate_name>] [--fee_currency=<fee_currency>] [--fee_amount=<fee_amount>] [--fee_address=<fee_address>] [--title=<title>] [--description=<description>] [--author=<author>] @@ -3016,6 +3025,11 @@ class Daemon(metaclass=JSONRPCServerType): --name=<name> : (str) name of the content (can only consist of a-z A-Z 0-9 and -(dash)) --bid=<bid> : (decimal) amount to back the claim --file_path=<file_path> : (str) path to file to be associated with name. + --validate_file : (bool) validate that the video container and encodings match + common web browser support or that optimization succeeds if specified. + FFmpeg is required + --optimize_file : (bool) transcode the video & audio if necessary to ensure + common web browser support. FFmpeg is required --allow_duplicate_name=<allow_duplicate_name> : (bool) create new claim even if one already exists with given name. default: false. --fee_currency=<fee_currency> : (string) specify fee currency @@ -3107,6 +3121,8 @@ class Daemon(metaclass=JSONRPCServerType): f"Use --allow-duplicate-name flag to override." ) + file_path = await self._video_file_analyzer.verify_or_repair(validate_file, optimize_file, file_path) + claim = Claim() claim.stream.update(file_path=file_path, sd_hash='0' * 96, **kwargs) tx = await Transaction.claim_create( diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py new file mode 100644 index 000000000..0f2280b72 --- /dev/null +++ b/lbry/file_analysis.py @@ -0,0 +1,329 @@ +import asyncio +import json +import logging +import os +import pathlib +import re +import shlex +import shutil + +from lbry.conf import TranscodeConfig + +log = logging.getLogger(__name__) + + +class VideoFileAnalyzer: + @staticmethod + def _matches(needles: list, haystack: list): + for needle in needles: + if needle in haystack: + return True + return False + + async def _execute(self, command, arguments): + process = await asyncio.create_subprocess_exec(self._conf.ffmpeg_folder + command, *shlex.split(arguments), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + stdout, stderr = await process.communicate() # returns when the streams are closed + return stdout.decode() + stderr.decode(), process.returncode + + async def _verify_ffmpeg_installed(self): + if self._ffmpeg_installed: + return + version, code = await self._execute("ffprobe", "-version") + if code != 0 or not version.startswith("ffprobe"): + raise Exception("Unable to locate ffprobe. Please install FFmpeg and ensure that it is callable via PATH.") + version, code = await self._execute("ffmpeg", "-version") + if code != 0 or not version.startswith("ffmpeg"): + raise Exception("Unable to locate ffmpeg. Please install FFmpeg and ensure that it is callable via PATH.") + log.debug("Using %s at %s", version.splitlines()[0].split(" Copyright")[0], + shutil.which(self._conf.ffmpeg_folder + "ffmpeg")) + self._ffmpeg_installed = True + + def __init__(self, conf: TranscodeConfig): + self._conf = conf + self._available_encoders = "" + self._ffmpeg_installed = False + + def _verify_container(self, scan_data: json): + container = scan_data["format"]["format_name"] + log.debug(" Detected container %s", container) + if not self._matches(container.split(","), ["webm", "mp4", "3gp", "ogg"]): + return "Container format is not in the approved list of WebM, MP4. Actual: " \ + + container + " [" + scan_data["format"]["format_long_name"] + "]" + return "" + + def _verify_video_encoding(self, scan_data: json): + for stream in scan_data["streams"]: + if stream["codec_type"] != "video": + continue + codec = stream["codec_name"] + log.debug(" Detected video codec %s encoding %s", codec, stream["pix_fmt"]) + if not self._matches(codec.split(","), ["h264", "vp8", "vp9", "av1", "theora"]): + return "Video codec is not in the approved list of H264, VP8, VP9, AV1, Theora. Actual: " \ + + codec + " [" + stream["codec_long_name"] + "]" + + if self._matches(codec.split(","), ["h264"]) and stream["pix_fmt"] != "yuv420p": + return "Video codec is H264, but its pixel format does not match the approved yuv420p. Actual: " \ + + stream["pix_fmt"] + + return "" + + @staticmethod + def _verify_bitrate(scan_data: json): + if "bit_rate" not in scan_data["format"]: + return "" + + bit_rate = float(scan_data["format"]["bit_rate"]) + log.debug(" Detected bitrate %s Mbps", str(bit_rate / 1000000.0)) + pixels = -1.0 + for stream in scan_data["streams"]: + if stream["codec_type"] == "video": + pieces = stream["r_frame_rate"].split('/', 1) + frame_rate = float(pieces[0]) if len(pieces) == 1 \ + else float(pieces[0]) / float(pieces[1]) + pixels = max(pixels, float(stream["height"]) * float(stream["width"]) * frame_rate) + + if pixels > 0.0 and pixels / bit_rate < 3.0: + return "Bits per second is excessive for this data; this may impact web streaming performance. Actual: " \ + + str(bit_rate / 1000000.0) + "Mbps" + + return "" + + async def _verify_faststart(self, scan_data: json, video_file): + container = scan_data["format"]["format_name"] + if self._matches(container.split(","), ["webm", "ogg"]): + return "" + + result, _ = await self._execute("ffprobe", "-v debug \"" + video_file + "\"") + iterator = re.finditer(r"\s+seeks:(\d+)\s+", result) + for match in iterator: + if int(match.group(1)) != 0: + return "Video stream descriptors are not at the start of the file (the faststart flag was not used)." + return "" + + def _verify_audio_encoding(self, scan_data: json): + for stream in scan_data["streams"]: + if stream["codec_type"] != "audio": + continue + codec = stream["codec_name"] + log.debug(" Detected audio codec %s", codec) + if not self._matches(codec.split(","), ["aac", "mp3", "flac", "vorbis", "opus"]): + return "Audio codec is not in the approved list of AAC, FLAC, MP3, Vorbis, and Opus. Actual: " \ + + codec + " [" + stream["codec_long_name"] + "]" + + return "" + + async def _verify_audio_volume(self, scan_data: json, seconds, video_file): + try: + validate_volume = int(seconds) > 0 + except ValueError: + validate_volume = 0 + + if not validate_volume: + return "" + + result, _ = await self._execute("ffmpeg", f"-i \"{video_file}\" -t {seconds}" + + f" -af volumedetect -vn -sn -dn -f null \"{os.devnull}\"") + try: + mean_volume = float(re.search(r"mean_volume:\s+([-+]?\d*\.\d+|\d+)", result).group(1)) + max_volume = float(re.search(r"max_volume:\s+([-+]?\d*\.\d+|\d+)", result).group(1)) + except Exception as e: + log.debug(" Failure in volume analysis. Message: %s", str(e)) + return "" + + if max_volume < -5.0 and mean_volume < -22.0: + return "Audio is at least five dB lower than prime. Actual max: " + str(max_volume) \ + + ", mean: " + str(mean_volume) + + log.debug(" Detected audio volume mean, max as %f dB, %f dB", mean_volume, max_volume) + + return "" + + @staticmethod + def _compute_crf(scan_data): + height = 240.0 + for stream in scan_data["streams"]: + if stream["codec_type"] == "video": + height = max(height, float(stream["height"])) + + # https://developers.google.com/media/vp9/settings/vod/ + return int(-0.011 * height + 40) + + async def _get_video_encoder(self, scan_data): + # use what the user said if it's there: + # if it's not there, use h264 if we can because it's way faster than the others + # if we don't have h264 use vp9; it's fairly compatible even though it's slow + + if not self._available_encoders: + self._available_encoders, _ = await self._execute("ffmpeg", "-encoders -v quiet") + + encoder = self._conf.video_encoder.split(" ", 1)[0] + if re.search(r"^\s*V..... " + encoder + r" ", self._available_encoders, re.MULTILINE): + return self._conf.video_encoder + + if re.search(r"^\s*V..... libx264 ", self._available_encoders, re.MULTILINE): + if encoder: + log.warning(" Using libx264 since the requested encoder was unavailable. Requested: %s", encoder) + return "libx264 -crf 19 -vf \"format=yuv420p\"" + + if not encoder: + encoder = "libx264" + + if re.search(r"^\s*V..... libvpx-vp9 ", self._available_encoders, re.MULTILINE): + log.warning(" Using libvpx-vp9 since the requested encoder was unavailable. Requested: %s", encoder) + crf = self._compute_crf(scan_data) + return "libvpx-vp9 -crf " + str(crf) + " b:v 0" + + if re.search(r"^\s*V..... libtheora", self._available_encoders, re.MULTILINE): + log.warning(" Using libtheora since the requested encoder was unavailable. Requested: %s", encoder) + return "libtheora -q:v 7" + + raise Exception("The video encoder is not available. Requested: " + encoder) + + async def _get_audio_encoder(self, scan_data, video_encoder): + # if the video encoding is theora or av1/vp8/vp9 use vorbis + # or we don't have a video encoding but we have an ogg or webm container use vorbis + # if we need to use vorbis see if the conf file has one else use our own params + # else use the user-set value if it exists + # else use aac + + if video_encoder: + wants_opus = any(encoder in video_encoder for encoder in ["av1", "vp8", "vp9", "theora"]) + else: # we're not re-encoding video + container = scan_data["format"]["format_name"] + wants_opus = self._matches(container.split(","), ["webm"]) + + if not self._available_encoders: + self._available_encoders, _ = await self._execute("ffmpeg", "-encoders -v quiet") + + if wants_opus and re.search(r"^\s*A..... libopus ", self._available_encoders, re.MULTILINE): + return "libopus -b:a 160k" + + if wants_opus and re.search(r"^\s*A..... libvorbis ", self._available_encoders, re.MULTILINE): + return "libvorbis -q:a 6" + + encoder = self._conf.audio_encoder.split(" ", 1)[0] + if re.search(r"^\s*A..... " + encoder + r" ", self._available_encoders, re.MULTILINE): + return self._conf.audio_encoder + + if re.search(r"^\s*A..... aac ", self._available_encoders, re.MULTILINE): + return "aac -b:a 192k" + + if not encoder: + encoder = "aac" + raise Exception("The audio encoder is not available. Requested: " + encoder) + + async def _get_volume_filter(self, scan_data): + return self._conf.volume_filter if self._conf.volume_filter else "-af loudnorm" + + def _get_best_container_extension(self, scan_data, video_encoder): + # the container is chosen by the video format + # if we are theora-encoded, we want ogg + # if we are vp8/vp9/av1 we want webm + # use mp4 for anything else + + if not video_encoder: # not re-encoding video + for stream in scan_data["streams"]: + if stream["codec_type"] != "video": + continue + codec = stream["codec_name"].split(",") + if self._matches(codec, ["theora"]): + return "ogg" + if self._matches(codec, ["vp8", "vp9", "av1"]): + return "webm" + + if "theora" in video_encoder: + return "ogg" + elif re.search("vp[89x]|av1", video_encoder.split(" ", 1)[0]): + return "webm" + return "mp4" + + async def verify_or_repair(self, validate, repair, file_path): + if not validate and not repair: + return file_path + + await self._verify_ffmpeg_installed() + + result, _ = await self._execute("ffprobe", + f"-v quiet -print_format json -show_format -show_streams \"{file_path}\"") + try: + scan_data = json.loads(result) + except Exception as e: + log.debug("Failure in JSON parsing ffprobe results. Message: %s", str(e)) + if validate: + raise Exception('Invalid video file: ' + file_path) + log.info("Unable to optimize %s . FFmpeg output was unreadable.", file_path) + return + + if "format" not in scan_data: + if validate: + raise Exception('Unexpected video file contents: ' + file_path) + log.info("Unable to optimize %s . FFmpeg output is missing the format section.", file_path) + return + + faststart_msg = await self._verify_faststart(scan_data, file_path) + log.debug("Analyzing %s:", file_path) + log.debug(" Detected faststart is %s", "false" if faststart_msg else "true") + container_msg = self._verify_container(scan_data) + bitrate_msg = self._verify_bitrate(scan_data) + video_msg = self._verify_video_encoding(scan_data) + audio_msg = self._verify_audio_encoding(scan_data) + volume_msg = await self._verify_audio_volume(scan_data, self._conf.volume_analysis_time, file_path) + messages = [container_msg, bitrate_msg, faststart_msg, video_msg, audio_msg, volume_msg] + + if not any(messages): + return file_path + + if not repair: + errors = "Streamability verification failed:\n" + for message in messages: + if message: + errors += " " + message + "\n" + + raise Exception(errors) + + # the plan for transcoding: + # we have to re-encode the video if it is in a nonstandard format + # we also re-encode if we are h264 but not yuv420p (both errors caught in video_msg) + # we also re-encode if our bitrate is too high + + try: + transcode_command = f"-i \"{file_path}\" -y -c:s copy -c:d copy -c:v " + + video_encoder = "" + if video_msg or bitrate_msg: + video_encoder = await self._get_video_encoder(scan_data) + transcode_command += video_encoder + " " + else: + transcode_command += "copy " + + transcode_command += "-movflags +faststart -c:a " + + if audio_msg or volume_msg: + audio_encoder = await self._get_audio_encoder(scan_data, video_encoder) + transcode_command += audio_encoder + " " + if volume_msg: + volume_filter = await self._get_volume_filter(scan_data) + transcode_command += volume_filter + " " + else: + transcode_command += "copy " + + path = pathlib.Path(file_path) + extension = self._get_best_container_extension(scan_data, video_encoder) + + # TODO: put it in a temp folder and delete it after we upload? + output = path.parent / (path.stem + "_fixed." + extension) + transcode_command += '"' + str(output) + '"' + + log.info("Proceeding on transcode via: ffmpeg %s", transcode_command) + result, code = await self._execute("ffmpeg", transcode_command) + if code != 0: + raise Exception("Failure to complete the transcode command. Output: " + result) + except Exception as e: + if validate: + raise + log.info("Unable to transcode %s . Message: %s", file_path, str(e)) + # TODO: delete partial output file here if it exists? + return file_path + + return output diff --git a/scripts/check_video.py b/scripts/check_video.py new file mode 100755 index 000000000..1a99a7251 --- /dev/null +++ b/scripts/check_video.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 + +import asyncio +import logging +import sys +import lbry.wallet # just to make the following line work: +from lbry.conf import TranscodeConfig +from lbry.file_analysis import VideoFileAnalyzer + + +def enable_logging(): + root = logging.getLogger() + root.setLevel(logging.DEBUG) + + handler = logging.StreamHandler(sys.stdout) + handler.setLevel(logging.DEBUG) + formatter = logging.Formatter('%(message)s') # %(asctime)s - %(levelname)s - + handler.setFormatter(formatter) + root.addHandler(handler) + + +async def main(): + if len(sys.argv) < 2: + print("Usage: <path to video file>", file=sys.stderr) + sys.exit(1) + video_file = sys.argv[1] + + enable_logging() + conf = TranscodeConfig() + analyzer = VideoFileAnalyzer(conf) + try: + await analyzer.verify_or_repair(True, False, video_file) + print("No concerns. Ship it!") + except Exception as e: + print(str(e)) + transcode = input("Would you like repair this via transcode now? [y/N] ") + if transcode == "y": + try: + new_video_file = await analyzer.verify_or_repair(True, True, video_file) + print("Successfully created ", new_video_file) + except Exception as e: + print("Unable to complete the transcode. Message: ", str(e)) + + +if __name__ == '__main__': + asyncio.run(main()) From fac28072ab1482187f740bb961b267a2d89a78d8 Mon Sep 17 00:00:00 2001 From: Brannon King <countprimes@gmail.com> Date: Thu, 16 Jan 2020 16:20:01 -0700 Subject: [PATCH 2/9] added unit tests, other minor fixes added universe try again try 4 --- .gitignore | 4 +- .gitlab-ci.yml | 2 + lbry/file_analysis.py | 174 +++++++++--------- setup.py | 1 + .../blockchain/test_transcoding.py | 149 +++++++++++++++ 5 files changed, 246 insertions(+), 84 deletions(-) create mode 100644 tests/integration/blockchain/test_transcoding.py diff --git a/.gitignore b/.gitignore index b0c32fdc0..36cb694de 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ /.tox /.coverage* /lbry-venv +/venv lbry.egg-info __pycache__ @@ -13,5 +14,4 @@ _trial_temp/ /tests/integration/blockchain/files /tests/.coverage.* -/lbry/wallet/bin -/venv +/lbry/wallet/bin \ No newline at end of file diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 347183a6d..2f9d2be2c 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -46,6 +46,8 @@ test:datanetwork-integration: test:blockchain-integration: stage: test script: + - apt-get update + - apt-get install -y --no-install-recommends ffmpeg - pip install tox-travis - tox -e blockchain diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index 0f2280b72..f1d3ea93e 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -26,17 +26,25 @@ class VideoFileAnalyzer: stdout, stderr = await process.communicate() # returns when the streams are closed return stdout.decode() + stderr.decode(), process.returncode + async def _verify_executable(self, name): + try: + version, code = await self._execute(name, "-version") + except Exception as e: + log.warning("Unable to run %s, but it was requested. Message: %s", name, str(e)) + code = -1 + version = "" + if code != 0 or not version.startswith(name): + raise Exception(f"Unable to locate or run {name}. Please install FFmpeg " + f"and ensure that it is callable via PATH or conf.ffmpeg_folder") + return version + async def _verify_ffmpeg_installed(self): if self._ffmpeg_installed: return - version, code = await self._execute("ffprobe", "-version") - if code != 0 or not version.startswith("ffprobe"): - raise Exception("Unable to locate ffprobe. Please install FFmpeg and ensure that it is callable via PATH.") - version, code = await self._execute("ffmpeg", "-version") - if code != 0 or not version.startswith("ffmpeg"): - raise Exception("Unable to locate ffmpeg. Please install FFmpeg and ensure that it is callable via PATH.") + await self._verify_executable("ffprobe") + version = await self._verify_executable("ffmpeg") log.debug("Using %s at %s", version.splitlines()[0].split(" Copyright")[0], - shutil.which(self._conf.ffmpeg_folder + "ffmpeg")) + shutil.which(f"{self._conf.ffmpeg_folder}ffmpeg")) self._ffmpeg_installed = True def __init__(self, conf: TranscodeConfig): @@ -48,8 +56,8 @@ class VideoFileAnalyzer: container = scan_data["format"]["format_name"] log.debug(" Detected container %s", container) if not self._matches(container.split(","), ["webm", "mp4", "3gp", "ogg"]): - return "Container format is not in the approved list of WebM, MP4. Actual: " \ - + container + " [" + scan_data["format"]["format_long_name"] + "]" + return "Container format is not in the approved list of WebM, MP4. " \ + f"Actual: {container} [{scan_data['format']['format_long_name']}]" return "" def _verify_video_encoding(self, scan_data: json): @@ -59,12 +67,12 @@ class VideoFileAnalyzer: codec = stream["codec_name"] log.debug(" Detected video codec %s encoding %s", codec, stream["pix_fmt"]) if not self._matches(codec.split(","), ["h264", "vp8", "vp9", "av1", "theora"]): - return "Video codec is not in the approved list of H264, VP8, VP9, AV1, Theora. Actual: " \ - + codec + " [" + stream["codec_long_name"] + "]" + return "Video codec is not in the approved list of H264, VP8, VP9, AV1, Theora. " \ + f"Actual: {codec} [{stream['codec_long_name']}]" if self._matches(codec.split(","), ["h264"]) and stream["pix_fmt"] != "yuv420p": - return "Video codec is H264, but its pixel format does not match the approved yuv420p. Actual: " \ - + stream["pix_fmt"] + return "Video codec is H264, but its pixel format does not match the approved yuv420p. " \ + f"Actual: {stream['pix_fmt']}" return "" @@ -84,17 +92,17 @@ class VideoFileAnalyzer: pixels = max(pixels, float(stream["height"]) * float(stream["width"]) * frame_rate) if pixels > 0.0 and pixels / bit_rate < 3.0: - return "Bits per second is excessive for this data; this may impact web streaming performance. Actual: " \ - + str(bit_rate / 1000000.0) + "Mbps" + return "Bits per second is excessive for this data; this may impact web streaming performance. " \ + f"Actual: {str(bit_rate / 1000000.0)} Mbps" return "" - async def _verify_faststart(self, scan_data: json, video_file): + async def _verify_fast_start(self, scan_data: json, video_file): container = scan_data["format"]["format_name"] if self._matches(container.split(","), ["webm", "ogg"]): return "" - result, _ = await self._execute("ffprobe", "-v debug \"" + video_file + "\"") + result, _ = await self._execute("ffprobe", f'-v debug "{video_file}"') iterator = re.finditer(r"\s+seeks:(\d+)\s+", result) for match in iterator: if int(match.group(1)) != 0: @@ -108,12 +116,12 @@ class VideoFileAnalyzer: codec = stream["codec_name"] log.debug(" Detected audio codec %s", codec) if not self._matches(codec.split(","), ["aac", "mp3", "flac", "vorbis", "opus"]): - return "Audio codec is not in the approved list of AAC, FLAC, MP3, Vorbis, and Opus. Actual: " \ - + codec + " [" + stream["codec_long_name"] + "]" + return "Audio codec is not in the approved list of AAC, FLAC, MP3, Vorbis, and Opus. " \ + f"Actual: {codec} [{stream['codec_long_name']}]" return "" - async def _verify_audio_volume(self, scan_data: json, seconds, video_file): + async def _verify_audio_volume(self, seconds, video_file): try: validate_volume = int(seconds) > 0 except ValueError: @@ -122,8 +130,8 @@ class VideoFileAnalyzer: if not validate_volume: return "" - result, _ = await self._execute("ffmpeg", f"-i \"{video_file}\" -t {seconds}" - + f" -af volumedetect -vn -sn -dn -f null \"{os.devnull}\"") + result, _ = await self._execute("ffmpeg", f'-i "{video_file}" -t {seconds} ' + f'-af volumedetect -vn -sn -dn -f null "{os.devnull}"') try: mean_volume = float(re.search(r"mean_volume:\s+([-+]?\d*\.\d+|\d+)", result).group(1)) max_volume = float(re.search(r"max_volume:\s+([-+]?\d*\.\d+|\d+)", result).group(1)) @@ -132,8 +140,8 @@ class VideoFileAnalyzer: return "" if max_volume < -5.0 and mean_volume < -22.0: - return "Audio is at least five dB lower than prime. Actual max: " + str(max_volume) \ - + ", mean: " + str(mean_volume) + return "Audio is at least five dB lower than prime. " \ + f"Actual max: {max_volume}, mean: {mean_volume}" log.debug(" Detected audio volume mean, max as %f dB, %f dB", mean_volume, max_volume) @@ -158,13 +166,13 @@ class VideoFileAnalyzer: self._available_encoders, _ = await self._execute("ffmpeg", "-encoders -v quiet") encoder = self._conf.video_encoder.split(" ", 1)[0] - if re.search(r"^\s*V..... " + encoder + r" ", self._available_encoders, re.MULTILINE): + if re.search(fr"^\s*V..... {encoder} ", self._available_encoders, re.MULTILINE): return self._conf.video_encoder if re.search(r"^\s*V..... libx264 ", self._available_encoders, re.MULTILINE): if encoder: log.warning(" Using libx264 since the requested encoder was unavailable. Requested: %s", encoder) - return "libx264 -crf 19 -vf \"format=yuv420p\"" + return 'libx264 -crf 19 -vf "format=yuv420p"' if not encoder: encoder = "libx264" @@ -172,48 +180,47 @@ class VideoFileAnalyzer: if re.search(r"^\s*V..... libvpx-vp9 ", self._available_encoders, re.MULTILINE): log.warning(" Using libvpx-vp9 since the requested encoder was unavailable. Requested: %s", encoder) crf = self._compute_crf(scan_data) - return "libvpx-vp9 -crf " + str(crf) + " b:v 0" + return f"libvpx-vp9 -crf {crf} -b:v 0" if re.search(r"^\s*V..... libtheora", self._available_encoders, re.MULTILINE): log.warning(" Using libtheora since the requested encoder was unavailable. Requested: %s", encoder) return "libtheora -q:v 7" - raise Exception("The video encoder is not available. Requested: " + encoder) + raise Exception(f"The video encoder is not available. Requested: {encoder}") - async def _get_audio_encoder(self, scan_data, video_encoder): - # if the video encoding is theora or av1/vp8/vp9 use vorbis - # or we don't have a video encoding but we have an ogg or webm container use vorbis - # if we need to use vorbis see if the conf file has one else use our own params + async def _get_audio_encoder(self, extension): + # if the video encoding is theora or av1/vp8/vp9 use opus (or fallback to vorbis) + # or we don't have a video encoding but we have an ogg or webm container use opus + # if we need to use opus/vorbis see if the conf file has it else use our own params # else use the user-set value if it exists # else use aac - if video_encoder: - wants_opus = any(encoder in video_encoder for encoder in ["av1", "vp8", "vp9", "theora"]) - else: # we're not re-encoding video - container = scan_data["format"]["format_name"] - wants_opus = self._matches(container.split(","), ["webm"]) - + wants_opus = extension != "mp4" if not self._available_encoders: self._available_encoders, _ = await self._execute("ffmpeg", "-encoders -v quiet") + encoder = self._conf.audio_encoder.split(" ", 1)[0] + if wants_opus and 'opus' in encoder: + return self._conf.audio_encoder + if wants_opus and re.search(r"^\s*A..... libopus ", self._available_encoders, re.MULTILINE): return "libopus -b:a 160k" + if wants_opus and 'vorbis' in encoder: + return self._conf.audio_encoder + if wants_opus and re.search(r"^\s*A..... libvorbis ", self._available_encoders, re.MULTILINE): return "libvorbis -q:a 6" - encoder = self._conf.audio_encoder.split(" ", 1)[0] - if re.search(r"^\s*A..... " + encoder + r" ", self._available_encoders, re.MULTILINE): + if re.search(fr"^\s*A..... {encoder} ", self._available_encoders, re.MULTILINE): return self._conf.audio_encoder if re.search(r"^\s*A..... aac ", self._available_encoders, re.MULTILINE): return "aac -b:a 192k" - if not encoder: - encoder = "aac" - raise Exception("The audio encoder is not available. Requested: " + encoder) + raise Exception(f"The audio encoder is not available. Requested: {encoder or 'aac'}") - async def _get_volume_filter(self, scan_data): + async def _get_volume_filter(self): return self._conf.volume_filter if self._conf.volume_filter else "-af loudnorm" def _get_best_container_extension(self, scan_data, video_encoder): @@ -234,42 +241,46 @@ class VideoFileAnalyzer: if "theora" in video_encoder: return "ogg" - elif re.search("vp[89x]|av1", video_encoder.split(" ", 1)[0]): + elif re.search(r"vp[89x]|av1", video_encoder.split(" ", 1)[0]): return "webm" return "mp4" + async def _get_scan_data(self, validate, file_path): + result, _ = await self._execute("ffprobe", + f'-v quiet -print_format json -show_format -show_streams "{file_path}"') + try: + scan_data = json.loads(result) + except Exception as e: + log.debug("Failure in JSON parsing ffprobe results. Message: %s", str(e)) + if validate: + raise Exception(f'Invalid video file: {file_path}') + log.info("Unable to optimize %s . FFmpeg output was unreadable.", file_path) + return + + if "format" not in scan_data: + if validate: + raise Exception(f'Unexpected video file contents in: {file_path}') + log.info("Unable to optimize %s . FFmpeg output is missing the format section.", file_path) + return + + return scan_data + async def verify_or_repair(self, validate, repair, file_path): if not validate and not repair: return file_path await self._verify_ffmpeg_installed() + scan_data = await self._get_scan_data(validate, file_path) - result, _ = await self._execute("ffprobe", - f"-v quiet -print_format json -show_format -show_streams \"{file_path}\"") - try: - scan_data = json.loads(result) - except Exception as e: - log.debug("Failure in JSON parsing ffprobe results. Message: %s", str(e)) - if validate: - raise Exception('Invalid video file: ' + file_path) - log.info("Unable to optimize %s . FFmpeg output was unreadable.", file_path) - return - - if "format" not in scan_data: - if validate: - raise Exception('Unexpected video file contents: ' + file_path) - log.info("Unable to optimize %s . FFmpeg output is missing the format section.", file_path) - return - - faststart_msg = await self._verify_faststart(scan_data, file_path) + fast_start_msg = await self._verify_fast_start(scan_data, file_path) log.debug("Analyzing %s:", file_path) - log.debug(" Detected faststart is %s", "false" if faststart_msg else "true") + log.debug(" Detected faststart is %s", "false" if fast_start_msg else "true") container_msg = self._verify_container(scan_data) bitrate_msg = self._verify_bitrate(scan_data) video_msg = self._verify_video_encoding(scan_data) audio_msg = self._verify_audio_encoding(scan_data) - volume_msg = await self._verify_audio_volume(scan_data, self._conf.volume_analysis_time, file_path) - messages = [container_msg, bitrate_msg, faststart_msg, video_msg, audio_msg, volume_msg] + volume_msg = await self._verify_audio_volume(self._conf.volume_analysis_time, file_path) + messages = [container_msg, bitrate_msg, fast_start_msg, video_msg, audio_msg, volume_msg] if not any(messages): return file_path @@ -278,7 +289,7 @@ class VideoFileAnalyzer: errors = "Streamability verification failed:\n" for message in messages: if message: - errors += " " + message + "\n" + errors += f" {message}\n" raise Exception(errors) @@ -288,37 +299,36 @@ class VideoFileAnalyzer: # we also re-encode if our bitrate is too high try: - transcode_command = f"-i \"{file_path}\" -y -c:s copy -c:d copy -c:v " + transcode_command = f'-i "{file_path}" -y -c:s copy -c:d copy -c:v ' video_encoder = "" if video_msg or bitrate_msg: video_encoder = await self._get_video_encoder(scan_data) - transcode_command += video_encoder + " " + transcode_command += f"{video_encoder} " else: transcode_command += "copy " transcode_command += "-movflags +faststart -c:a " - - if audio_msg or volume_msg: - audio_encoder = await self._get_audio_encoder(scan_data, video_encoder) - transcode_command += audio_encoder + " " - if volume_msg: - volume_filter = await self._get_volume_filter(scan_data) - transcode_command += volume_filter + " " - else: - transcode_command += "copy " - path = pathlib.Path(file_path) extension = self._get_best_container_extension(scan_data, video_encoder) + if audio_msg or volume_msg: + audio_encoder = await self._get_audio_encoder(extension) + transcode_command += f"{audio_encoder} " + if volume_msg: + volume_filter = await self._get_volume_filter() + transcode_command += f"{volume_filter} " + else: + transcode_command += "copy " + # TODO: put it in a temp folder and delete it after we upload? - output = path.parent / (path.stem + "_fixed." + extension) - transcode_command += '"' + str(output) + '"' + output = path.parent / f"{path.stem}_fixed.{extension}" + transcode_command += f'"{output}"' log.info("Proceeding on transcode via: ffmpeg %s", transcode_command) result, code = await self._execute("ffmpeg", transcode_command) if code != 0: - raise Exception("Failure to complete the transcode command. Output: " + result) + raise Exception(f"Failure to complete the transcode command. Output: {result}") except Exception as e: if validate: raise diff --git a/setup.py b/setup.py index 42ed2035f..5c60f9fa1 100644 --- a/setup.py +++ b/setup.py @@ -33,6 +33,7 @@ setup( ], }, install_requires=[ + 'wheel', 'aiohttp==3.5.4', 'aioupnp==0.0.17', 'appdirs==1.4.3', diff --git a/tests/integration/blockchain/test_transcoding.py b/tests/integration/blockchain/test_transcoding.py new file mode 100644 index 000000000..125b4c795 --- /dev/null +++ b/tests/integration/blockchain/test_transcoding.py @@ -0,0 +1,149 @@ +import logging +import pathlib +import time + +import lbry.wallet # just to make the following line work: +from lbry.conf import TranscodeConfig +from lbry.file_analysis import VideoFileAnalyzer +from tests.integration.blockchain.test_claim_commands import ClaimTestCase + +log = logging.getLogger(__name__) + + +class MeasureTime: + def __init__(self, text): + print(text, end="...", flush=True) + + def __enter__(self): + self.start = time.perf_counter() + + def __exit__(self, exc_type, exc_val, exc_tb): + end = time.perf_counter() + print(f" done in {end - self.start:.6f}s", flush=True) + + +class TranscodeValidation(ClaimTestCase): + + def make_name(self, name, extension=""): + path = pathlib.Path(self.video_file_name) + return path.parent / f"{path.stem}_{name}{extension or path.suffix}" + + async def asyncSetUp(self): + await super().asyncSetUp() + self.conf = TranscodeConfig() + self.conf.volume_analysis_time = 0 # disable it as the test file isn't very good here + self.analyzer = VideoFileAnalyzer(self.conf) + file_ogg = self.make_name("ogg", ".ogg") + if not file_ogg.exists(): + command = f'-i "{self.video_file_name}" -c:v libtheora -q:v 4 -c:a libvorbis -q:a 4 ' \ + f'-c:s copy -c:d copy "{file_ogg}"' + with MeasureTime(f"Creating {file_ogg.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + file_webm = self.make_name("webm", ".webm") + if not file_webm.exists(): + command = f'-i "{self.video_file_name}" -c:v libvpx-vp9 -crf 36 -b:v 0 -cpu-used 2 ' \ + f'-c:a libopus -b:a 128k -c:s copy -c:d copy "{file_webm}"' + with MeasureTime(f"Creating {file_webm.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + self.should_work = [self.video_file_name, str(file_ogg), str(file_webm)] + + async def test_should_work(self): + for should_work_file_name in self.should_work: + new_file_name = await self.analyzer.verify_or_repair(True, False, should_work_file_name) + self.assertEqual(should_work_file_name, new_file_name) + + async def test_volume(self): + try: + self.conf.volume_analysis_time = 200 + with self.assertRaisesRegex(Exception, "lower than prime"): + await self.analyzer.verify_or_repair(True, False, self.video_file_name) + finally: + self.conf.volume_analysis_time = 0 + + async def test_container(self): + file_name = self.make_name("bad_container", ".avi") + if not file_name.exists(): + command = f'-i "{self.video_file_name}" -c copy -map 0 "{file_name}"' + with MeasureTime(f"Creating {file_name.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + with self.assertRaisesRegex(Exception, "Container format is not in the approved list"): + await self.analyzer.verify_or_repair(True, False, file_name) + + fixed_file = await self.analyzer.verify_or_repair(True, True, file_name) + pathlib.Path(fixed_file).unlink() + + async def test_video_codec(self): + file_name = self.make_name("bad_video_codec_1") + if not file_name.exists(): + command = f'-i "{self.video_file_name}" -c copy -map 0 -c:v libx265 -preset superfast "{file_name}"' + with MeasureTime(f"Creating {file_name.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + with self.assertRaisesRegex(Exception, "Video codec is not in the approved list"): + await self.analyzer.verify_or_repair(True, False, file_name) + with self.assertRaisesRegex(Exception, "faststart flag was not used"): + await self.analyzer.verify_or_repair(True, False, file_name) + + fixed_file = await self.analyzer.verify_or_repair(True, True, file_name) + pathlib.Path(fixed_file).unlink() + + async def test_video_format(self): + file_name = self.make_name("bad_video_format_1") + if not file_name.exists(): + command = f'-i "{self.video_file_name}" -c copy -map 0 -c:v libx264 ' \ + f'-vf format=yuv444p "{file_name}"' + with MeasureTime(f"Creating {file_name.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + with self.assertRaisesRegex(Exception, "pixel format does not match the approved"): + await self.analyzer.verify_or_repair(True, False, file_name) + + fixed_file = await self.analyzer.verify_or_repair(True, True, file_name) + pathlib.Path(fixed_file).unlink() + + async def test_audio_codec(self): + file_name = self.make_name("bad_audio_codec_1", ".mkv") + if not file_name.exists(): + command = f'-i "{self.video_file_name}" -c copy -map 0 -c:a pcm_s16le "{file_name}"' + with MeasureTime(f"Creating {file_name.name}"): + output, code = await self.analyzer._execute("ffmpeg", command) + self.assertEqual(code, 0, output) + + with self.assertRaisesRegex(Exception, "Audio codec is not in the approved list"): + await self.analyzer.verify_or_repair(True, False, file_name) + + fixed_file = await self.analyzer.verify_or_repair(True, True, file_name) + pathlib.Path(fixed_file).unlink() + + async def test_extension_choice(self): + + for file_name in self.should_work: + scan_data = await self.analyzer._get_scan_data(True, file_name) + extension = self.analyzer._get_best_container_extension(scan_data, "") + self.assertEqual(extension, pathlib.Path(file_name).suffix[1:]) + + extension = self.analyzer._get_best_container_extension("", "libx264 -crf 23") + self.assertEqual("mp4", extension) + + extension = self.analyzer._get_best_container_extension("", "libvpx-vp9 -crf 23") + self.assertEqual("webm", extension) + + extension = self.analyzer._get_best_container_extension("", "libtheora") + self.assertEqual("ogg", extension) + + async def test_no_ffmpeg(self): + try: + self.conf.ffmpeg_folder = "I don't really exist/" + with self.assertRaisesRegex(Exception, "Unable to locate"): + await self.analyzer.verify_or_repair(True, False, self.video_file_name) + finally: + self.conf.ffmpeg_folder = "" + From 47e8f74da9879239f6066f346cfaf109f54b92e9 Mon Sep 17 00:00:00 2001 From: Brannon King <countprimes@gmail.com> Date: Thu, 16 Jan 2020 17:51:49 -0700 Subject: [PATCH 3/9] changed to list append, relative claim test --- lbry/file_analysis.py | 30 +++++++++---------- .../blockchain/test_transcoding.py | 3 +- .../blockchain/test_wallet_commands.py | 5 ++-- 3 files changed, 18 insertions(+), 20 deletions(-) diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index f1d3ea93e..135942411 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -286,12 +286,9 @@ class VideoFileAnalyzer: return file_path if not repair: - errors = "Streamability verification failed:\n" - for message in messages: - if message: - errors += f" {message}\n" - - raise Exception(errors) + errors = ["Streamability verification failed:"] + errors.extend(filter(None, messages)) + raise Exception("\n ".join(errors)) # the plan for transcoding: # we have to re-encode the video if it is in a nonstandard format @@ -299,34 +296,35 @@ class VideoFileAnalyzer: # we also re-encode if our bitrate is too high try: - transcode_command = f'-i "{file_path}" -y -c:s copy -c:d copy -c:v ' + transcode_command = [f'-i "{file_path}" -y -c:s copy -c:d copy -c:v'] video_encoder = "" if video_msg or bitrate_msg: video_encoder = await self._get_video_encoder(scan_data) - transcode_command += f"{video_encoder} " + transcode_command.append(video_encoder) else: - transcode_command += "copy " + transcode_command.append("copy") - transcode_command += "-movflags +faststart -c:a " + transcode_command.append("-movflags +faststart -c:a") path = pathlib.Path(file_path) extension = self._get_best_container_extension(scan_data, video_encoder) if audio_msg or volume_msg: audio_encoder = await self._get_audio_encoder(extension) - transcode_command += f"{audio_encoder} " + transcode_command.append(audio_encoder) if volume_msg: volume_filter = await self._get_volume_filter() - transcode_command += f"{volume_filter} " + transcode_command.append(volume_filter) else: - transcode_command += "copy " + transcode_command.append("copy") # TODO: put it in a temp folder and delete it after we upload? output = path.parent / f"{path.stem}_fixed.{extension}" - transcode_command += f'"{output}"' + transcode_command.append(f'"{output}"') - log.info("Proceeding on transcode via: ffmpeg %s", transcode_command) - result, code = await self._execute("ffmpeg", transcode_command) + ffmpeg_command = " ".join(transcode_command) + log.info("Proceeding on transcode via: ffmpeg %s", ffmpeg_command) + result, code = await self._execute("ffmpeg", ffmpeg_command) if code != 0: raise Exception(f"Failure to complete the transcode command. Output: {result}") except Exception as e: diff --git a/tests/integration/blockchain/test_transcoding.py b/tests/integration/blockchain/test_transcoding.py index 125b4c795..12ed4c51d 100644 --- a/tests/integration/blockchain/test_transcoding.py +++ b/tests/integration/blockchain/test_transcoding.py @@ -5,7 +5,7 @@ import time import lbry.wallet # just to make the following line work: from lbry.conf import TranscodeConfig from lbry.file_analysis import VideoFileAnalyzer -from tests.integration.blockchain.test_claim_commands import ClaimTestCase +from .test_claim_commands import ClaimTestCase log = logging.getLogger(__name__) @@ -146,4 +146,3 @@ class TranscodeValidation(ClaimTestCase): await self.analyzer.verify_or_repair(True, False, self.video_file_name) finally: self.conf.ffmpeg_folder = "" - diff --git a/tests/integration/blockchain/test_wallet_commands.py b/tests/integration/blockchain/test_wallet_commands.py index 126f3fd9a..8500efa3c 100644 --- a/tests/integration/blockchain/test_wallet_commands.py +++ b/tests/integration/blockchain/test_wallet_commands.py @@ -199,8 +199,9 @@ class WalletEncryptionAndSynchronization(CommandTestCase): await self.confirm_tx(sendtxid, self.daemon2.ledger) def assertWalletEncrypted(self, wallet_path, encrypted): - wallet = json.load(open(wallet_path)) - self.assertEqual(wallet['accounts'][0]['private_key'][1:4] != 'prv', encrypted) + with open(wallet_path) as opened: + wallet = json.load(opened) + self.assertEqual(wallet['accounts'][0]['private_key'][1:4] != 'prv', encrypted) async def test_sync(self): daemon, daemon2 = self.daemon, self.daemon2 From a90b60799ae4974dd64a931f2ca01678ca8916d8 Mon Sep 17 00:00:00 2001 From: Brannon King <countprimes@gmail.com> Date: Thu, 30 Jan 2020 10:37:08 -0700 Subject: [PATCH 4/9] Fixed check_video.py on Windows using a cross-platform workaround fixed proactor use in the SDK fixed linter --- lbry/conf.py | 8 ++++++++ lbry/extras/cli.py | 2 ++ lbry/file_analysis.py | 17 +++++++++-------- scripts/check_video.py | 37 +++++++++++++++++++++++++------------ 4 files changed, 44 insertions(+), 20 deletions(-) diff --git a/lbry/conf.py b/lbry/conf.py index 09810d6a3..5ab16951f 100644 --- a/lbry/conf.py +++ b/lbry/conf.py @@ -1,5 +1,6 @@ import os import re +import platform import sys import typing import logging @@ -461,6 +462,13 @@ class BaseConfig: if self.persisted.upgrade(): self.persisted.save() + @property + def needs_proactor(self): + major, minor, _ = platform.python_version_tuple() + if int(major) > 3 or (int(major) == 3 and int(minor) > 7): + return False + return platform.system() == "Windows" + class TranscodeConfig(BaseConfig): diff --git a/lbry/extras/cli.py b/lbry/extras/cli.py index acdd1933e..8c3699826 100644 --- a/lbry/extras/cli.py +++ b/lbry/extras/cli.py @@ -262,6 +262,8 @@ def setup_logging(logger: logging.Logger, args: argparse.Namespace, conf: Config def run_daemon(args: argparse.Namespace, conf: Config): + if conf.needs_proactor: + asyncio.set_event_loop(asyncio.ProactorEventLoop()) loop = asyncio.get_event_loop() if args.verbose is not None: loop.set_debug(True) diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index 135942411..1e94d6da8 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -21,7 +21,8 @@ class VideoFileAnalyzer: return False async def _execute(self, command, arguments): - process = await asyncio.create_subprocess_exec(self._conf.ffmpeg_folder + command, *shlex.split(arguments), + args = shlex.split(arguments) + process = await asyncio.create_subprocess_exec(self._conf.ffmpeg_folder + command, *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, stderr = await process.communicate() # returns when the streams are closed return stdout.decode() + stderr.decode(), process.returncode @@ -34,8 +35,8 @@ class VideoFileAnalyzer: code = -1 version = "" if code != 0 or not version.startswith(name): - raise Exception(f"Unable to locate or run {name}. Please install FFmpeg " - f"and ensure that it is callable via PATH or conf.ffmpeg_folder") + raise FileNotFoundError(f"Unable to locate or run {name}. Please install FFmpeg " + f"and ensure that it is callable via PATH or conf.ffmpeg_folder") return version async def _verify_ffmpeg_installed(self): @@ -54,7 +55,7 @@ class VideoFileAnalyzer: def _verify_container(self, scan_data: json): container = scan_data["format"]["format_name"] - log.debug(" Detected container %s", container) + log.debug(" Detected container is %s", container) if not self._matches(container.split(","), ["webm", "mp4", "3gp", "ogg"]): return "Container format is not in the approved list of WebM, MP4. " \ f"Actual: {container} [{scan_data['format']['format_long_name']}]" @@ -65,7 +66,7 @@ class VideoFileAnalyzer: if stream["codec_type"] != "video": continue codec = stream["codec_name"] - log.debug(" Detected video codec %s encoding %s", codec, stream["pix_fmt"]) + log.debug(" Detected video codec is %s, format is %s", codec, stream["pix_fmt"]) if not self._matches(codec.split(","), ["h264", "vp8", "vp9", "av1", "theora"]): return "Video codec is not in the approved list of H264, VP8, VP9, AV1, Theora. " \ f"Actual: {codec} [{stream['codec_long_name']}]" @@ -82,7 +83,7 @@ class VideoFileAnalyzer: return "" bit_rate = float(scan_data["format"]["bit_rate"]) - log.debug(" Detected bitrate %s Mbps", str(bit_rate / 1000000.0)) + log.debug(" Detected bitrate is %s Mbps", str(bit_rate / 1000000.0)) pixels = -1.0 for stream in scan_data["streams"]: if stream["codec_type"] == "video": @@ -114,7 +115,7 @@ class VideoFileAnalyzer: if stream["codec_type"] != "audio": continue codec = stream["codec_name"] - log.debug(" Detected audio codec %s", codec) + log.debug(" Detected audio codec is %s", codec) if not self._matches(codec.split(","), ["aac", "mp3", "flac", "vorbis", "opus"]): return "Audio codec is not in the approved list of AAC, FLAC, MP3, Vorbis, and Opus. " \ f"Actual: {codec} [{stream['codec_long_name']}]" @@ -143,7 +144,7 @@ class VideoFileAnalyzer: return "Audio is at least five dB lower than prime. " \ f"Actual max: {max_volume}, mean: {mean_volume}" - log.debug(" Detected audio volume mean, max as %f dB, %f dB", mean_volume, max_volume) + log.debug(" Detected audio volume has mean, max of %f, %f dB", mean_volume, max_volume) return "" diff --git a/scripts/check_video.py b/scripts/check_video.py index 1a99a7251..f3215060c 100755 --- a/scripts/check_video.py +++ b/scripts/check_video.py @@ -3,7 +3,9 @@ import asyncio import logging import sys -import lbry.wallet # just to make the following line work: + +# noinspection PyUnresolvedReferences +import lbry.wallet # needed to make the following line work (it's a bug): from lbry.conf import TranscodeConfig from lbry.file_analysis import VideoFileAnalyzer @@ -19,21 +21,15 @@ def enable_logging(): root.addHandler(handler) -async def main(): - if len(sys.argv) < 2: - print("Usage: <path to video file>", file=sys.stderr) - sys.exit(1) - video_file = sys.argv[1] - - enable_logging() - conf = TranscodeConfig() - analyzer = VideoFileAnalyzer(conf) +async def process_video(analyzer, video_file): try: await analyzer.verify_or_repair(True, False, video_file) print("No concerns. Ship it!") + except FileNotFoundError as e: + print(str(e)) except Exception as e: print(str(e)) - transcode = input("Would you like repair this via transcode now? [y/N] ") + transcode = input("Would you like to make a repaired clone now? [y/N] ") if transcode == "y": try: new_video_file = await analyzer.verify_or_repair(True, True, video_file) @@ -42,5 +38,22 @@ async def main(): print("Unable to complete the transcode. Message: ", str(e)) +def main(): + if len(sys.argv) < 2: + print("Usage: check_video.py <path to video file>", file=sys.stderr) + sys.exit(1) + + enable_logging() + + video_file = sys.argv[1] + conf = TranscodeConfig() + analyzer = VideoFileAnalyzer(conf) + loop = asyncio.ProactorEventLoop() if conf.needs_proactor else asyncio.get_event_loop() + try: + loop.run_until_complete(process_video(analyzer, video_file)) + except KeyboardInterrupt: + pass + + if __name__ == '__main__': - asyncio.run(main()) + main() From 85ad972ca829be5bb94ca96567d62df1d95264db Mon Sep 17 00:00:00 2001 From: Brannon King <countprimes@gmail.com> Date: Fri, 31 Jan 2020 15:27:32 -0700 Subject: [PATCH 5/9] return string instead of path object for compatibility with hachoir --- lbry/file_analysis.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index 1e94d6da8..5f837c532 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -335,4 +335,4 @@ class VideoFileAnalyzer: # TODO: delete partial output file here if it exists? return file_path - return output + return str(output) From 1780ddd32900d4fa449ba3532ee2b47e86ecd08a Mon Sep 17 00:00:00 2001 From: Brannon King <countprimes@gmail.com> Date: Mon, 3 Feb 2020 15:53:27 -0700 Subject: [PATCH 6/9] added ffmpeg status, addressed items from code review linter --- lbry/conf.py | 8 --- lbry/extras/cli.py | 4 +- lbry/extras/daemon/daemon.py | 2 + lbry/file_analysis.py | 59 ++++++++++++------- scripts/check_video.py | 7 ++- .../blockchain/test_transcoding.py | 28 ++++++--- 6 files changed, 67 insertions(+), 41 deletions(-) diff --git a/lbry/conf.py b/lbry/conf.py index 5ab16951f..09810d6a3 100644 --- a/lbry/conf.py +++ b/lbry/conf.py @@ -1,6 +1,5 @@ import os import re -import platform import sys import typing import logging @@ -462,13 +461,6 @@ class BaseConfig: if self.persisted.upgrade(): self.persisted.save() - @property - def needs_proactor(self): - major, minor, _ = platform.python_version_tuple() - if int(major) > 3 or (int(major) == 3 and int(minor) > 7): - return False - return platform.system() == "Windows" - class TranscodeConfig(BaseConfig): diff --git a/lbry/extras/cli.py b/lbry/extras/cli.py index 8c3699826..e0f6de36a 100644 --- a/lbry/extras/cli.py +++ b/lbry/extras/cli.py @@ -3,6 +3,7 @@ import sys import shutil import signal import pathlib +import platform import json import asyncio import argparse @@ -262,7 +263,8 @@ def setup_logging(logger: logging.Logger, args: argparse.Namespace, conf: Config def run_daemon(args: argparse.Namespace, conf: Config): - if conf.needs_proactor: + if sys.version_info < (3, 8) and platform.system() == "Windows": + # TODO: remove after we move to requiring Python 3.8 asyncio.set_event_loop(asyncio.ProactorEventLoop()) loop = asyncio.get_event_loop() if args.verbose is not None: diff --git a/lbry/extras/daemon/daemon.py b/lbry/extras/daemon/daemon.py index cd80d7698..d7333e967 100644 --- a/lbry/extras/daemon/daemon.py +++ b/lbry/extras/daemon/daemon.py @@ -853,6 +853,7 @@ class Daemon(metaclass=JSONRPCServerType): """ connection_code = await self.get_connection_status() + ffmpeg_status = await self._video_file_analyzer.status() response = { 'installation_id': self.installation_id, @@ -863,6 +864,7 @@ class Daemon(metaclass=JSONRPCServerType): 'code': connection_code, 'message': CONNECTION_MESSAGES[connection_code], }, + 'ffmpeg_status': ffmpeg_status } for component in self.component_manager.components: status = await component.get_status() diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index 5f837c532..75d119647 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -13,12 +13,6 @@ log = logging.getLogger(__name__) class VideoFileAnalyzer: - @staticmethod - def _matches(needles: list, haystack: list): - for needle in needles: - if needle in haystack: - return True - return False async def _execute(self, command, arguments): args = shlex.split(arguments) @@ -44,34 +38,55 @@ class VideoFileAnalyzer: return await self._verify_executable("ffprobe") version = await self._verify_executable("ffmpeg") - log.debug("Using %s at %s", version.splitlines()[0].split(" Copyright")[0], - shutil.which(f"{self._conf.ffmpeg_folder}ffmpeg")) + self._which = shutil.which(f"{self._conf.ffmpeg_folder}ffmpeg") self._ffmpeg_installed = True + log.debug("Using %s at %s", version.splitlines()[0].split(" Copyright")[0], self._which) def __init__(self, conf: TranscodeConfig): self._conf = conf self._available_encoders = "" self._ffmpeg_installed = False + self._which = None - def _verify_container(self, scan_data: json): + async def status(self, reset=False): + if reset: + self._available_encoders = "" + self._ffmpeg_installed = False + self._which = None + + installed = True + try: + await self._verify_ffmpeg_installed() + except FileNotFoundError: + installed = False + + return { + "available": installed, + "which": self._which, + "analyze_audio_volume": int(self._conf.volume_analysis_time) > 0 + } + + @staticmethod + def _verify_container(scan_data: json): container = scan_data["format"]["format_name"] log.debug(" Detected container is %s", container) - if not self._matches(container.split(","), ["webm", "mp4", "3gp", "ogg"]): + if not {"webm", "mp4", "3gp", "ogg"}.intersection(container.split(",")): return "Container format is not in the approved list of WebM, MP4. " \ f"Actual: {container} [{scan_data['format']['format_long_name']}]" return "" - def _verify_video_encoding(self, scan_data: json): + @staticmethod + def _verify_video_encoding(scan_data: json): for stream in scan_data["streams"]: if stream["codec_type"] != "video": continue codec = stream["codec_name"] log.debug(" Detected video codec is %s, format is %s", codec, stream["pix_fmt"]) - if not self._matches(codec.split(","), ["h264", "vp8", "vp9", "av1", "theora"]): + if not {"h264", "vp8", "vp9", "av1", "theora"}.intersection(codec.split(",")): return "Video codec is not in the approved list of H264, VP8, VP9, AV1, Theora. " \ f"Actual: {codec} [{stream['codec_long_name']}]" - if self._matches(codec.split(","), ["h264"]) and stream["pix_fmt"] != "yuv420p": + if "h264" in codec.split(",") and stream["pix_fmt"] != "yuv420p": return "Video codec is H264, but its pixel format does not match the approved yuv420p. " \ f"Actual: {stream['pix_fmt']}" @@ -100,7 +115,7 @@ class VideoFileAnalyzer: async def _verify_fast_start(self, scan_data: json, video_file): container = scan_data["format"]["format_name"] - if self._matches(container.split(","), ["webm", "ogg"]): + if {"webm", "ogg"}.intersection(container.split(",")): return "" result, _ = await self._execute("ffprobe", f'-v debug "{video_file}"') @@ -110,13 +125,14 @@ class VideoFileAnalyzer: return "Video stream descriptors are not at the start of the file (the faststart flag was not used)." return "" - def _verify_audio_encoding(self, scan_data: json): + @staticmethod + def _verify_audio_encoding(scan_data: json): for stream in scan_data["streams"]: if stream["codec_type"] != "audio": continue codec = stream["codec_name"] log.debug(" Detected audio codec is %s", codec) - if not self._matches(codec.split(","), ["aac", "mp3", "flac", "vorbis", "opus"]): + if not {"aac", "mp3", "flac", "vorbis", "opus"}.intersection(codec.split(",")): return "Audio codec is not in the approved list of AAC, FLAC, MP3, Vorbis, and Opus. " \ f"Actual: {codec} [{stream['codec_long_name']}]" @@ -126,7 +142,7 @@ class VideoFileAnalyzer: try: validate_volume = int(seconds) > 0 except ValueError: - validate_volume = 0 + validate_volume = False if not validate_volume: return "" @@ -224,7 +240,8 @@ class VideoFileAnalyzer: async def _get_volume_filter(self): return self._conf.volume_filter if self._conf.volume_filter else "-af loudnorm" - def _get_best_container_extension(self, scan_data, video_encoder): + @staticmethod + def _get_best_container_extension(scan_data, video_encoder): # the container is chosen by the video format # if we are theora-encoded, we want ogg # if we are vp8/vp9/av1 we want webm @@ -235,9 +252,9 @@ class VideoFileAnalyzer: if stream["codec_type"] != "video": continue codec = stream["codec_name"].split(",") - if self._matches(codec, ["theora"]): + if "theora" in codec: return "ogg" - if self._matches(codec, ["vp8", "vp9", "av1"]): + if {"vp8", "vp9", "av1"}.intersection(codec): return "webm" if "theora" in video_encoder: @@ -260,7 +277,7 @@ class VideoFileAnalyzer: if "format" not in scan_data: if validate: - raise Exception(f'Unexpected video file contents in: {file_path}') + raise FileNotFoundError(f'Unexpected or absent video file contents at: {file_path}') log.info("Unable to optimize %s . FFmpeg output is missing the format section.", file_path) return diff --git a/scripts/check_video.py b/scripts/check_video.py index f3215060c..ee3a26899 100755 --- a/scripts/check_video.py +++ b/scripts/check_video.py @@ -2,6 +2,7 @@ import asyncio import logging +import platform import sys # noinspection PyUnresolvedReferences @@ -48,9 +49,11 @@ def main(): video_file = sys.argv[1] conf = TranscodeConfig() analyzer = VideoFileAnalyzer(conf) - loop = asyncio.ProactorEventLoop() if conf.needs_proactor else asyncio.get_event_loop() + if sys.version_info < (3, 8) and platform.system() == "Windows": + # TODO: remove after we move to requiring Python 3.8 + asyncio.set_event_loop(asyncio.ProactorEventLoop()) try: - loop.run_until_complete(process_video(analyzer, video_file)) + asyncio.run(process_video(analyzer, video_file)) except KeyboardInterrupt: pass diff --git a/tests/integration/blockchain/test_transcoding.py b/tests/integration/blockchain/test_transcoding.py index 12ed4c51d..3fec6cc33 100644 --- a/tests/integration/blockchain/test_transcoding.py +++ b/tests/integration/blockchain/test_transcoding.py @@ -34,6 +34,7 @@ class TranscodeValidation(ClaimTestCase): self.conf.volume_analysis_time = 0 # disable it as the test file isn't very good here self.analyzer = VideoFileAnalyzer(self.conf) file_ogg = self.make_name("ogg", ".ogg") + self.video_file_ogg = str(file_ogg) if not file_ogg.exists(): command = f'-i "{self.video_file_name}" -c:v libtheora -q:v 4 -c:a libvorbis -q:a 4 ' \ f'-c:s copy -c:d copy "{file_ogg}"' @@ -42,6 +43,7 @@ class TranscodeValidation(ClaimTestCase): self.assertEqual(code, 0, output) file_webm = self.make_name("webm", ".webm") + self.video_file_webm = str(file_webm) if not file_webm.exists(): command = f'-i "{self.video_file_name}" -c:v libvpx-vp9 -crf 36 -b:v 0 -cpu-used 2 ' \ f'-c:a libopus -b:a 128k -c:s copy -c:d copy "{file_webm}"' @@ -49,12 +51,13 @@ class TranscodeValidation(ClaimTestCase): output, code = await self.analyzer._execute("ffmpeg", command) self.assertEqual(code, 0, output) - self.should_work = [self.video_file_name, str(file_ogg), str(file_webm)] - async def test_should_work(self): - for should_work_file_name in self.should_work: - new_file_name = await self.analyzer.verify_or_repair(True, False, should_work_file_name) - self.assertEqual(should_work_file_name, new_file_name) + new_file_name = await self.analyzer.verify_or_repair(True, False, self.video_file_name) + self.assertEqual(self.video_file_name, new_file_name) + new_file_name = await self.analyzer.verify_or_repair(True, False, self.video_file_ogg) + self.assertEqual(self.video_file_ogg, new_file_name) + new_file_name = await self.analyzer.verify_or_repair(True, False, self.video_file_webm) + self.assertEqual(self.video_file_webm, new_file_name) async def test_volume(self): try: @@ -125,10 +128,17 @@ class TranscodeValidation(ClaimTestCase): async def test_extension_choice(self): - for file_name in self.should_work: - scan_data = await self.analyzer._get_scan_data(True, file_name) - extension = self.analyzer._get_best_container_extension(scan_data, "") - self.assertEqual(extension, pathlib.Path(file_name).suffix[1:]) + scan_data = await self.analyzer._get_scan_data(True, self.video_file_name) + extension = self.analyzer._get_best_container_extension(scan_data, "") + self.assertEqual(extension, pathlib.Path(self.video_file_name).suffix[1:]) + + scan_data = await self.analyzer._get_scan_data(True, self.video_file_ogg) + extension = self.analyzer._get_best_container_extension(scan_data, "") + self.assertEqual(extension, "ogg") + + scan_data = await self.analyzer._get_scan_data(True, self.video_file_webm) + extension = self.analyzer._get_best_container_extension(scan_data, "") + self.assertEqual(extension, "webm") extension = self.analyzer._get_best_container_extension("", "libx264 -crf 23") self.assertEqual("mp4", extension) From 6525ee65102f4d25db309320e5839dd1c37bfd86 Mon Sep 17 00:00:00 2001 From: Lex Berezhny <lex@damoti.com> Date: Mon, 3 Feb 2020 22:05:23 -0500 Subject: [PATCH 7/9] moved __init__ to top of class --- lbry/file_analysis.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/lbry/file_analysis.py b/lbry/file_analysis.py index 75d119647..baf62b499 100644 --- a/lbry/file_analysis.py +++ b/lbry/file_analysis.py @@ -14,6 +14,12 @@ log = logging.getLogger(__name__) class VideoFileAnalyzer: + def __init__(self, conf: TranscodeConfig): + self._conf = conf + self._available_encoders = "" + self._ffmpeg_installed = False + self._which = None + async def _execute(self, command, arguments): args = shlex.split(arguments) process = await asyncio.create_subprocess_exec(self._conf.ffmpeg_folder + command, *args, @@ -42,12 +48,6 @@ class VideoFileAnalyzer: self._ffmpeg_installed = True log.debug("Using %s at %s", version.splitlines()[0].split(" Copyright")[0], self._which) - def __init__(self, conf: TranscodeConfig): - self._conf = conf - self._available_encoders = "" - self._ffmpeg_installed = False - self._which = None - async def status(self, reset=False): if reset: self._available_encoders = "" From 153bdf576afc871ded19ad875819bd0d9fc4e228 Mon Sep 17 00:00:00 2001 From: Lex Berezhny <lex@damoti.com> Date: Mon, 3 Feb 2020 22:05:47 -0500 Subject: [PATCH 8/9] removed wheel from setup.py --- setup.py | 1 - 1 file changed, 1 deletion(-) diff --git a/setup.py b/setup.py index 5c60f9fa1..42ed2035f 100644 --- a/setup.py +++ b/setup.py @@ -33,7 +33,6 @@ setup( ], }, install_requires=[ - 'wheel', 'aiohttp==3.5.4', 'aioupnp==0.0.17', 'appdirs==1.4.3', From 3a3c63956a7d2794f53635b87a46852277c8a9e9 Mon Sep 17 00:00:00 2001 From: Lex Berezhny <lex@damoti.com> Date: Mon, 3 Feb 2020 22:06:15 -0500 Subject: [PATCH 9/9] removed unnecessary try/finally --- .../blockchain/test_transcoding.py | 21 +++++++------------ 1 file changed, 7 insertions(+), 14 deletions(-) diff --git a/tests/integration/blockchain/test_transcoding.py b/tests/integration/blockchain/test_transcoding.py index 3fec6cc33..47fe511c7 100644 --- a/tests/integration/blockchain/test_transcoding.py +++ b/tests/integration/blockchain/test_transcoding.py @@ -2,10 +2,9 @@ import logging import pathlib import time -import lbry.wallet # just to make the following line work: +from .test_claim_commands import ClaimTestCase from lbry.conf import TranscodeConfig from lbry.file_analysis import VideoFileAnalyzer -from .test_claim_commands import ClaimTestCase log = logging.getLogger(__name__) @@ -60,12 +59,9 @@ class TranscodeValidation(ClaimTestCase): self.assertEqual(self.video_file_webm, new_file_name) async def test_volume(self): - try: - self.conf.volume_analysis_time = 200 - with self.assertRaisesRegex(Exception, "lower than prime"): - await self.analyzer.verify_or_repair(True, False, self.video_file_name) - finally: - self.conf.volume_analysis_time = 0 + self.conf.volume_analysis_time = 200 + with self.assertRaisesRegex(Exception, "lower than prime"): + await self.analyzer.verify_or_repair(True, False, self.video_file_name) async def test_container(self): file_name = self.make_name("bad_container", ".avi") @@ -150,9 +146,6 @@ class TranscodeValidation(ClaimTestCase): self.assertEqual("ogg", extension) async def test_no_ffmpeg(self): - try: - self.conf.ffmpeg_folder = "I don't really exist/" - with self.assertRaisesRegex(Exception, "Unable to locate"): - await self.analyzer.verify_or_repair(True, False, self.video_file_name) - finally: - self.conf.ffmpeg_folder = "" + self.conf.ffmpeg_folder = "I don't really exist/" + with self.assertRaisesRegex(Exception, "Unable to locate"): + await self.analyzer.verify_or_repair(True, False, self.video_file_name)