"""Validate analysis of video.""" from dataclasses import dataclass from decimal import Decimal import math from video.constants import job_io_fields from video.logic.activity_task_logger import activity_task_logger from video.models import ows_video from video.utils import maths INPUT_FIELDS = [ job_io_fields.VIDEO_INNER_WIDTH_PIXELS, job_io_fields.VIDEO_INNER_HEIGHT_PIXELS, job_io_fields.SILENT_INTERVAL_AT_BEGINNING_ENDS_AT_SECONDS, job_io_fields.SILENT_INTERVAL_AT_END_BEGINS_AT_SECONDS, job_io_fields.MAX_VOLUME_DECIBELS, job_io_fields.BLACK_INTERVAL_AT_BEGINNING_ENDS_AT_SECONDS, job_io_fields.BLACK_INTERVAL_AT_END_BEGINS_AT_SECONDS, job_io_fields.LEFT_AUDIO_CHANNEL_RMS_VOLUME_DECIBELS, job_io_fields.RIGHT_AUDIO_CHANNEL_RMS_VOLUME_DECIBELS, job_io_fields.VIDEO_STREAM_FRAME_RATE_FRAMES_PER_SECOND, job_io_fields.VIDEO_STREAM_PIXEL_ASPECT_RATIO, # Normally derived from the black/silent intervals, but read here so an # caller-supplied clip boundary (the re-encode trim override) overrides # the derived trim. job_io_fields.INPUT_CLIP_START_TIMECODE, job_io_fields.INPUT_CLIP_END_TIMECODE, ] @dataclass(frozen=True) class VideoResizingRule: """Video resizing rule.""" input_inner_width: int input_inner_height: int output_inner_width: int output_inner_height: int output_outer_width: int output_outer_height: int @dataclass(frozen=True) class SupportedOuterResolution: """Supported outer resolution.""" MAX_ALLOWED_DIMENSION_CHANGE_PERCENT = 4 width: int height: int def calculate_resolution_distance(self, width: int, height: int) -> float: return maths.euclidean_distance_2d(width, height, self.width, self.height) def calculate_aspect_ratio_distance(self, width: int, height: int) -> float: input_inner_aspect_ratio = width / height output_inner_aspect_ratio = self.width / self.height return abs(input_inner_aspect_ratio - output_inner_aspect_ratio) def calculate_inner_resolution(self, width, height): full_width_scaling_factor = ( self.width / width ) full_height_scaling_factor = ( self.height / height ) should_scale_to_full_width = ( full_width_scaling_factor < full_height_scaling_factor ) if should_scale_to_full_width: return ( self.width, maths.find_multiple_of_n_nearest_x( 2, height * full_width_scaling_factor, ), ) return ( maths.find_multiple_of_n_nearest_x( 2, width * full_height_scaling_factor, ), self.height, ) def is_resolution_outside_scaling_limits(self, width: int, height: int) -> bool: inner_width, inner_height = self.calculate_inner_resolution(width, height) width_change_percent = abs(maths.percent_change(width, inner_width)) height_change_percent = abs(maths.percent_change(height, inner_height)) return not ( width_change_percent <= SupportedOuterResolution.MAX_ALLOWED_DIMENSION_CHANGE_PERCENT and height_change_percent <= SupportedOuterResolution.MAX_ALLOWED_DIMENSION_CHANGE_PERCENT ) SECONDS_PER_MINUTE = 60 MINUTES_PER_HOUR = 60 SECONDS_PER_HOUR = SECONDS_PER_MINUTE * MINUTES_PER_HOUR def seconds_to_timecode(seconds, frames_per_second): """Convert seconds to timecode. Args: seconds (float): Seconds. frames_per_second (float): Frames per second. Returns: dict: Output """ hh = math.floor(seconds / SECONDS_PER_HOUR) mm = math.floor( (seconds - (hh * SECONDS_PER_HOUR)) / SECONDS_PER_MINUTE) ss = math.floor( seconds - (hh * SECONDS_PER_HOUR) - (mm * SECONDS_PER_MINUTE)) ff = math.floor( frames_per_second * ( seconds - ( hh * SECONDS_PER_HOUR) - (mm * SECONDS_PER_MINUTE) - ss)) return '{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}'.format( hh=hh, mm=mm, ss=ss, ff=ff) def map_to_supported_resolution(inputs): """Map to supported resolution. Args: inputs (dict): Inputs. """ input_video_inner_width_pixels = inputs[job_io_fields.VIDEO_INNER_WIDTH_PIXELS] input_video_pixel_aspect_ratio = inputs[job_io_fields.VIDEO_STREAM_PIXEL_ASPECT_RATIO] # Rounding to nearest even number to avoid odd resolutions resulting from # scaling by pixel aspect ratio to get the equivalent width with square # pixels. input_video_inner_width_pixels = maths.find_multiple_of_n_nearest_x( 2, input_video_inner_width_pixels * input_video_pixel_aspect_ratio ) input_video_inner_height_pixels = inputs[job_io_fields.VIDEO_INNER_HEIGHT_PIXELS] video_resizing_rules = get_video_resizing_rules() for video_resizing_rule in video_resizing_rules: if ( video_resizing_rule.input_inner_width == input_video_inner_width_pixels and video_resizing_rule.input_inner_height == input_video_inner_height_pixels ): return { job_io_fields.VIDEO_OUTPUT_INNER_WIDTH_PIXELS: video_resizing_rule.output_inner_width, job_io_fields.VIDEO_OUTPUT_OUTER_WIDTH_PIXELS: video_resizing_rule.output_outer_width, job_io_fields.VIDEO_OUTPUT_INNER_HEIGHT_PIXELS: video_resizing_rule.output_inner_height, job_io_fields.VIDEO_OUTPUT_OUTER_HEIGHT_PIXELS: video_resizing_rule.output_outer_height, } supported_outer_resolutions = get_supported_outer_resolutions() closest_supported_outer_resolution = sorted( supported_outer_resolutions, key=lambda supported_outer_resolution: ( supported_outer_resolution.is_resolution_outside_scaling_limits( input_video_inner_width_pixels, input_video_inner_height_pixels ), supported_outer_resolution.calculate_resolution_distance( input_video_inner_width_pixels, input_video_inner_height_pixels), supported_outer_resolution.calculate_aspect_ratio_distance( input_video_inner_width_pixels, input_video_inner_height_pixels), ) )[0] if closest_supported_outer_resolution.is_resolution_outside_scaling_limits( input_video_inner_width_pixels, input_video_inner_height_pixels, ): return { job_io_fields.ERROR_CANNOT_RESIZE_TO_STANDARD_RESOLUTION: 'Cannot convert video to a standard resolution.' } ( video_output_inner_width_pixels, video_output_inner_height_pixels, ) = closest_supported_outer_resolution.calculate_inner_resolution( input_video_inner_width_pixels, input_video_inner_height_pixels, ) return { job_io_fields.VIDEO_OUTPUT_INNER_WIDTH_PIXELS: video_output_inner_width_pixels, job_io_fields.VIDEO_OUTPUT_OUTER_WIDTH_PIXELS: closest_supported_outer_resolution.width, job_io_fields.VIDEO_OUTPUT_INNER_HEIGHT_PIXELS: video_output_inner_height_pixels, job_io_fields.VIDEO_OUTPUT_OUTER_HEIGHT_PIXELS: closest_supported_outer_resolution.height, } def calculate_volume_adjustment(inputs): """Calculate volume adjustment. MediaConvert can adjust volume by -60 dB through +6 dB. Args: inputs (dict): Inputs. """ target_volume_db = -2 audio_volume_adjustment_db = float( target_volume_db - Decimal(str(inputs[job_io_fields.MAX_VOLUME_DECIBELS]))) if audio_volume_adjustment_db > 6: return { job_io_fields.ERROR_VOLUME_TOO_LOW: 'Volume level must be at least -8 dB.' } return { job_io_fields.AUDIO_VOLUME_ADJUSTMENT_DB: audio_volume_adjustment_db } def calculate_trimming(inputs): """Calculate trimming. The clip is derived from where the leading and trailing black-and-silent dead zones end and begin. A caller-supplied clip boundary (the re-encode trim override) takes precedence per end. Args: inputs (dict): Inputs. """ frame_rate = inputs[ job_io_fields.VIDEO_STREAM_FRAME_RATE_FRAMES_PER_SECOND] if job_io_fields.INPUT_CLIP_START_TIMECODE in inputs: input_clip_start_time = inputs[ job_io_fields.INPUT_CLIP_START_TIMECODE] else: video_start_time = inputs[ job_io_fields.BLACK_INTERVAL_AT_BEGINNING_ENDS_AT_SECONDS] audio_start_time = inputs[ job_io_fields.SILENT_INTERVAL_AT_BEGINNING_ENDS_AT_SECONDS] input_clip_start_time = None if video_start_time and audio_start_time: input_clip_start_time = seconds_to_timecode( Decimal(str(min(video_start_time, audio_start_time))), Decimal(str(frame_rate))) if job_io_fields.INPUT_CLIP_END_TIMECODE in inputs: input_clip_end_time = inputs[ job_io_fields.INPUT_CLIP_END_TIMECODE] else: video_end_time = inputs[ job_io_fields.BLACK_INTERVAL_AT_END_BEGINS_AT_SECONDS] audio_end_time = inputs[ job_io_fields.SILENT_INTERVAL_AT_END_BEGINS_AT_SECONDS] input_clip_end_time = None if video_end_time and audio_end_time: input_clip_end_time = seconds_to_timecode( Decimal(str(max(video_end_time, audio_end_time))), Decimal(str(frame_rate))) return { job_io_fields.INPUT_CLIP_START_TIMECODE: input_clip_start_time, job_io_fields.INPUT_CLIP_END_TIMECODE: input_clip_end_time } def check_for_dual_mono_audio(inputs): """Check for dual mono audio. Args: inputs (dict): Inputs. """ if (inputs[job_io_fields.LEFT_AUDIO_CHANNEL_RMS_VOLUME_DECIBELS] == ( inputs[job_io_fields.RIGHT_AUDIO_CHANNEL_RMS_VOLUME_DECIBELS])): return { job_io_fields.ERROR_DUAL_MONO_AUDIO: 'Left and right audio channels must be different.' } return {} @activity_task_logger(INPUT_FIELDS) def validate_analysis(inputs): """Validate the analysis step in video pipeline. Args: inputs (dict): Inputs. Returns: dict: Outputs """ return { **map_to_supported_resolution(inputs), **calculate_volume_adjustment(inputs), **calculate_trimming(inputs), **check_for_dual_mono_audio(inputs), } def get_video_resizing_rules(): """Get video resizing rules.""" video_resizing_rules = ows_video.get_video_resizing_rules() return [ VideoResizingRule( input_inner_width=video_resizing_rule["input_inner_width"], input_inner_height=video_resizing_rule["input_inner_height"], output_inner_width=video_resizing_rule["output_inner_width"], output_inner_height=video_resizing_rule["output_inner_height"], output_outer_width=video_resizing_rule["output_outer_width"], output_outer_height=video_resizing_rule["output_outer_height"], ) for video_resizing_rule in video_resizing_rules ] def get_supported_outer_resolutions() -> [SupportedOuterResolution]: """Get supported outer resolutions.""" supported_outer_resolutions = ows_video.get_supported_outer_resolutions() return [ SupportedOuterResolution( width=supported_outer_resolution[1], height=supported_outer_resolution[2], ) for supported_outer_resolution in supported_outer_resolutions ]