From 87b557bc3883aa34bef81704f8ac5d416ee45ec2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fr=C3=A9d=C3=A9ric=20Tronel?= Date: Sat, 29 Aug 2026 11:24:50 +0200 Subject: [PATCH] Add functions related to ffmpeg, ppm files and timestamps to dedicated modules. --- src/tscut/tools/ffmpeg.py | 412 +++++++++++++++++ src/tscut/tools/ffprobe.py | 181 ++++++++ src/tscut/tools/ppm.py | 83 ++++ src/tscut/tools/timeframe.py | 222 +++++++++ src/tscut/tscut.py | 857 +---------------------------------- 5 files changed, 910 insertions(+), 845 deletions(-) create mode 100644 src/tscut/tools/ffmpeg.py create mode 100644 src/tscut/tools/ppm.py create mode 100644 src/tscut/tools/timeframe.py diff --git a/src/tscut/tools/ffmpeg.py b/src/tscut/tools/ffmpeg.py new file mode 100644 index 0000000..97b3586 --- /dev/null +++ b/src/tscut/tools/ffmpeg.py @@ -0,0 +1,412 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2026 Frédéric Tronel + +import logging +from math import floor, ceil, log +from datetime import timedelta +from os import ( + read, + SEEK_SET, + lseek, + memfd_create, + set_inheritable, + write, + close +) +from typing import IO +from subprocess import PIPE, Popen +from io import TextIOWrapper + +from typeguard import typechecked +from tqdm import tqdm + +from tscut.tools.ffprobe import ( + get_video_dimensions, + with_subtitles, + get_frames_in_stream +) +from tscut.tools.ppm import dump_ppm +from tscut.tools.timeframe import parse_timestamp, get_packet_duration + +logger = logging.getLogger(__name__) + + +@typechecked +def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], input_format:str, + output_file: IO[bytes], output_format:str, duration: timedelta): + width, height = get_video_dimensions(ffprobe_path, input_file) + subtitles = with_subtitles(ffprobe_path, input_file) + + infd = input_file.fileno() + outfd = output_file.fileno() + set_inheritable(infd, True) + set_inheritable(outfd, True) + + if logger.getEffectiveLevel() == logging.DEBUG: + log_level = [] + else: + log_level = [ '-loglevel', 'quiet' ] + + params = [ffmpeg_path, '-y',]+log_level+['-progress', '/dev/stdout', '-canvas_size', + f'{width:d}x{height:d}', '-f', input_format, + '-i', f'/proc/self/fd/{infd:d}', '-map', '0:v', + '-map', '0:a'] + if subtitles: + params.extend(['-map', '0:s']) + params.extend(['-bsf:v', 'h264_mp4toannexb,dump_extra=freq=keyframe', '-vcodec', 'copy', + '-acodec', 'copy']) + if subtitles: + params.extend(['-scodec', 'dvdsub']) + params.extend(['-r:0', '25', '-f', output_format, f'/proc/self/fd/{outfd:d}']) + + logger.debug('Executing %s', params) + + with Popen(params, stdout=PIPE, close_fds=False) as ffmpeg: + pb = tqdm(TextIOWrapper(ffmpeg.stdout, encoding="utf-8"), + total=int(duration/timedelta(seconds=1)), unit='s', desc='Conversion') + for line in pb: + if line.startswith('out_time='): + ts = line.split('=')[1].strip() + ts = parse_timestamp(ts) + if ts is not None: + pb.n = int(ts/timedelta(seconds=1)) + pb.update() + status = ffmpeg.wait() + if status != 0: + logger.error('Conversion failed with status code: %d', status) + + +@typechecked +def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_frames:int, + width:int=640, height:int=480) -> tuple[bytes,int]|tuple[None,None]: + """ + Extract pictures from a video file using FFmpeg. + + This function runs the FFmpeg binary to extract a specified number of frames from a video file, + starting at a given time. + The extracted frames are stored in memory as PPM images and returned as a tuple containing + the image data and a file descriptor to the memory created by memfd_create. + + Args: + ffmpeg_path (str): The path to the FFmpeg binary. + input_file (IO[bytes]): The input video file. + begin (timedelta): The start time of the extraction. + nb_frames (int): The number of frames to extract. + width (int, optional): The width of the extracted images. Defaults to 640. + height (int, optional): The height of the extracted images. Defaults to 480. + + Returns: + tuple[bytes, int] | tuple[None, None]: + - A tuple containing the extracted image data as bytes and a file descriptor + - A tuple containing None, None if the extraction fails + """ + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + outfd = memfd_create('pictures', flags=0) + set_inheritable(outfd, True) + # PPM header + # "P6\nWIDTH HEIGHT\n255\n" + header_len=2+1+ceil(log(width, 10))+1+ceil(log(height, 10))+1+3+1 + logger.debug('Header length: %d', header_len) + image_length = width*height*3+header_len + length = image_length*nb_frames + logger.debug("Estimated length: %d", length) + + command = [ffmpeg_path, '-loglevel', 'quiet' ,'-y', '-ss', f'{begin}', '-i', + f'/proc/self/fd/{infd}', '-s', f'{width:d}x{height:d}', '-vframes', f'{nb_frames:d}', + '-c:v', 'ppm','-f', 'image2pipe', f'/proc/self/fd/{outfd:d}'] + logger.debug('Executing: %s', command) + + images = b'' + with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg: + status = ffmpeg.wait() + if status != 0: + logger.error('Conversion failed with status code: %d', status) + return None, None + + lseek(outfd, 0, SEEK_SET) + images = read(outfd,length) + if len(images) != length: + logger.error("Received %d bytes but %d were expected.", len(images), length) + return None, None + + lseek(outfd, 0, SEEK_SET) + return images, outfd + +@typechecked +def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, output_filename:str, + packet_duration:int, sub_channel:int=0, + nb_packets:int=0, sample_rate:int=48000, + nb_channels:int=2) -> tuple[bytes,int]|tuple[None,None]: + outfd = memfd_create(output_filename, flags=0) + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + set_inheritable(outfd, True) + sound = b'' + length = int(nb_channels*sample_rate*4*nb_packets*packet_duration/1000) + + command = [ffmpeg_path, '-y', '-loglevel', 'quiet', '-ss', f'{begin}', + '-i', f'/proc/self/fd/{infd}', f'-frames:a:{sub_channel:d}', f'{nb_packets+1:d}', + '-c:a', 'pcm_s32le', '-sample_rate', f'{sample_rate:d}', + '-channels', f'{nb_channels:d}', '-f', 's32le', f'/proc/self/fd/{outfd:d}'] + logger.debug('Executing: %s', command) + + with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg: + status = ffmpeg.wait() + if status != 0: + logger.error('Sound extraction returns error code: %d', status) + return None, None + + lseek(outfd, 0, SEEK_SET) + sound = read(outfd, length) + + if len(sound) != length: + logger.info("Received %d bytes but %d were expected (channels=%d, freq=%d, packets=%d,\ + duration=%d ms).", len(sound), length, nb_channels, sample_rate, nb_packets, + packet_duration) + return None, None + + return sound, outfd + +@typechecked +def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], begin:timedelta, + end:timedelta, streams, files_prefix, nb_frames:int, framerate:float, + width:int, height:int, temporaries, dump_mem_fd:bool=False): + # The command line for encoding only video track + video_encoder_params = [ ffmpeg_path, '-y', '-loglevel', 'quiet'] + video_input_params = [] + video_codec_params = [] + + # The command line to create a MKV file with the rest of tracks + generic_encoder_params = [ ffmpeg_path, '-y', '-loglevel', 'quiet' ] + generic_input_params = [] + generic_codec_params = [] + + if begin < end: + video_id=0 + audio_id=0 + subtitle_id=0 + memfds = [] + for stream in streams: + if stream['codec_type'] == 'video': + logger.info("Extracting %d frames of video stream v:%d", nb_frames, video_id) + sar = stream['sample_aspect_ratio'] + dar = stream['display_aspect_ratio'] + pixel_format = stream['pix_fmt'] + color_range = stream['color_range'] + color_space =stream['color_space'] + color_transfer = stream['color_transfer'] + color_primaries = stream['color_primaries'] + level = int(stream['level']) + level = f'{floor(level/10):d}.{level%10:d}' + chroma_location = stream['chroma_location'] + field_order = stream + match field_order: + case 'progressive': + interlaced_options = ['-field_order', '0'] + case 'tt': + interlaced_options = ['-top', '1', f'-flags:v:{video_id:d}', '+ilme+ildct', + '-field_order', '1'] + case 'bb': + interlaced_options = ['-top', '0', f'-flags:v:{video_id:d}', '+ilme+ildct', + '-field_order','2'] + case 'tb': + interlaced_options = ['-top', '1', f'-flags:v:{video_id:d}', '+ilme+ildct', + '-field_order', '3'] + case 'bt': + interlaced_options = ['-top', '0', f'-flags:v:{video_id:d}', '+ilme+ildct', + '-field_order', '4'] + case _: + interlaced_options = [] + + # ======================================= # + # TODO: adjust SAR and DAR + # https://superuser.com/questions/907933/correct-aspect-ratio-without-re-encoding-video-file + # SAR: -aspect width:height + # DAR: -bsf:v sample_aspect_ratio=1:video_format + logger.warning('Missing SAR adjustment for: %s', sar) + logger.warning('Missing DAR adjustment for: %s', dar) + logger.warning('Missing treatment for chroma location: %s', chroma_location) + codec = stream['codec_name'] + images_bytes, memfd = extract_pictures(ffmpeg_path, input_file=input_file, + begin=begin, nb_frames=nb_frames, + width=width, height=height) + if images_bytes is None: + logger.error('Impossible to extract picture from video stream.') + exit(-1) + + memfds.append(memfd) + if dump_mem_fd: + dump_ppm(images_bytes, f'{files_prefix}-{video_id:d}', temporaries) + + # We rewind to zero the memory file descriptor + lseek(memfd, 0, SEEK_SET) + set_inheritable(memfd, True) + + video_input_params.extend(['-framerate', f'{framerate:f}', '-f', 'image2pipe', '-i', + f'/proc/self/fd/{memfd:d}']) + video_codec_params.extend([f'-c:v:{video_id:d}', codec, f'-level:v:{video_id:d}', + level, '-pix_fmt', pixel_format]) + video_codec_params.extend(interlaced_options) + video_codec_params.extend([f'-colorspace:v:{video_id}', color_space, + f'-color_primaries:v:{video_id:d}', color_primaries, + f'-color_trc:v:{video_id:d}', color_transfer, + f'-color_range:v:{video_id:d}', color_range]) + video_id=video_id+1 + elif stream['codec_type'] == 'audio': + logger.debug('Audio stream: %s', stream) + sample_rate = int(stream['sample_rate']) + nb_channels = int(stream['channels']) + if 'bit_rate' in stream: + bit_rate = int(stream['bit_rate']) + else: + bit_rate = 128000 + codec = stream['codec_name'] + if 'tags' in stream: + if 'language' in stream['tags']: + generic_codec_params.extend([f'-metadata:s:a:{audio_id:d}', + f"language={stream['tags']['language']}"]) + packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin, + end=end, stream_kind='a', sub_stream_id=audio_id) + nb_packets = len(packets) + logger.debug("Found %d packets to be extracted from audio track.", nb_packets) + if nb_packets > 0: + packet_duration = get_packet_duration(packets[0]) + if packet_duration is None: + return None + else: + packet_duration = 0 + + logger.info("Extracting %d packets of audio stream: a:%d" , nb_packets, audio_id) + tmpname = f'{files_prefix}-{audio_id:d}.pcm' + + sound_bytes, memfd = extract_sound(ffmpeg_path=ffmpeg_path, input_file=input_file, + begin=begin, nb_packets=nb_packets, + packet_duration=packet_duration, + output_filename=tmpname, + sample_rate=sample_rate, nb_channels=nb_channels) + + if sound_bytes is None: + logger.error('Impossible to extract sound track') + exit(-1) + + memfds.append(memfd) + + if dump_mem_fd: + try: + with open(tmpname,'wb') as output: + temporaries.append(output) + outfd = output.fileno() + pos = 0 + while pos < len(sound_bytes): + pos+=write(outfd, sound_bytes[pos:]) + except OSError: + logger.error('Impossible to create file: %s', tmpname) + return None + + # We rewind to zero the memory file descriptor + lseek(memfd, 0, SEEK_SET) + set_inheritable(memfd, True) + + generic_input_params.extend(['-f', 's32le', '-ar', f'{sample_rate:d}', '-ac', + f'{nb_channels:d}', '-i', f'/proc/self/fd/{memfd:d}']) + generic_codec_params.extend([f'-c:a:{audio_id:d}', codec, f'-b:a:{audio_id:d}', + f'{bit_rate:d}']) + audio_id=audio_id+1 + elif stream['codec_type'] == 'subtitle': + logger.info("Extracting a subtitle stream: s:%d", subtitle_id) + codec = stream['codec_name'] + generic_input_params.extend(['-i', './empty.idx']) + if 'tags' in stream: + if 'language' in stream['tags']: + generic_codec_params.extend([f'-metadata:s:s:{subtitle_id:d}', + f"language={stream['tags']['language']}"]) + generic_codec_params.extend([f'-c:s:{subtitle_id:d}', 'copy']) + subtitle_id=subtitle_id+1 + else: + logger.error("Unknown stream type: %s", stream['codec_type']) + + # Create a new MKV movie with all streams (except videos) that have been extracted. + generic_encoder_params.extend(generic_input_params) + + for index in range(audio_id+subtitle_id): + generic_encoder_params.extend(['-map', f'{index:d}']) + generic_encoder_params.extend(generic_codec_params) + + mkv_filename = f'{files_prefix}.mkv' + try: + mkv_output = open(mkv_filename,'wb+') + except OSError: + logger.error('Impossible to create file: %s', mkv_filename) + return None + + mkvoutfd = mkv_output.fileno() + set_inheritable(mkvoutfd, True) + generic_encoder_params.extend(['-f', 'matroska', f'/proc/self/fd/{mkvoutfd:d}']) + + logger.info('Encoding all streams (except video) into a MKV file: %s', mkv_filename) + logger.debug('Executing: %s', generic_encoder_params) + with Popen(generic_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg: + status = ffmpeg.wait() + if status != 0: + logger.error('Encoding failed with status code: %d', status) + return None + + temporaries.append(mkv_output) + + h264_filename = f'{files_prefix}.h264' + try: + h264_output = open(h264_filename,'wb+') + except OSError: + logger.error('Impossible to create file: %s', h264_filename) + return None + + h264outfd = h264_output.fileno() + set_inheritable(h264outfd, True) + + video_encoder_params.extend(video_input_params) + video_encoder_params.extend(video_codec_params) + + video_encoder_params.extend([ '-x264opts', f'keyint=1:sps-id={1:d}','-bsf:v', + 'h264_mp4toannexb,dump_extra=freq=keyframe,h264_metadata=\ + overscan_appropriate_flag=1:sample_aspect_ratio=1:video_format=\ + 0:chroma_sample_loc_type=0','-f', 'h264', + f'/proc/self/fd/{h264outfd:d}']) + + logger.info('Encoding video into a H264 file: %s', h264_filename) + logger.debug('Executing: %s', video_encoder_params) + with Popen(video_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg: + status = ffmpeg.wait() + if status != 0: + logger.error('Encoding failed with status code: %d', status) + return None + + temporaries.append(h264_output) + + h264_ts_filename = f'{files_prefix}-ts.txt' + try: + h264_ts_output = open(h264_ts_filename,'w+', encoding='utf8') + except OSError: + logger.error('Impossible to create file: %s', h264_ts_filename) + return None + + h264_ts_output.write('# timestamp format v2\n') + ts = 0 + for _ in range(nb_frames): + ts = ts+ceil(1000/framerate) + h264_ts_output.write(f'{ts:d}\n') + h264_ts_output.flush() + h264_ts_output.seek(0) + + temporaries.append(h264_ts_output) + + for memfd in memfds: + close(memfd) + + return h264_output, h264_ts_output, mkv_output + + # Nothing to be done. We are already at a i-frame boundary. + return None, None + diff --git a/src/tscut/tools/ffprobe.py b/src/tscut/tools/ffprobe.py index 04bd759..c5912a3 100644 --- a/src/tscut/tools/ffprobe.py +++ b/src/tscut/tools/ffprobe.py @@ -18,6 +18,8 @@ from io import BytesIO from typeguard import typechecked +from tscut.tools.timeframe import get_ts_frame + logger = logging.getLogger(__name__) @@ -240,3 +242,182 @@ def with_subtitles(ffprobe_path:str, input_file: IO[bytes]) -> bool: logger.error('Impossible to retrieve streams inside file') return False + +def find_subtitles_tracks(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + + command = [ffprobe_path, '-loglevel','quiet', '-i', f'/proc/self/fd/{infd:d}', + '-select_streams', 's', '-show_entries', 'stream=index:stream_tags=language', + '-of', 'json'] + logger.debug('Executing: %s', command) + + with Popen(command, stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'streams' in out: + return out['streams'] + logger.error('Impossible to retrieve format of file') + + ffprobe.wait() + return None + +@typechecked +def get_frames_in_stream(ffprobe_path: str, input_file: IO[bytes], begin:timedelta, end:timedelta, + stream_kind:str, sub_stream_id:int=0) -> list[dict]|None: + infd = input_file.fileno() + set_inheritable(infd, True) + + command = [ffprobe_path, '-loglevel', 'quiet', '-read_intervals', f'{begin}%{end}', + '-show_entries', 'frame', '-select_streams', + f'{stream_kind}:{sub_stream_id:d}','-of', 'json', f'/proc/self/fd/{infd:d}'] + logger.debug('Executing: %s', command) + + with Popen(command, stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + frames = json.load(BytesIO(out)) + status = ffprobe.wait() + if status != 0: + logger.error('ffprobe failed with status code: %d', status) + return None + + # Sort frames by timestamp + tmp = {} + if 'frames' in frames: + frames = frames['frames'] + for frame in frames: + ts = get_ts_frame(frame) + if ts is None: + return None + if begin <= ts <= end: + tmp[ts]=frame + + res = [] + for ts in sorted(tmp): + res.append(tmp[ts]) + return res + + logger.error('Impossible to retrieve frames inside file around [%s,%s]', begin, end) + return None + +# TODO: Finish implementation of this function and use it. +@typechecked +def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:timedelta, + before: bool=True, delta: timedelta=timedelta(seconds=2)): + zero = timedelta() + tbegin = timestamp-delta + tend = timestamp+delta + tbegin = max(tbegin, zero) + + infd = input_file.fileno() + set_inheritable(infd, True) + + logger.debug('Looking for IDR frame in [%s, %s]', tbegin, tend) + + idrs = [] + + # Retains only IDR frame + with Popen([ffprobe_path, '-loglevel', 'quiet', '-read_intervals', f'{tbegin}%{tend}', + '-skip_frame', 'nokey', '-show_entries', 'frame', '-select_streams', 'v:0', + '-of', 'json', f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + frames = json.load(BytesIO(out)) + status = ffprobe.wait() + if status != 0: + logger.error('ffprobe failed with status code: %d', status) + return None + + if 'frames' in frames: + frames = frames['frames'] + for frame in frames: + ts = get_ts_frame(frame) + if ts is None: + return None + if tbegin <= ts <= tend: + idrs.append(frame) + else: + logger.error('Impossible to retrieve IDR frames inside file around [%s,%s]', + tbegin, tend) + return None + + return None + +@typechecked +def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], + timestamp:timedelta, before:bool=True, + delta_max:timedelta=timedelta(seconds=15))-> tuple[int,dict]: + infd = input_file.fileno() + set_inheritable(infd, True) + + delta = timedelta(seconds=1) + + iframe = None + while delta < delta_max: + zero = timedelta() + if before: + tbegin = timestamp-delta + else: + tbegin = timestamp + if not before: + tend = timestamp+delta + else: + tend = timestamp + tbegin = max(tbegin, zero) + logger.debug('Looking for an iframe in [%s, %s]', tbegin, tend) + + frames = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=tbegin, end=tend, + stream_kind='v') + if frames is None: + logger.debug('Found no frame in [%s, %s]', tbegin, tend) + delta+=timedelta(seconds=1) + continue + + iframes = [] + for frame in frames: + if frame['pict_type'] == 'I': + iframes.append(frame) + + found = False + for frame in iframes: + ts = get_ts_frame(frame) + if ts is None: + logger.warning('I-frame with no timestamp: %s', frame) + continue + + if before and ts <= timestamp: + found = True + iframe = frame + if not before and ts >= timestamp: + found = True + iframe = frame + break + + if found: + logger.info("Found i-frame at: %s", iframe) + break + + delta+=timedelta(seconds=1) + + if iframe is not None: + its = get_ts_frame(iframe) + nb_frames = 0 + for frame in frames: + ts = get_ts_frame(frame) + if ts is None: + logger.warning('Frame without timestamp: %s', frame) + continue + + if before: + if its <= ts <= timestamp: + logger.info("Retrieve a frame between %s and %s at %s", its, timestamp, ts) + nb_frames = nb_frames+1 + else: + if timestamp <= ts <= its: + logger.info("Retrieve a frame between %s and %s at %s", timestamp, ts, its) + nb_frames = nb_frames+1 + else: + logger.error("Impossible to find I-frame between: %s and %s", tbegin, tend) + return 0, None + + return(nb_frames, iframe) diff --git a/src/tscut/tools/ppm.py b/src/tscut/tools/ppm.py new file mode 100644 index 0000000..62d6800 --- /dev/null +++ b/src/tscut/tools/ppm.py @@ -0,0 +1,83 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2026 Frédéric Tronel + + +import logging +import re +from math import ceil, log +from os import write +from typing import IO +from io import BytesIO + + +from typeguard import typechecked + +@typechecked +def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None: + """ + Dump PPM pictures from a bytes buffer to files. + + This function takes a bytes buffer containing PPM pictures, a prefix for the output file names, + and a list of temporary files. + It extracts each PPM picture from the buffer, checks its validity, and writes it to a file. + The output files are named according to the prefix and a zero-padded three-digit number. + + Args: + pictures (bytes): The bytes buffer containing the PPM pictures. + prefix (str): The prefix for the output file names. + temporaries (list[IO[bytes]]): A list of temporary files that will be used to store + the output files. + + Returns: + None + + Raises: + None, but logs errors if: + - the PPM picture is not valid (e.g. wrong magic number, dimensions, or color encoding) + - an I/O error occurs while creating or writing to an output file + """ + logger = logging.getLogger(__name__) + + # "P6\nWIDTH HEIGHT\n255\n" + pos = 0 + picture = 0 + + logger.debug('Dumping %d pictures: %s', len(pictures),prefix) + + while pos[0-9]+) (?P[0-9]+)\n$') + m = pattern.match(dimensions) + if m is not None: + width = int(m['width']) + height = int(m['height']) + else: + logger.error('Impossible to parse dimensions of picture') + return + else: + logger.error('Not a PPM picture') + return + + if max_value != 255: + logger.error('Not a valid PPM picture. Color are not encoded on byte. Max value: %d', + max_value) + + header_len=2+1+ceil(log(width, 10))+1+ceil(log(height, 10))+1+3+1 + try: + with open(filename, 'wb') as out: + temporaries.append(out) + outfd = out.fileno() + length=header_len+3*width*height + nb_bytes = 0 + while nb_bytes < length: + nb_bytes+=write(outfd, pictures[pos+nb_bytes:pos+length]) + pos+=length + picture+=1 + except OSError: + logger.error('Impossible to create file: %s', filename) diff --git a/src/tscut/tools/timeframe.py b/src/tscut/tools/timeframe.py new file mode 100644 index 0000000..08bf7c9 --- /dev/null +++ b/src/tscut/tools/timeframe.py @@ -0,0 +1,222 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2026 Frédéric Tronel + + +import logging +import re +from datetime import timedelta + +from typeguard import typechecked + + +@typechecked +def parse_timestamp(ts:str) -> timedelta|None: + """ + Parse a timestamp string into a timedelta object. + + This function takes a string representing a timestamp in the format HH:MM:SS[.us] and returns + a timedelta object representing the corresponding time interval. + The timestamp string can have an optional microsecond component. + + Args: + ts (str): The timestamp string to parse. + + Returns: + timedelta | None: + - A timedelta object representing the parsed timestamp + - None if: + - the timestamp string is not in the correct format + - the timestamp values are out of range (e.g. hour > 23, minute > 59, etc.) + """ + logger = logging.getLogger(__name__) + + ts_reg_exp = (r'^(?P[0-9]{1,2}):(?P[0-9]{1,2})' + r':(?P[0-9]{1,2})(\.(?P[0-9]{1,6}))?$') + p = re.compile(ts_reg_exp) + m = p.match(ts) + if m is None: + logger.warning("Impossible to parse timestamp: %s", ts) + return None + + values = m.groupdict() + hour = 0 + minute = 0 + second = 0 + us = 0 + if values['hour'] is not None: + hour = int(values['hour']) + if values['minute'] is not None: + minute = int(values['minute']) + if values['second'] is not None: + second = int(values['second']) + if values['us'] is not None: + us = int(values['us']) + + if hour < 0 or hour > 23: + logger.error("hour must be in [0,24[") + return None + if minute < 0 or minute > 59: + logger.error("minute must be in [0,60[") + return None + if second < 0 or second > 59: + logger.error("second must be in [0,60[") + return None + if us < 0 or us > 1000000: + logger.error("milliseconds must be in [0,1000000[") + return None + res = timedelta(hours=hour, minutes=minute, seconds=second, microseconds=us) + + return res + +@typechecked +def parse_time_interval(interval: str) -> tuple[timedelta, timedelta] | None: + """ + Parse a time interval string into a tuple of two timedelta objects. + + This function takes a string representing a time interval in the format + HH:MM:SS[.ms]-HH:MM:SS[.ms] and returns a tuple of two timedelta objects representing + the start and end times of the interval. + The time interval string can have an optional millisecond component. + + Args: + interval (str): The time interval string to parse. + + Returns: + tuple[timedelta, timedelta] | None: + - A tuple of two timedelta objects representing the start and end times of the interval + - None if: + - the time interval string is not in the correct format + - the time values are out of range (e.g. hour > 23, minute > 59, etc.) + - the end time is before the start time (non-monotonic interval) + """ + logger = logging.getLogger(__name__) + + interval_reg_exp = (r'^(?P[0-9]{1,2}):(?P[0-9]{1,2}):(?P[0-9]{1,2})' + r'(\.(?P[0-9]{1,3}))?-(?P[0-9]{1,2}):(?P[0-9]{1,2})' + r':(?P[0-9]{1,2})(\.(?P[0-9]{1,3}))?$') + p = re.compile(interval_reg_exp) + m = p.match(interval) + if m is None: + logger.error("Impossible to parse time interval") + return None + + values = m.groupdict() + hour1 = 0 + minute1 = 0 + second1 = 0 + ms1 = 0 + hour2 = 0 + minute2 = 0 + second2 = 0 + ms2 = 0 + if values['hour1'] is not None: + hour1 = int(values['hour1']) + if values['minute1'] is not None: + minute1 = int(values['minute1']) + if values['second1'] is not None: + second1 = int(values['second1']) + if values['ms1'] is not None: + ms1 = int(values['ms1']) + if values['hour2'] is not None: + hour2 = int(values['hour2']) + if values['minute2'] is not None: + minute2 = int(values['minute2']) + if values['second2'] is not None: + second2 = int(values['second2']) + if values['ms2'] is not None: + ms2 = int(values['ms2']) + + if hour1 < 0 or hour1 > 23: + logger.error("hour must be in [0,24[") + return None, None + if minute1 < 0 or minute1 > 59: + logger.error("minute must be in [0,60[") + return None, None + if second1 < 0 or second1 > 59: + logger.error("second must be in [0,60[") + return None, None + if ms1 < 0 or ms1 > 1000: + logger.error("milliseconds must be in [0,1000[") + return None, None + + if hour2 < 0 or hour2 > 23: + logger.error("hour must be in [0,24[") + return None, None + if minute2 < 0 or minute2 > 59: + logger.error("minute must be in [0,60[") + return None, None + if second2 < 0 or second2 > 59: + logger.error("second must be in [0,60[") + return None, None + if ms2 < 0 or ms2 > 1000: + logger.error("milliseconds must be in [0,1000[") + return None, None + + ts1 = timedelta(hours=hour1, minutes=minute1, seconds=second1, microseconds=ms1*1000) + ts2 = timedelta(hours=hour2, minutes=minute2, seconds=second2, microseconds=ms2*1000) + + if ts2 < ts1: + logger.error("Non monotonic interval") + return None,None + + return (ts1, ts2) + +@typechecked +def compare_time_interval(interval1: tuple[timedelta, timedelta], + interval2: tuple[timedelta, timedelta]) -> int: + """ + Compare two time intervals. + + This function compares two time intervals represented by tuples of two timedelta objects. + It returns an integer indicating the relationship between the two intervals: + - -1 if interval 1 is before interval 2 + - 1 if interval 1 is after interval 2 + - 0 if the two intervals overlap or are equal + + Args: + interval1 (tuple[timedelta, timedelta]): The first time interval + interval2 (tuple[timedelta, timedelta]): The second time interval + + Returns: + int: The relationship between the two time intervals + """ + ts11,ts12 = interval1 + ts21,ts22 = interval2 + + if ts12 < ts21: + return -1 + if ts22 < ts11: + return 1 + + return 0 + + +@typechecked +def get_ts_frame(frame: dict) -> timedelta|None: + logger = logging.getLogger(__name__) + + if 'pts_time' in frame: + pts_time = float(frame['pts_time']) + elif 'pkt_pts_time' in frame: + pts_time = float(frame['pkt_pts_time']) + else: + logger.error('Impossible to find timestamp of frame %s', frame) + return None + + ts = timedelta(seconds=pts_time) + return ts + +@typechecked +def get_packet_duration(packet: dict) -> int: + logger = logging.getLogger(__name__) + + if 'duration' in packet: + duration = int(packet['duration']) + elif 'pkt_duration' in packet: + duration = int(packet['pkt_duration']) + else: + logger.error('Impossible to find duration of packet %s', packet) + return None + + return duration diff --git a/src/tscut/tscut.py b/src/tscut/tscut.py index 154642e..37b4621 100755 --- a/src/tscut/tscut.py +++ b/src/tscut/tscut.py @@ -43,7 +43,18 @@ from tscut.tools.ffprobe import ( get_video_dimensions, get_movie_duration, get_streams, - with_subtitles + with_subtitles, + find_subtitles_tracks, + get_nearest_iframe, + get_ts_frame +) +from tscut.tools.ffmpeg import ( + extract_all_streams, + ffmpeg_convert +) +from tscut.tools.timeframe import ( + compare_time_interval, + parse_time_interval ) @@ -468,425 +479,7 @@ def change_codec_private_data(mkvinfo_path:str, input_file: IO[bytes], codec_dat -@typechecked -def parse_timestamp(ts:str) -> timedelta|None: - """ - Parse a timestamp string into a timedelta object. - This function takes a string representing a timestamp in the format HH:MM:SS[.us] and returns - a timedelta object representing the corresponding time interval. - The timestamp string can have an optional microsecond component. - - Args: - ts (str): The timestamp string to parse. - - Returns: - timedelta | None: - - A timedelta object representing the parsed timestamp - - None if: - - the timestamp string is not in the correct format - - the timestamp values are out of range (e.g. hour > 23, minute > 59, etc.) - """ - logger = logging.getLogger(__name__) - - ts_reg_exp = (r'^(?P[0-9]{1,2}):(?P[0-9]{1,2})' - r':(?P[0-9]{1,2})(\.(?P[0-9]{1,6}))?$') - p = re.compile(ts_reg_exp) - m = p.match(ts) - if m is None: - logger.warning("Impossible to parse timestamp: %s", ts) - return None - - values = m.groupdict() - hour = 0 - minute = 0 - second = 0 - us = 0 - if values['hour'] is not None: - hour = int(values['hour']) - if values['minute'] is not None: - minute = int(values['minute']) - if values['second'] is not None: - second = int(values['second']) - if values['us'] is not None: - us = int(values['us']) - - if hour < 0 or hour > 23: - logger.error("hour must be in [0,24[") - return None - if minute < 0 or minute > 59: - logger.error("minute must be in [0,60[") - return None - if second < 0 or second > 59: - logger.error("second must be in [0,60[") - return None - if us < 0 or us > 1000000: - logger.error("milliseconds must be in [0,1000000[") - return None - res = timedelta(hours=hour, minutes=minute, seconds=second, microseconds=us) - - return res - -@typechecked -def parse_time_interval(interval: str) -> tuple[timedelta, timedelta] | None: - """ - Parse a time interval string into a tuple of two timedelta objects. - - This function takes a string representing a time interval in the format HH:MM:SS[.ms]-HH:MM:SS[.ms] and returns a tuple of two timedelta objects representing the start and end times of the interval. - The time interval string can have an optional millisecond component. - - Args: - interval (str): The time interval string to parse. - - Returns: - tuple[timedelta, timedelta] | None: - - A tuple of two timedelta objects representing the start and end times of the interval - - None if: - - the time interval string is not in the correct format - - the time values are out of range (e.g. hour > 23, minute > 59, etc.) - - the end time is before the start time (non-monotonic interval) - """ - logger = logging.getLogger(__name__) - - interval_reg_exp = (r'^(?P[0-9]{1,2}):(?P[0-9]{1,2}):(?P[0-9]{1,2})' - r'(\.(?P[0-9]{1,3}))?-(?P[0-9]{1,2}):(?P[0-9]{1,2})' - r':(?P[0-9]{1,2})(\.(?P[0-9]{1,3}))?$') - p = re.compile(interval_reg_exp) - m = p.match(interval) - if m is None: - logger.error("Impossible to parse time interval") - return None - - values = m.groupdict() - hour1 = 0 - minute1 = 0 - second1 = 0 - ms1 = 0 - hour2 = 0 - minute2 = 0 - second2 = 0 - ms2 = 0 - if values['hour1'] is not None: - hour1 = int(values['hour1']) - if values['minute1'] is not None: - minute1 = int(values['minute1']) - if values['second1'] is not None: - second1 = int(values['second1']) - if values['ms1'] is not None: - ms1 = int(values['ms1']) - if values['hour2'] is not None: - hour2 = int(values['hour2']) - if values['minute2'] is not None: - minute2 = int(values['minute2']) - if values['second2'] is not None: - second2 = int(values['second2']) - if values['ms2'] is not None: - ms2 = int(values['ms2']) - - if hour1 < 0 or hour1 > 23: - logger.error("hour must be in [0,24[") - return None, None - if minute1 < 0 or minute1 > 59: - logger.error("minute must be in [0,60[") - return None, None - if second1 < 0 or second1 > 59: - logger.error("second must be in [0,60[") - return None, None - if ms1 < 0 or ms1 > 1000: - logger.error("milliseconds must be in [0,1000[") - return None, None - - if hour2 < 0 or hour2 > 23: - logger.error("hour must be in [0,24[") - return None, None - if minute2 < 0 or minute2 > 59: - logger.error("minute must be in [0,60[") - return None, None - if second2 < 0 or second2 > 59: - logger.error("second must be in [0,60[") - return None, None - if ms2 < 0 or ms2 > 1000: - logger.error("milliseconds must be in [0,1000[") - return None, None - - ts1 = timedelta(hours=hour1, minutes=minute1, seconds=second1, microseconds=ms1*1000) - ts2 = timedelta(hours=hour2, minutes=minute2, seconds=second2, microseconds=ms2*1000) - - if ts2 < ts1: - logger.error("Non monotonic interval") - return None,None - - return (ts1, ts2) - -@typechecked -def compare_time_interval(interval1: tuple[timedelta, timedelta], - interval2: tuple[timedelta, timedelta]) -> int: - """ - Compare two time intervals. - - This function compares two time intervals represented by tuples of two timedelta objects. - It returns an integer indicating the relationship between the two intervals: - - -1 if interval 1 is before interval 2 - - 1 if interval 1 is after interval 2 - - 0 if the two intervals overlap or are equal - - Args: - interval1 (tuple[timedelta, timedelta]): The first time interval - interval2 (tuple[timedelta, timedelta]): The second time interval - - Returns: - int: The relationship between the two time intervals - """ - ts11,ts12 = interval1 - ts21,ts22 = interval2 - - if ts12 < ts21: - return -1 - elif ts22 < ts11: - return 1 - else: - return 0 - -@typechecked -def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], input_format:str, - output_file: IO[bytes], output_format:str, duration: timedelta): - logger = logging.getLogger(__name__) - - width, height = get_video_dimensions(ffprobe_path, input_file) - subtitles = with_subtitles(ffprobe_path, input_file) - - infd = input_file.fileno() - outfd = output_file.fileno() - set_inheritable(infd, True) - set_inheritable(outfd, True) - - if logger.getEffectiveLevel() == logging.DEBUG: - log = [] - else: - log = [ '-loglevel', 'quiet' ] - - params = [ffmpeg_path, '-y',]+log+['-progress', '/dev/stdout', '-canvas_size', - f'{width:d}x{height:d}', '-f', input_format, - '-i', f'/proc/self/fd/{infd:d}', '-map', '0:v', - '-map', '0:a'] - if subtitles: - params.extend(['-map', '0:s']) - params.extend(['-bsf:v', 'h264_mp4toannexb,dump_extra=freq=keyframe', '-vcodec', 'copy', - '-acodec', 'copy']) - if subtitles: - params.extend(['-scodec', 'dvdsub']) - params.extend(['-r:0', '25', '-f', output_format, f'/proc/self/fd/{outfd:d}']) - - logger.debug('Executing %s', params) - - with Popen(params, stdout=PIPE, close_fds=False) as ffmpeg: - pb = tqdm(TextIOWrapper(ffmpeg.stdout, encoding="utf-8"), - total=int(duration/timedelta(seconds=1)), unit='s', desc='Conversion') - for line in pb: - if line.startswith('out_time='): - ts = line.split('=')[1].strip() - ts = parse_timestamp(ts) - if ts is not None: - pb.n = int(ts/timedelta(seconds=1)) - pb.update() - status = ffmpeg.wait() - if status != 0: - logger.error('Conversion failed with status code: %d', status) - -@typechecked -def get_ts_frame(frame: dict) -> timedelta|None: - logger = logging.getLogger(__name__) - - if 'pts_time' in frame: - pts_time = float(frame['pts_time']) - elif 'pkt_pts_time' in frame: - pts_time = float(frame['pkt_pts_time']) - else: - logger.error('Impossible to find timestamp of frame %s', frame) - return None - - ts = timedelta(seconds=pts_time) - return ts - -@typechecked -def get_packet_duration(packet: dict) -> int: - logger = logging.getLogger(__name__) - - if 'duration' in packet: - duration = int(packet['duration']) - elif 'pkt_duration' in packet: - duration = int(packet['pkt_duration']) - else: - logger.error('Impossible to find duration of packet %s', packet) - return None - - return duration - -@typechecked -def get_frames_in_stream(ffprobe_path: str, input_file: IO[bytes], begin:timedelta, end:timedelta, - stream_kind:str, sub_stream_id:int=0) -> list[dict]|None: - logger = logging.getLogger(__name__) - infd = input_file.fileno() - set_inheritable(infd, True) - - command = [ffprobe_path, '-loglevel', 'quiet', '-read_intervals', f'{begin}%{end}', - '-show_entries', 'frame', '-select_streams', - f'{stream_kind}:{sub_stream_id:d}','-of', 'json', f'/proc/self/fd/{infd:d}'] - logger.debug('Executing: %s', command) - - with Popen(command, stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - frames = json.load(BytesIO(out)) - status = ffprobe.wait() - if status != 0: - logger.error('ffprobe failed with status code: %d', status) - return None - - # Sort frames by timestamp - tmp = {} - if 'frames' in frames: - frames = frames['frames'] - for frame in frames: - ts = get_ts_frame(frame) - if ts is None: - return None - if begin <= ts <= end: - tmp[ts]=frame - - res = [] - for ts in sorted(tmp): - res.append(tmp[ts]) - return res - else: - logger.error('Impossible to retrieve frames inside file around [%s,%s]', begin, end) - return None - -# TODO: Finish implementation of this function and use it. -@typechecked -def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:timedelta, - before: bool=True, delta: timedelta=timedelta(seconds=2)): - # pylint: disable=W0613 - logger = logging.getLogger(__name__) - - zero = timedelta() - tbegin = timestamp-delta - tend = timestamp+delta - tbegin = max(tbegin, zero) - - infd = input_file.fileno() - set_inheritable(infd, True) - - logger.debug('Looking for IDR frame in [%s, %s]', tbegin, tend) - - idrs = [] - - # Retains only IDR frame - with Popen([ffprobe_path, '-loglevel', 'quiet', '-read_intervals', f'{tbegin}%{tend}', - '-skip_frame', 'nokey', '-show_entries', 'frame', '-select_streams', 'v:0', - '-of', 'json', f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - frames = json.load(BytesIO(out)) - status = ffprobe.wait() - if status != 0: - logger.error('ffprobe failed with status code: %d', status) - return None - - if 'frames' in frames: - frames = frames['frames'] - for frame in frames: - ts = get_ts_frame(frame) - if ts is None: - return None - if tbegin <= ts <= tend: - idrs.append(frame) - else: - logger.error('Impossible to retrieve IDR frames inside file around [%s,%s]', - tbegin, tend) - return None - - return None - -@typechecked -def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], - timestamp:timedelta, before:bool=True, - delta_max:timedelta=timedelta(seconds=15))-> tuple[int,dict]: - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - set_inheritable(infd, True) - - delta = timedelta(seconds=1) - - iframe = None - while delta < delta_max: - zero = timedelta() - if before: - tbegin = timestamp-delta - else: - tbegin = timestamp - if not before: - tend = timestamp+delta - else: - tend = timestamp - tbegin = max(tbegin, zero) - logger.debug('Looking for an iframe in [%s, %s]', tbegin, tend) - - frames = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=tbegin, end=tend, - stream_kind='v') - if frames is None: - logger.debug('Found no frame in [%s, %s]', tbegin, tend) - delta+=timedelta(seconds=1) - continue - - iframes = [] - for frame in frames: - if frame['pict_type'] == 'I': - iframes.append(frame) - - found = False - for frame in iframes: - ts = get_ts_frame(frame) - if ts is None: - logger.warning('I-frame with no timestamp: %s', frame) - continue - - if before and ts <= timestamp: - found = True - iframe = frame - if not before and ts >= timestamp: - found = True - iframe = frame - break - - if found: - logger.info("Found i-frame at: %s", iframe) - break - else: - delta+=timedelta(seconds=1) - continue - - if iframe is not None: - its = get_ts_frame(iframe) - nb_frames = 0 - for frame in frames: - ts = get_ts_frame(frame) - if ts is None: - logger.warning('Frame without timestamp: %s', frame) - continue - - if before: - if its <= ts <= timestamp: - logger.info("Retrieve a frame between %s and %s at %s", its, timestamp, ts) - nb_frames = nb_frames+1 - else: - if timestamp <= ts <= its: - logger.info("Retrieve a frame between %s and %s at %s", timestamp, ts, its) - nb_frames = nb_frames+1 - else: - logger.error("Impossible to find I-frame between: %s and %s", tbegin, tend) - return 0, None - - return(nb_frames, iframe) @typechecked def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[bytes], @@ -931,412 +524,7 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt elif status == 2: logger.error('Extraction returns errors') -@typechecked -def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_frames:int, - width:int=640, height:int=480) -> tuple[bytes,int]|tuple[None,None]: - """ - Extract pictures from a video file using FFmpeg. - This function runs the FFmpeg binary to extract a specified number of frames from a video file, - starting at a given time. - The extracted frames are stored in memory as PPM images and returned as a tuple containing - the image data and a file descriptor to the memory created by memfd_create. - - Args: - ffmpeg_path (str): The path to the FFmpeg binary. - input_file (IO[bytes]): The input video file. - begin (timedelta): The start time of the extraction. - nb_frames (int): The number of frames to extract. - width (int, optional): The width of the extracted images. Defaults to 640. - height (int, optional): The height of the extracted images. Defaults to 480. - - Returns: - tuple[bytes, int] | tuple[None, None]: - - A tuple containing the extracted image data as bytes and a file descriptor - - A tuple containing None, None if the extraction fails - """ - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - outfd = memfd_create('pictures', flags=0) - set_inheritable(outfd, True) - # PPM header - # "P6\nWIDTH HEIGHT\n255\n" - header_len=2+1+ceil(log(width, 10))+1+ceil(log(height, 10))+1+3+1 - logger.debug('Header length: %d', header_len) - image_length = width*height*3+header_len - length = image_length*nb_frames - logger.debug("Estimated length: %d", length) - - command = [ffmpeg_path, '-loglevel', 'quiet' ,'-y', '-ss', f'{begin}', '-i', - f'/proc/self/fd/{infd}', '-s', f'{width:d}x{height:d}', '-vframes', f'{nb_frames:d}', - '-c:v', 'ppm','-f', 'image2pipe', f'/proc/self/fd/{outfd:d}'] - logger.debug('Executing: %s', command) - - images = b'' - with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg: - status = ffmpeg.wait() - if status != 0: - logger.error('Conversion failed with status code: %d', status) - return None, None - - lseek(outfd, 0, SEEK_SET) - images = read(outfd,length) - if len(images) != length: - logger.error("Received %d bytes but %d were expected.", len(images), length) - return None, None - - lseek(outfd, 0, SEEK_SET) - return images, outfd - -@typechecked -def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, output_filename:str, - packet_duration:int, sub_channel:int=0, - nb_packets:int=0, sample_rate:int=48000, - nb_channels:int=2) -> tuple[bytes,int]|tuple[None,None]: - logger = logging.getLogger(__name__) - - outfd = memfd_create(output_filename, flags=0) - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - set_inheritable(outfd, True) - sound = b'' - length = int(nb_channels*sample_rate*4*nb_packets*packet_duration/1000) - - command = [ffmpeg_path, '-y', '-loglevel', 'quiet', '-ss', f'{begin}', - '-i', f'/proc/self/fd/{infd}', f'-frames:a:{sub_channel:d}', f'{nb_packets+1:d}', - '-c:a', 'pcm_s32le', '-sample_rate', f'{sample_rate:d}', - '-channels', f'{nb_channels:d}', '-f', 's32le', f'/proc/self/fd/{outfd:d}'] - logger.debug('Executing: %s', command) - - with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg: - status = ffmpeg.wait() - if status != 0: - logger.error('Sound extraction returns error code: %d', status) - return None, None - - lseek(outfd, 0, SEEK_SET) - sound = read(outfd, length) - - if len(sound) != length: - logger.info("Received %d bytes but %d were expected (channels=%d, freq=%d, packets=%d,\ - duration=%d ms).", len(sound), length, nb_channels, sample_rate, nb_packets, - packet_duration) - return None, None - - return sound, outfd - -@typechecked -def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None: - """ - Dump PPM pictures from a bytes buffer to files. - - This function takes a bytes buffer containing PPM pictures, a prefix for the output file names, and a list of temporary files. - It extracts each PPM picture from the buffer, checks its validity, and writes it to a file. - The output files are named according to the prefix and a zero-padded three-digit number. - - Args: - pictures (bytes): The bytes buffer containing the PPM pictures. - prefix (str): The prefix for the output file names. - temporaries (list[IO[bytes]]): A list of temporary files that will be used to store the output files. - - Returns: - None - - Raises: - None, but logs errors if: - - the PPM picture is not valid (e.g. wrong magic number, dimensions, or color encoding) - - an I/O error occurs while creating or writing to an output file - """ - logger = logging.getLogger(__name__) - - # "P6\nWIDTH HEIGHT\n255\n" - pos = 0 - picture = 0 - - logger.debug('Dumping %d pictures: %s', len(pictures),prefix) - - while pos[0-9]+) (?P[0-9]+)\n$') - m = pattern.match(dimensions) - if m is not None: - width = int(m['width']) - height = int(m['height']) - else: - logger.error('Impossible to parse dimensions of picture') - return - else: - logger.error('Not a PPM picture') - return - - if max_value != 255: - logger.error('Not a valid PPM picture. Color are not encoded on byte. Max value: %d', - max_value) - - header_len=2+1+ceil(log(width, 10))+1+ceil(log(height, 10))+1+3+1 - try: - with open(filename, 'wb') as out: - temporaries.append(out) - outfd = out.fileno() - length=header_len+3*width*height - nb_bytes = 0 - while nb_bytes < length: - nb_bytes+=write(outfd, pictures[pos+nb_bytes:pos+length]) - pos+=length - picture+=1 - except OSError: - logger.error('Impossible to create file: %s', filename) - -@typechecked -def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], begin:timedelta, - end:timedelta, streams, files_prefix, nb_frames:int, framerate:float, - width:int, height:int, temporaries, dump_mem_fd:bool=False): - logger = logging.getLogger(__name__) - - # The command line for encoding only video track - video_encoder_params = [ ffmpeg_path, '-y', '-loglevel', 'quiet'] - video_input_params = [] - video_codec_params = [] - - # The command line to create a MKV file with the rest of tracks - generic_encoder_params = [ ffmpeg_path, '-y', '-loglevel', 'quiet' ] - generic_input_params = [] - generic_codec_params = [] - - if begin < end: - video_id=0 - audio_id=0 - subtitle_id=0 - memfds = [] - for stream in streams: - if stream['codec_type'] == 'video': - logger.info("Extracting %d frames of video stream v:%d", nb_frames, video_id) - sar = stream['sample_aspect_ratio'] - dar = stream['display_aspect_ratio'] - pixel_format = stream['pix_fmt'] - color_range = stream['color_range'] - color_space =stream['color_space'] - color_transfer = stream['color_transfer'] - color_primaries = stream['color_primaries'] - level = int(stream['level']) - level = f'{floor(level/10):d}.{level%10:d}' - chroma_location = stream['chroma_location'] - field_order = stream - match field_order: - case 'progressive': - interlaced_options = ['-field_order', '0'] - case 'tt': - interlaced_options = ['-top', '1', f'-flags:v:{video_id:d}', '+ilme+ildct', - '-field_order', '1'] - case 'bb': - interlaced_options = ['-top', '0', f'-flags:v:{video_id:d}', '+ilme+ildct', - '-field_order','2'] - case 'tb': - interlaced_options = ['-top', '1', f'-flags:v:{video_id:d}', '+ilme+ildct', - '-field_order', '3'] - case 'bt': - interlaced_options = ['-top', '0', f'-flags:v:{video_id:d}', '+ilme+ildct', - '-field_order', '4'] - case _: - interlaced_options = [] - - # ======================================= # - # TODO: adjust SAR and DAR - # https://superuser.com/questions/907933/correct-aspect-ratio-without-re-encoding-video-file - # SAR: -aspect width:height - # DAR: -bsf:v sample_aspect_ratio=1:video_format - logger.warning('Missing SAR adjustment for: %s', sar) - logger.warning('Missing DAR adjustment for: %s', dar) - logger.warning('Missing treatment for chroma location: %s', chroma_location) - codec = stream['codec_name'] - images_bytes, memfd = extract_pictures(ffmpeg_path, input_file=input_file, - begin=begin, nb_frames=nb_frames, - width=width, height=height) - if images_bytes is None: - logger.error('Impossible to extract picture from video stream.') - exit(-1) - - memfds.append(memfd) - if dump_mem_fd: - dump_ppm(images_bytes, f'{files_prefix}-{video_id:d}', temporaries) - - # We rewind to zero the memory file descriptor - lseek(memfd, 0, SEEK_SET) - set_inheritable(memfd, True) - - video_input_params.extend(['-framerate', f'{framerate:f}', '-f', 'image2pipe', '-i', - f'/proc/self/fd/{memfd:d}']) - video_codec_params.extend([f'-c:v:{video_id:d}', codec, f'-level:v:{video_id:d}', - level, '-pix_fmt', pixel_format]) - video_codec_params.extend(interlaced_options) - video_codec_params.extend([f'-colorspace:v:{video_id}', color_space, - f'-color_primaries:v:{video_id:d}', color_primaries, - f'-color_trc:v:{video_id:d}', color_transfer, - f'-color_range:v:{video_id:d}', color_range]) - video_id=video_id+1 - elif stream['codec_type'] == 'audio': - logger.debug('Audio stream: %s', stream) - sample_rate = int(stream['sample_rate']) - nb_channels = int(stream['channels']) - if 'bit_rate' in stream: - bit_rate = int(stream['bit_rate']) - else: - bit_rate = 128000 - codec = stream['codec_name'] - if 'tags' in stream: - if 'language' in stream['tags']: - generic_codec_params.extend([f'-metadata:s:a:{audio_id:d}', - f"language={stream['tags']['language']}"]) - packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin, - end=end, stream_kind='a', sub_stream_id=audio_id) - nb_packets = len(packets) - logger.debug("Found %d packets to be extracted from audio track.", nb_packets) - if nb_packets > 0: - packet_duration = get_packet_duration(packets[0]) - if packet_duration is None: - return None - else: - packet_duration = 0 - - logger.info("Extracting %d packets of audio stream: a:%d" , nb_packets, audio_id) - tmpname = f'{files_prefix}-{audio_id:d}.pcm' - - sound_bytes, memfd = extract_sound(ffmpeg_path=ffmpeg_path, input_file=input_file, - begin=begin, nb_packets=nb_packets, - packet_duration=packet_duration, - output_filename=tmpname, - sample_rate=sample_rate, nb_channels=nb_channels) - - if sound_bytes is None: - logger.error('Impossible to extract sound track') - exit(-1) - - memfds.append(memfd) - - if dump_mem_fd: - try: - with open(tmpname,'wb') as output: - temporaries.append(output) - outfd = output.fileno() - pos = 0 - while pos < len(sound_bytes): - pos+=write(outfd, sound_bytes[pos:]) - except OSError: - logger.error('Impossible to create file: %s', tmpname) - return None - - # We rewind to zero the memory file descriptor - lseek(memfd, 0, SEEK_SET) - set_inheritable(memfd, True) - - generic_input_params.extend(['-f', 's32le', '-ar', f'{sample_rate:d}', '-ac', - f'{nb_channels:d}', '-i', f'/proc/self/fd/{memfd:d}']) - generic_codec_params.extend([f'-c:a:{audio_id:d}', codec, f'-b:a:{audio_id:d}', - f'{bit_rate:d}']) - audio_id=audio_id+1 - elif stream['codec_type'] == 'subtitle': - logger.info("Extracting a subtitle stream: s:%d", subtitle_id) - codec = stream['codec_name'] - generic_input_params.extend(['-i', './empty.idx']) - if 'tags' in stream: - if 'language' in stream['tags']: - generic_codec_params.extend([f'-metadata:s:s:{subtitle_id:d}', - f"language={stream['tags']['language']}"]) - generic_codec_params.extend([f'-c:s:{subtitle_id:d}', 'copy']) - subtitle_id=subtitle_id+1 - else: - logger.error("Unknown stream type: %s", stream['codec_type']) - - # Create a new MKV movie with all streams (except videos) that have been extracted. - generic_encoder_params.extend(generic_input_params) - - for index in range(audio_id+subtitle_id): - generic_encoder_params.extend(['-map', f'{index:d}']) - generic_encoder_params.extend(generic_codec_params) - - mkv_filename = f'{files_prefix}.mkv' - try: - mkv_output = open(mkv_filename,'wb+') - except OSError: - logger.error('Impossible to create file: %s', mkv_filename) - return None - - mkvoutfd = mkv_output.fileno() - set_inheritable(mkvoutfd, True) - generic_encoder_params.extend(['-f', 'matroska', f'/proc/self/fd/{mkvoutfd:d}']) - - logger.info('Encoding all streams (except video) into a MKV file: %s', mkv_filename) - logger.debug('Executing: %s', generic_encoder_params) - with Popen(generic_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg: - status = ffmpeg.wait() - if status != 0: - logger.error('Encoding failed with status code: %d', status) - return None - - temporaries.append(mkv_output) - - h264_filename = f'{files_prefix}.h264' - try: - h264_output = open(h264_filename,'wb+') - except OSError: - logger.error('Impossible to create file: %s', h264_filename) - return None - - h264outfd = h264_output.fileno() - set_inheritable(h264outfd, True) - - video_encoder_params.extend(video_input_params) - video_encoder_params.extend(video_codec_params) - - video_encoder_params.extend([ '-x264opts', f'keyint=1:sps-id={1:d}','-bsf:v', - 'h264_mp4toannexb,dump_extra=freq=keyframe,h264_metadata=\ - overscan_appropriate_flag=1:sample_aspect_ratio=1:video_format=\ - 0:chroma_sample_loc_type=0','-f', 'h264', - f'/proc/self/fd/{h264outfd:d}']) - - logger.info('Encoding video into a H264 file: %s', h264_filename) - logger.debug('Executing: %s', video_encoder_params) - with Popen(video_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg: - status = ffmpeg.wait() - if status != 0: - logger.error('Encoding failed with status code: %d', status) - return None - - temporaries.append(h264_output) - - h264_ts_filename = f'{files_prefix}-ts.txt' - try: - h264_ts_output = open(h264_ts_filename,'w+', encoding='utf8') - except OSError: - logger.error('Impossible to create file: %s', h264_ts_filename) - return None - - h264_ts_output.write('# timestamp format v2\n') - ts = 0 - for _ in range(nb_frames): - ts = ts+ceil(1000/framerate) - h264_ts_output.write(f'{ts:d}\n') - h264_ts_output.flush() - h264_ts_output.seek(0) - - temporaries.append(h264_ts_output) - - for memfd in memfds: - close(memfd) - - return h264_output, h264_ts_output, mkv_output - - else: - # Nothing to be done. We are already at a i-frame boundary. - return None, None # Merge a list of mkv files passed as input, and produce a new MKV as output @typechecked @@ -1421,28 +609,7 @@ def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str, return out -def find_subtitles_tracks(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: - logger = logging.getLogger(__name__) - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - - command = [ffprobe_path, '-loglevel','quiet', '-i', f'/proc/self/fd/{infd:d}', - '-select_streams', 's', '-show_entries', 'stream=index:stream_tags=language', - '-of', 'json'] - logger.debug('Executing: %s', command) - - with Popen(command, stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'streams' in out: - return out['streams'] - else: - logger.error('Impossible to retrieve format of file') - - ffprobe.wait() - return None @typechecked def extract_track_from_mkv(mkvextract_path: str, input_file: IO[bytes], index,