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

Linting and typing fixes.

Browse Source
This commit is contained in:
Frédéric Tronel
2026-08-29 16:07:29 +02:00
parent 294d857206
commit 0ff2e3a6dd
7 changed files with 43 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/exceptions.py
+3 -4
View File
@@ -3,10 +3,6 @@
# Copyright (C) 2026 Frédéric Tronel
from typing import IO
from typeguard import typechecked
class TSCutError(Exception):
"""Base exception for tscut."""
@@ -21,3 +17,6 @@ class ExternalToolError(TSCutError):
class InvalidMediaError(TSCutError):
"""Raised when input media cannot be processed."""
class UnimplementedFeature(TSCutError):
"""Raised when a rare feature is encountered and not yet implemented"""
src/tscut/h264/bitstream.py
+1 -1
View File
@@ -284,7 +284,7 @@ def write_scaling_list(buf:bytearray, bit_position: int, size, matrix:list[int],
bit_position = write_signed_exp_golomb(buf, bit_position, delta)
else:
logger.error('Not yet implemented')
exit(-1)
raise UnimplementedFeature("Optimized scaling list of H264 header is not implemented")
# reverse = deltas.reverse()
# compressed = False
# while len(reverse)>0:
src/tscut/tools/ffmpeg.py
+20 -42
View File
@@ -3,31 +3,24 @@
# 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 math import ceil, floor, log
from os import SEEK_SET, close, lseek, memfd_create, read, set_inheritable, write
from subprocess import PIPE, Popen
from typing import IO
from typeguard import typechecked
from tqdm import tqdm
from typeguard import typechecked
from tscut.exceptions import ExternalToolError, InvalidMediaError
from tscut.tools.ffprobe import (
get_frames_in_stream,
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
from tscut.tools.timeframe import get_packet_duration, parse_timestamp
logger = logging.getLogger(__name__)
@@ -46,10 +39,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp
set_inheritable(infd, True)
set_inheritable(outfd, True)
if logger.getEffectiveLevel() == logging.DEBUG:
log_level = []
else:
log_level = [ '-loglevel', 'quiet' ]
log_level = [] if logger.getEffectiveLevel() == logging.DEBUG else ['-loglevel', 'quiet']
params = [ffmpeg_path, '-y',]+log_level+['-progress', '/dev/stdout', '-canvas_size',
f'{width:d}x{height:d}', '-f', input_format,
@@ -82,7 +72,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp
@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]:
width:int=640, height:int=480) -> tuple[bytes,int]:
"""
Extract pictures from a video file using FFmpeg.
@@ -126,13 +116,13 @@ def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_
status = ffmpeg.wait()
if status != 0:
logger.error('Conversion failed with status code: %d', status)
return None, None
raise ExternalToolError("")
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
raise InvalidMediaError("")
lseek(outfd, 0, SEEK_SET)
return images, outfd
@@ -141,7 +131,7 @@ def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_
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]:
nb_channels:int=2) -> tuple[bytes,int]:
outfd = memfd_create(output_filename, flags=0)
infd = input_file.fileno()
lseek(infd, 0, SEEK_SET)
@@ -160,7 +150,7 @@ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, outpu
status = ffmpeg.wait()
if status != 0:
logger.error('Sound extraction returns error code: %d', status)
return None, None
raise ExternalToolError("")
lseek(outfd, 0, SEEK_SET)
sound = read(outfd, length)
@@ -169,7 +159,7 @@ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, outpu
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
raise InvalidMediaError("")
return sound, outfd
@@ -236,9 +226,6 @@ 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,
begin=begin, nb_frames=nb_frames,
width=width, height=height)
if images_bytes is None or memfd is None:
logger.error('Impossible to extract picture from video stream.')
exit(-1)
memfds.append(memfd)
if dump_mem_fd:
@@ -262,14 +249,10 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
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
bit_rate = int(stream['bit_rate']) if 'bit_rate' in stream else 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}',
if 'tags' in stream and '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)
@@ -294,10 +277,6 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
output_filename=tmpname,
sample_rate=sample_rate, nb_channels=nb_channels)
if sound_bytes is None or memfd is None:
logger.error('Impossible to extract sound track')
exit(-1)
memfds.append(memfd)
if dump_mem_fd:
@@ -325,9 +304,8 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
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}',
if 'tags' in stream and '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
src/tscut/tools/ffprobe.py
+13 -21
View File
@@ -4,17 +4,17 @@
import json
import logging
from math import floor
from datetime import timedelta
import os
from datetime import timedelta
from io import BytesIO
from math import floor
from os import (
SEEK_SET,
lseek,
set_inheritable,
)
from typing import IO
from subprocess import PIPE, Popen
from io import BytesIO
from typing import IO
from typeguard import typechecked
@@ -63,9 +63,8 @@ def get_frame_rate(ffprobe_path:str, input_file: IO[bytes]) -> float|None:
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 'interlaced_frame' in frame and frame['interlaced_frame'] == 1:
interlaced = True
if 'pts_time' in frame:
ts = float(frame['pts_time'])
if min_ts is None:
@@ -165,8 +164,7 @@ def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|Non
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
return timedelta(seconds=duration)
logger.error('Impossible to retrieve duration of movie')
return None
@@ -327,22 +325,22 @@ def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:ti
status = ffprobe.wait()
if status != 0:
logger.error('ffprobe failed with status code: %d', status)
return None
return
if 'frames' in frames:
frames = frames['frames']
for frame in frames:
ts = get_ts_frame(frame)
if ts is None:
return None
return
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
return None
return
@typechecked
def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes],
@@ -356,14 +354,8 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes],
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 = timestamp - delta if before else timestamp
tend = timestamp + delta if not before else timestamp
tbegin = max(tbegin, zero)
logger.debug('Looking for an iframe in [%s, %s]', tbegin, tend)
src/tscut/tools/ppm.py
+2 -2
View File
@@ -5,14 +5,14 @@
import logging
import re
from io import BytesIO
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:
"""
src/tscut/tools/timeframe.py
+2 -4
View File
@@ -65,9 +65,8 @@ def parse_timestamp(ts:str) -> timedelta|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
return timedelta(hours=hour, minutes=minute, seconds=second, microseconds=us)
@typechecked
def parse_time_interval(interval: str) -> tuple[timedelta, timedelta] | tuple[None, None]:
@@ -204,8 +203,7 @@ def get_ts_frame(frame: dict) -> timedelta|None:
logger.error('Impossible to find timestamp of frame %s', frame)
return None
ts = timedelta(seconds=pts_time)
return ts
return timedelta(seconds=pts_time)
@typechecked
def get_packet_duration(packet: dict) -> int | None:
src/tscut/tscut.py
+2 -3
View File
@@ -143,8 +143,7 @@ def concatenate_h264_parts(h264parts: list[IO[bytes]], output: IO[bytes]) -> Non
def concatenate_h264_ts_parts(h264_ts_parts: list[IO[bytes]], output: IO[bytes]) -> None:
logger = logging.getLogger(__name__)
header = '# timestamp format v2\n'.encode('ascii')
header = '# timestamp format v2\n'
output.write(header)
last = 0.
@@ -166,7 +165,7 @@ def concatenate_h264_ts_parts(h264_ts_parts: list[IO[bytes]], output: IO[bytes])
break
ts = offset + float(line)
last = max(last,ts)
output.write(f'{ts:f}\n'.encode('ascii'))
output.write(f'{ts:f}\n')
if first:
first = False