"""Detect video location and dimensions within black border.""" from __future__ import annotations from dataclasses import dataclass import re import ffmpeg from video.connectors import s3 as s3_connector from video.constants import job_io_fields from video.logic.activity_task_logger import activity_task_logger from video.utils import text JOB_INPUTS = [ job_io_fields.INPUT_VIDEO_S3_BUCKET, job_io_fields.INPUT_VIDEO_S3_KEY, job_io_fields.VIDEO_STREAM_WIDTH_PIXELS, job_io_fields.VIDEO_STREAM_HEIGHT_PIXELS, ] class CropInfoNotFoundError(Exception): """Crop info not found error.""" pass @dataclass(frozen=True) class Crop: """Crop.""" CROP_TOLERANCE_PIXELS = 16 outer_width: int # offset of active area from left edge of frame left_offset: int inner_width: int @property def right_offset(self) -> int: """Offset of active area from right edge of frame.""" return Crop._get_opposite_offset( outer_dimension=self.outer_width, offset=self.left_offset, inner_dimension=self.inner_width, ) outer_height: int # offset of active area from top edge of frame top_offset: int inner_height: int @property def bottom_offset(self) -> int: """Offset of active area from bottom edge of frame.""" return Crop._get_opposite_offset( outer_dimension=self.outer_height, offset=self.top_offset, inner_dimension=self.inner_height, ) @staticmethod def _pick_tighter_crop_offset_within_tolerance( *, crop_offset: int, tighter_crop_offset: int, ) -> int: """Pick tighter crop offset within tolerance.""" if (tighter_crop_offset - crop_offset) <= Crop.CROP_TOLERANCE_PIXELS: return tighter_crop_offset return crop_offset @staticmethod def compute_tightened_crop_within_tolerance( *, looser_crop: Crop, tighter_crop: Crop, ) -> Crop: """Get tightened crop within tolerance.""" left_offset = Crop._pick_tighter_crop_offset_within_tolerance( crop_offset=looser_crop.left_offset, tighter_crop_offset=tighter_crop.left_offset, ) right_offset = Crop._pick_tighter_crop_offset_within_tolerance( crop_offset=looser_crop.right_offset, tighter_crop_offset=tighter_crop.right_offset, ) top_offset = Crop._pick_tighter_crop_offset_within_tolerance( crop_offset=looser_crop.top_offset, tighter_crop_offset=tighter_crop.top_offset, ) bottom_offset = Crop._pick_tighter_crop_offset_within_tolerance( crop_offset=looser_crop.bottom_offset, tighter_crop_offset=tighter_crop.bottom_offset, ) return Crop( outer_width=looser_crop.outer_width, left_offset=left_offset, inner_width=Crop._get_inner_dimension( outer_dimension=looser_crop.outer_width, start_offset=left_offset, end_offset=right_offset, ), outer_height=looser_crop.outer_height, top_offset=top_offset, inner_height=Crop._get_inner_dimension( outer_dimension=looser_crop.outer_height, start_offset=top_offset, end_offset=bottom_offset, ), ) @staticmethod def _compute_1d_symmetric_crop_within_tolerance( *, start_offset: int, inner_dimension: int, end_offset: int, ) -> tuple: """Compute 1D symmetric crop within tolerance. Returns: tuple: offset, inner_dimension """ offset_difference = abs(start_offset - end_offset) if offset_difference <= Crop.CROP_TOLERANCE_PIXELS: return start_offset, inner_dimension return ( min(start_offset, end_offset), inner_dimension + offset_difference, ) def get_symmetric_crop_within_tolerance(self) -> Crop: """Get symmetric crop within tolerance. We allow a bit of asymmetry to ensure we are able to crop out inactive pixels when the active image area is offset slightly from the center of the frame. """ left_offset, inner_width = ( Crop._compute_1d_symmetric_crop_within_tolerance( start_offset=self.left_offset, inner_dimension=self.inner_width, end_offset=self.right_offset, ) ) top_offset, inner_height = ( Crop._compute_1d_symmetric_crop_within_tolerance( start_offset=self.top_offset, inner_dimension=self.inner_height, end_offset=self.bottom_offset, ) ) return Crop( outer_width=self.outer_width, left_offset=left_offset, inner_width=inner_width, outer_height=self.outer_height, top_offset=top_offset, inner_height=inner_height, ) @staticmethod def _get_opposite_offset( *, outer_dimension: int, offset: int, inner_dimension: int, ): """Get opposite offset.""" return outer_dimension - (offset + inner_dimension) @staticmethod def _get_inner_dimension( *, outer_dimension: int, start_offset: int, end_offset: int, ) -> int: """Get inner dimension.""" return outer_dimension - (start_offset + end_offset) def cropdetect(*, video_url: str, limit: float) -> tuple[int, int, int, int]: """Ffmpeg cropdetect wrapper. Returns: tuple: width, height, x, y """ # cropdetect parameters # https://ffmpeg.org/ffmpeg-filters.html#cropdetect round_ = 2 reset = 0 output_options = { 'an': None, # don't process audio stream } (stdout, stderr) = ( ffmpeg .input(video_url) .filter( 'cropdetect', limit=limit, round=round_, reset=reset ) .output('pipe:', format='null', **output_options) .run(capture_stderr=True) ) cropping_info_regex = r'crop=(\d+):(\d+):(\d+):(\d+)' stderr_lines = text.bytes_to_lines_of_text(stderr) for line in reversed(stderr_lines): cropping_info_match = re.search(cropping_info_regex, line) if not cropping_info_match: continue width, height, x, y = cropping_info_match.groups() return int(width), int(height), int(x), int(y) raise CropInfoNotFoundError('No crop info found') def multi_pass_cropdetect( *, video_url: str, outer_width: int, outer_height: int, ) -> Crop: """Multi-pass cropdetect. Run cropdetect twice with different limits to detect the active image area. Lower limit prevents mistaking dark backgrounds as letterboxing or pillarboxing. Higher limit helps to detect inactive pixels at the interface between the active image area and the inactive image area that can be a bit brighter. To prevent cropping out dark backgrounds so we only use values resulting from the higher limit pass if they are within a certain tolerance of the values resulting from the lower limit pass. """ # Over the time this app has been deployed we have made adjustments to the # cropdetect limit to prevent cropping out dark backgrounds. We have found # that 16 / 255 works well. width, height, x, y = cropdetect(video_url=video_url, limit=16 / 255) looser_crop = Crop( outer_width=outer_width, left_offset=x, inner_width=width, outer_height=outer_height, top_offset=y, inner_height=height, ) # 24 / 255 is the default value for the limit parameter in cropdetect. width, height, x, y = cropdetect(video_url=video_url, limit=24 / 255) tighter_crop = Crop( outer_width=outer_width, left_offset=x, inner_width=width, outer_height=outer_height, top_offset=y, inner_height=height, ) return Crop.compute_tightened_crop_within_tolerance( looser_crop=looser_crop, tighter_crop=tighter_crop, ) @activity_task_logger(JOB_INPUTS) def detect_video_location_and_dimensions_within_black_border(inputs): """Detect video location and dimensions within black border. Args: inputs (dict): Inputs. Returns: dict: Outputs. """ s3_bucket = inputs[job_io_fields.INPUT_VIDEO_S3_BUCKET] s3_key = inputs[job_io_fields.INPUT_VIDEO_S3_KEY] outer_width = inputs[job_io_fields.VIDEO_STREAM_WIDTH_PIXELS] outer_height = inputs[job_io_fields.VIDEO_STREAM_HEIGHT_PIXELS] video_url = s3_connector.get_url_for_s3_object(s3_bucket, s3_key) try: multi_pass_cropdetect_result = multi_pass_cropdetect( video_url=video_url, outer_width=outer_width, outer_height=outer_height, ) except CropInfoNotFoundError: return { job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_X_PIXELS: None, job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_Y_PIXELS: None, job_io_fields.VIDEO_INNER_WIDTH_PIXELS: None, job_io_fields.VIDEO_INNER_HEIGHT_PIXELS: None, } # Some videos will have a dark background and position all of their content # closer to one of the edges of the frame. In cases like this, cropdetect # will return a crop that removes the dark background which has the effect # of centering the content. We want to keep the content in the same # position within the frame as it is in the original video, so we need to # adjust the crop to be symmetric. We allow a bit of shifting to ensure we # are able to crop out inactive pixels when the active image area is offset # slightly from the center of the frame but not so much that it results in # a noticeable shift in the position of the content. # See https://theorchard.atlassian.net/browse/DIS-1185 for such a case. symmetric_crop = ( multi_pass_cropdetect_result.get_symmetric_crop_within_tolerance() ) return { job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_X_PIXELS: symmetric_crop.left_offset, job_io_fields.VIDEO_INNER_TOP_LEFT_CORNER_Y_PIXELS: symmetric_crop.top_offset, job_io_fields.VIDEO_INNER_WIDTH_PIXELS: symmetric_crop.inner_width, job_io_fields.VIDEO_INNER_HEIGHT_PIXELS: symmetric_crop.inner_height, }