Public/removeads
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Transfer functions that use ffprobe into a dedicated module.

Browse Source
This commit is contained in:
Frédéric Tronel
2026-08-29 10:39:51 +02:00
parent 700d1fedc5
commit 795d314f13
3 changed files with 248 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/tools/__init__.py
View File
src/tscut/tools/ffprobe.py
+238
View File
@@ -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
src/tscut/tscut.py
+10 -228
View File
@@ -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: