Fix typing errors in ffprobe and ffmpeg related functions.
This commit is contained in:
@@ -24,7 +24,7 @@ from tqdm import tqdm
|
|||||||
from tscut.tools.ffprobe import (
|
from tscut.tools.ffprobe import (
|
||||||
get_video_dimensions,
|
get_video_dimensions,
|
||||||
with_subtitles,
|
with_subtitles,
|
||||||
get_frames_in_stream
|
get_frames_in_stream,
|
||||||
)
|
)
|
||||||
from tscut.tools.ppm import dump_ppm
|
from tscut.tools.ppm import dump_ppm
|
||||||
from tscut.tools.timeframe import parse_timestamp, get_packet_duration
|
from tscut.tools.timeframe import parse_timestamp, get_packet_duration
|
||||||
@@ -36,6 +36,9 @@ logger = logging.getLogger(__name__)
|
|||||||
def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], input_format:str,
|
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):
|
output_file: IO[bytes], output_format:str, duration: timedelta):
|
||||||
width, height = get_video_dimensions(ffprobe_path, input_file)
|
width, height = get_video_dimensions(ffprobe_path, input_file)
|
||||||
|
if width is None or height is None:
|
||||||
|
return
|
||||||
|
|
||||||
subtitles = with_subtitles(ffprobe_path, input_file)
|
subtitles = with_subtitles(ffprobe_path, input_file)
|
||||||
|
|
||||||
infd = input_file.fileno()
|
infd = input_file.fileno()
|
||||||
@@ -67,8 +70,8 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp
|
|||||||
total=int(duration/timedelta(seconds=1)), unit='s', desc='Conversion')
|
total=int(duration/timedelta(seconds=1)), unit='s', desc='Conversion')
|
||||||
for line in pb:
|
for line in pb:
|
||||||
if line.startswith('out_time='):
|
if line.startswith('out_time='):
|
||||||
ts = line.split('=')[1].strip()
|
ts_str = line.split('=')[1].strip()
|
||||||
ts = parse_timestamp(ts)
|
ts = parse_timestamp(ts_str)
|
||||||
if ts is not None:
|
if ts is not None:
|
||||||
pb.n = int(ts/timedelta(seconds=1))
|
pb.n = int(ts/timedelta(seconds=1))
|
||||||
pb.update()
|
pb.update()
|
||||||
@@ -199,8 +202,8 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
|
|||||||
color_space =stream['color_space']
|
color_space =stream['color_space']
|
||||||
color_transfer = stream['color_transfer']
|
color_transfer = stream['color_transfer']
|
||||||
color_primaries = stream['color_primaries']
|
color_primaries = stream['color_primaries']
|
||||||
level = int(stream['level'])
|
level_int = int(stream['level'])
|
||||||
level = f'{floor(level/10):d}.{level%10:d}'
|
level = f'{floor(level_int/10):d}.{level_int%10:d}'
|
||||||
chroma_location = stream['chroma_location']
|
chroma_location = stream['chroma_location']
|
||||||
field_order = stream
|
field_order = stream
|
||||||
match field_order:
|
match field_order:
|
||||||
@@ -233,7 +236,7 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
|
|||||||
images_bytes, memfd = extract_pictures(ffmpeg_path, input_file=input_file,
|
images_bytes, memfd = extract_pictures(ffmpeg_path, input_file=input_file,
|
||||||
begin=begin, nb_frames=nb_frames,
|
begin=begin, nb_frames=nb_frames,
|
||||||
width=width, height=height)
|
width=width, height=height)
|
||||||
if images_bytes is None:
|
if images_bytes is None or memfd is None:
|
||||||
logger.error('Impossible to extract picture from video stream.')
|
logger.error('Impossible to extract picture from video stream.')
|
||||||
exit(-1)
|
exit(-1)
|
||||||
|
|
||||||
@@ -270,6 +273,9 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
|
|||||||
f"language={stream['tags']['language']}"])
|
f"language={stream['tags']['language']}"])
|
||||||
packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin,
|
packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin,
|
||||||
end=end, stream_kind='a', sub_stream_id=audio_id)
|
end=end, stream_kind='a', sub_stream_id=audio_id)
|
||||||
|
if packets is None:
|
||||||
|
logger.error("Impossible to retrieve audio packets")
|
||||||
|
return None
|
||||||
nb_packets = len(packets)
|
nb_packets = len(packets)
|
||||||
logger.debug("Found %d packets to be extracted from audio track.", nb_packets)
|
logger.debug("Found %d packets to be extracted from audio track.", nb_packets)
|
||||||
if nb_packets > 0:
|
if nb_packets > 0:
|
||||||
@@ -288,7 +294,7 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
|
|||||||
output_filename=tmpname,
|
output_filename=tmpname,
|
||||||
sample_rate=sample_rate, nb_channels=nb_channels)
|
sample_rate=sample_rate, nb_channels=nb_channels)
|
||||||
|
|
||||||
if sound_bytes is None:
|
if sound_bytes is None or memfd is None:
|
||||||
logger.error('Impossible to extract sound track')
|
logger.error('Impossible to extract sound track')
|
||||||
exit(-1)
|
exit(-1)
|
||||||
|
|
||||||
|
|||||||
@@ -173,7 +173,8 @@ def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|Non
|
|||||||
|
|
||||||
# ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts
|
# ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts
|
||||||
@typechecked
|
@typechecked
|
||||||
def get_video_dimensions(ffprobe_path:str, input_file: IO[bytes]) -> tuple[int,int]|None:
|
def get_video_dimensions(ffprobe_path:str,
|
||||||
|
input_file: IO[bytes]) -> tuple[int,int]| tuple[None,None]:
|
||||||
infd = input_file.fileno()
|
infd = input_file.fileno()
|
||||||
lseek(infd, 0, SEEK_SET)
|
lseek(infd, 0, SEEK_SET)
|
||||||
set_inheritable(infd, True)
|
set_inheritable(infd, True)
|
||||||
@@ -188,7 +189,7 @@ def get_video_dimensions(ffprobe_path:str, input_file: IO[bytes]) -> tuple[int,i
|
|||||||
return int(video['width']), int(video['height'])
|
return int(video['width']), int(video['height'])
|
||||||
|
|
||||||
logger.error('Impossible to retrieve dimensions of video')
|
logger.error('Impossible to retrieve dimensions of video')
|
||||||
return None
|
return None, None
|
||||||
|
|
||||||
@typechecked
|
@typechecked
|
||||||
def get_streams(ffprobe_path:str, input_file: IO[bytes]) -> list|None:
|
def get_streams(ffprobe_path:str, input_file: IO[bytes]) -> list|None:
|
||||||
|
|||||||
Reference in New Issue
Block a user