diff --git a/src/tscut/tools/__init__.py b/src/tscut/tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/tscut/tools/ffprobe.py b/src/tscut/tools/ffprobe.py new file mode 100644 index 0000000..110f7c9 --- /dev/null +++ b/src/tscut/tools/ffprobe.py @@ -0,0 +1,238 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright (C) 2026 Frédéric Tronel + +import json +import logging +from math import floor +from datetime import timedelta +import os +from os import ( + SEEK_SET, + lseek, + set_inheritable, +) +from typing import IO +from subprocess import PIPE, Popen +from io import BytesIO + +from typeguard import typechecked + +logger = logging.getLogger(__name__) + + +@typechecked +def get_frame_rate(ffprobe_path:str, input_file: IO[bytes]) -> float|None: + """ + Retrieves the frame rate of a video file using the ffprobe tool. + + This function runs the ffprobe binary with the specified input file and parses the output + to extract the frame rate. + It uses two methods to calculate the frame rate: one based on the timestamp of the frames + and another based on the duration of the frames. + If the two calculated frame rates are significantly different, the function returns an error + + Args: + ffprobe_path (str): The path to the ffprobe binary. + input_file (IO[bytes]): The input video file. + + Returns: + float | None: + - The frame rate of the video file as a floating-point number + - None if an error occurs while running the ffprobe binary or if the calculated + frame rates are inconsistent + """ + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + + mean_duration = 0. + nb_frames1 = 0 + nb_frames2 = 0 + min_ts = None + max_ts = None + interlaced = False + + params = [ffprobe_path, '-loglevel', 'quiet', '-select_streams', 'v', '-show_frames', + '-read_intervals', '00%+30', '-of', 'json', f'/proc/self/fd/{infd:d}'] + env = {**os.environ, 'LANG': 'C'} + with Popen(params, stdout=PIPE, close_fds=False, env=env) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'frames' in out: + for frame in out['frames']: + if 'interlaced_frame' in frame: + if frame['interlaced_frame'] == 1: + interlaced = True + if 'pts_time' in frame: + ts = float(frame['pts_time']) + if min_ts is None: + min_ts = ts + if max_ts is None: + max_ts = ts + min_ts = min(min_ts, ts) + max_ts = max(max_ts, ts) + nb_frames1+=1 + if 'duration_time' in frame: + mean_duration+=float(frame['duration_time']) + nb_frames2+=1 + else: + return None + + ffprobe.wait() + + if ffprobe.returncode != 0: + logger.error("ffprobe returns an error code: %d", ffprobe.returncode) + return None + + frame_rate1 = nb_frames1/(max_ts-min_ts) + frame_rate2 = nb_frames2 / mean_duration + + if abs(frame_rate1 - frame_rate2) > 0.2: + if not interlaced: + logger.error('Video is not interlaced and the disperancy between frame rates is too \ + big: %f / %f', frame_rate1, frame_rate2) + return None + if abs(frame_rate1*2 - frame_rate2) < 0.2: + return frame_rate2/2 + logger.error('Video is interlaced and the disperancy between frame rates is too big:\ + %f / %f', frame_rate1, frame_rate2) + return None + + return frame_rate2 + +@typechecked +def get_subtitles_tracks(ffprobe_path:str, mkv_path: str) -> dict[str,str]|None: + tracks={} + + with Popen([ffprobe_path, '-loglevel', 'quiet', '-select_streams', 's', '-show_entries', + 'stream=index,codec_name:stream_tags=language', '-of', 'json', mkv_path], + stdout=PIPE) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'streams' in out: + for stream in out['streams']: + index = stream['index'] + codec = stream['codec'] + lang = stream['tags']['language'] + if codec == 'dvd_subtitle': + if lang not in tracks: + tracks[lang] = [index] + else: + current_langs = tracks[lang] + current_langs.append(index) + tracks[lang] = current_langs + else: + return None + + ffprobe.wait() + + if ffprobe.returncode != 0: + logger.error("ffprobe returns an error code: %d", ffprobe.returncode) + return None + + return tracks + +@typechecked +def get_format(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_format', '-of', 'json', '-i', + f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'format' in out: + return out['format'] + logger.error('Impossible to retrieve format of file') + + return None + +@typechecked +def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|None: + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_format', '-of', 'json', '-i', + f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'format' in out and 'duration' in out['format']: + duration = floor(float(out['format']['duration'])) + ts = timedelta(seconds=duration) + return ts + logger.error('Impossible to retrieve duration of movie') + + return None + +# ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts +@typechecked +def get_video_dimensions(ffprobe_path:str, input_file: IO[bytes]) -> tuple[int,int]|None: + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + with Popen([ffprobe_path, '-loglevel', 'quiet', '-select_streams', 'v:0', '-show_entries',\ + 'stream=width,height', '-of', 'json', '-i', f'/proc/self/fd/{infd:d}'],\ + stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'streams' in out: + video = out['streams'][0] + if ('width' in video) and ('height' in video): + return int(video['width']), int(video['height']) + + logger.error('Impossible to retrieve dimensions of video') + return None + +@typechecked +def get_streams(ffprobe_path:str, input_file: IO[bytes]) -> list|None: + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_streams', '-of', 'json', '-i', + f'/proc/self/fd/{infd:d}'], 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 streams inside file') + + return None + +@typechecked +def with_subtitles(ffprobe_path:str, input_file: IO[bytes]) -> bool: + """ + Checks if a media file contains subtitles using the ffprobe tool. + + This function runs the ffprobe binary with the specified input file and parses the output + to determine if the file contains subtitles. + It returns True if at least one subtitle stream is found, False otherwise. + + Args: + ffprobe_path (str): The path to the ffprobe binary. + input_file (IO[bytes]): The input media file. + + Returns: + bool: + - True if the media file contains at least one subtitle stream + - False if: + - the media file does not contain any subtitle streams + - an error occurs while running the ffprobe binary + - the streams information cannot be retrieved from the media file + """ + infd = input_file.fileno() + lseek(infd, 0, SEEK_SET) + set_inheritable(infd, True) + with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_streams', '-of', 'json', '-i', + f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: + out, _ = ffprobe.communicate() + out = json.load(BytesIO(out)) + if 'streams' in out: + streams = out['streams'] + for stream in streams: + if 'codec_type' in stream and stream['codec_type'] == 'subtitle': + return True + else: + logger.error('Impossible to retrieve streams inside file') + + return False diff --git a/src/tscut/tscut.py b/src/tscut/tscut.py index f0e57e5..154642e 100755 --- a/src/tscut/tscut.py +++ b/src/tscut/tscut.py @@ -36,6 +36,16 @@ from typeguard import typechecked from tscut.h264.avc import (dump_codec_private_data, get_avc_config_from_h264, parse_codec_private) +from tscut.matroska.ebml import change_ebml_element_size +from tscut.tools.ffprobe import ( + get_format, + get_frame_rate, + get_video_dimensions, + get_movie_duration, + get_streams, + with_subtitles +) + # Useful SPS/PPS discussion. # https://copyprogramming.com/howto/including-sps-and-pps-in-a-raw-h264-track @@ -142,120 +152,7 @@ def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]|None: return res -@typechecked -def get_frame_rate(ffprobe_path:str, input_file: IO[bytes]) -> float|None: - """ - Retrieves the frame rate of a video file using the ffprobe tool. - This function runs the ffprobe binary with the specified input file and parses the output - to extract the frame rate. - It uses two methods to calculate the frame rate: one based on the timestamp of the frames - and another based on the duration of the frames. - If the two calculated frame rates are significantly different, the function returns an error - - Args: - ffprobe_path (str): The path to the ffprobe binary. - input_file (IO[bytes]): The input video file. - - Returns: - float | None: - - The frame rate of the video file as a floating-point number - - None if an error occurs while running the ffprobe binary or if the calculated - frame rates are inconsistent - """ - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - - mean_duration = 0. - nb_frames1 = 0 - nb_frames2 = 0 - min_ts = None - max_ts = None - interlaced = False - - params = [ffprobe_path, '-loglevel', 'quiet', '-select_streams', 'v', '-show_frames', - '-read_intervals', '00%+30', '-of', 'json', f'/proc/self/fd/{infd:d}'] - env = {**os.environ, 'LANG': 'C'} - with Popen(params, stdout=PIPE, close_fds=False, env=env) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'frames' in out: - for frame in out['frames']: - if 'interlaced_frame' in frame: - if frame['interlaced_frame'] == 1: - interlaced = True - if 'pts_time' in frame: - ts = float(frame['pts_time']) - if min_ts is None: - min_ts = ts - if max_ts is None: - max_ts = ts - min_ts = min(min_ts, ts) - max_ts = max(max_ts, ts) - nb_frames1+=1 - if 'duration_time' in frame: - mean_duration+=float(frame['duration_time']) - nb_frames2+=1 - else: - return None - - ffprobe.wait() - - if ffprobe.returncode != 0: - logger.error("ffprobe returns an error code: %d", ffprobe.returncode) - return None - - frame_rate1 = nb_frames1/(max_ts-min_ts) - frame_rate2 = nb_frames2 / mean_duration - - if abs(frame_rate1 - frame_rate2) > 0.2: - if not interlaced: - logger.error('Video is not interlaced and the disperancy between frame rates is too \ - big: %f / %f', frame_rate1, frame_rate2) - return None - if abs(frame_rate1*2 - frame_rate2) < 0.2: - return frame_rate2/2 - logger.error('Video is interlaced and the disperancy between frame rates is too big:\ - %f / %f', frame_rate1, frame_rate2) - return None - - return frame_rate2 - -@typechecked -def get_subtitles_tracks(ffprobe_path:str, mkv_path: str) -> dict[str,str]|None: - logger = logging.getLogger(__name__) - tracks={} - - with Popen([ffprobe_path, '-loglevel', 'quiet', '-select_streams', 's', '-show_entries', - 'stream=index,codec_name:stream_tags=language', '-of', 'json', mkv_path], - stdout=PIPE) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'streams' in out: - for stream in out['streams']: - index = stream['index'] - codec = stream['codec'] - lang = stream['tags']['language'] - if codec == 'dvd_subtitle': - if lang not in tracks: - tracks[lang] = [index] - else: - current_langs = tracks[lang] - current_langs.append(index) - tracks[lang] = current_langs - else: - return None - - ffprobe.wait() - - if ffprobe.returncode != 0: - logger.error("ffprobe returns an error code: %d", ffprobe.returncode) - return None - - return tracks @typechecked def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], @@ -569,122 +466,7 @@ def change_codec_private_data(mkvinfo_path:str, input_file: IO[bytes], codec_dat # been resized). delta+=change_ebml_element_size(input_file, pos, delta) -@typechecked -def get_format(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) - with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_format', '-of', 'json', '-i', - f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'format' in out: - return out['format'] - else: - logger.error('Impossible to retrieve format of file') - - return None - -@typechecked -def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|None: - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_format', '-of', 'json', '-i', - f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'format' in out and 'duration' in out['format']: - duration = floor(float(out['format']['duration'])) - ts = timedelta(seconds=duration) - return ts - else: - logger.error('Impossible to retrieve duration of movie') - - return None - -# ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts -@typechecked -def get_video_dimensions(ffprobe_path:str, input_file: IO[bytes]) -> tuple[int,int]: - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - with Popen([ffprobe_path, '-loglevel', 'quiet', '-select_streams', 'v:0', '-show_entries',\ - 'stream=width,height', '-of', 'json', '-i', f'/proc/self/fd/{infd:d}'],\ - stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'streams' in out: - video = out['streams'][0] - if ('width' in video) and ('height' in video): - return int(video['width']), int(video['height']) - - logger.error('Impossible to retrieve dimensions of video') - exit(-1) - -@typechecked -def get_streams(ffprobe_path:str, input_file: IO[bytes]) -> list|None: - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_streams', '-of', 'json', '-i', - f'/proc/self/fd/{infd:d}'], 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 streams inside file') - - return None - -@typechecked -def with_subtitles(ffprobe_path:str, input_file: IO[bytes]) -> bool: - """ - Checks if a media file contains subtitles using the ffprobe tool. - - This function runs the ffprobe binary with the specified input file and parses the output - to determine if the file contains subtitles. - It returns True if at least one subtitle stream is found, False otherwise. - - Args: - ffprobe_path (str): The path to the ffprobe binary. - input_file (IO[bytes]): The input media file. - - Returns: - bool: - - True if the media file contains at least one subtitle stream - - False if: - - the media file does not contain any subtitle streams - - an error occurs while running the ffprobe binary - - the streams information cannot be retrieved from the media file - """ - logger = logging.getLogger(__name__) - - infd = input_file.fileno() - lseek(infd, 0, SEEK_SET) - set_inheritable(infd, True) - with Popen([ffprobe_path, '-loglevel', 'quiet', '-show_streams', '-of', 'json', '-i', - f'/proc/self/fd/{infd:d}'], stdout=PIPE, close_fds=False) as ffprobe: - out, _ = ffprobe.communicate() - out = json.load(BytesIO(out)) - if 'streams' in out: - streams = out['streams'] - for stream in streams: - if 'codec_type' in stream and stream['codec_type'] == 'subtitle': - return True - else: - logger.error('Impossible to retrieve streams inside file') - - return False @typechecked def parse_timestamp(ts:str) -> timedelta|None: