9 changed files with 865 additions and 8 deletions
-
6.gitmodules
-
1FFcuesplitter
-
1ffcuesplitter
-
353ffcuesplitter/cuesplitter.py
-
41ffcuesplitter/exceptions.py
-
230ffcuesplitter/ffmpeg.py
-
91ffcuesplitter/ffprobe.py
-
64ffcuesplitter/str_utils.py
-
86ffcuesplitter/utils.py
@ -1,6 +0,0 @@ |
|||
[submodule "FFcuesplitter"] |
|||
path = FFcuesplitter |
|||
url = https://github.com/jeanslack/FFcuesplitter.git |
|||
[submodule "deflacue"] |
|||
path = deflacue |
|||
url = https://github.com/idlesign/deflacue.git |
|||
@ -1 +0,0 @@ |
|||
FFcuesplitter/ffcuesplitter |
|||
@ -0,0 +1,353 @@ |
|||
""" |
|||
First release: January 16 2022 |
|||
|
|||
Name: cuesplitter.py |
|||
Porpose: FFmpeg based audio splitter for Cue sheet files |
|||
Platform: MacOs, Gnu/Linux, FreeBSD |
|||
Writer: jeanslack <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: February 06 2022 |
|||
Code checker: flake8 and pylint |
|||
#################################################################### |
|||
|
|||
This file is part of FFcuesplitter. |
|||
|
|||
FFcuesplitter is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
FFcuesplitter is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with FFcuesplitter. If not, see <http://www.gnu.org/licenses/>. |
|||
""" |
|||
import os |
|||
import shutil |
|||
import tempfile |
|||
import chardet |
|||
from deflacue.deflacue import CueParser |
|||
from ffcuesplitter.str_utils import msgdebug, msg |
|||
from ffcuesplitter.exceptions import (InvalidFileError, |
|||
FFCueSplitterError, |
|||
) |
|||
from ffcuesplitter.ffprobe import ffprobe |
|||
from ffcuesplitter.ffmpeg import FFMpeg |
|||
|
|||
|
|||
class FFCueSplitter(FFMpeg): |
|||
""" |
|||
This class implements an interface for retrieve the required |
|||
data to accurately split audio CD images using FFmpeg. |
|||
|
|||
Usage: |
|||
>>> from ffcuesplitter.cuesplitter import FFCueSplitter |
|||
|
|||
Splittings: |
|||
>>> split = FFCueSplitter('/home/user/my_file.cue') |
|||
>>> split.open_cuefile() |
|||
>>> split.do_operations() |
|||
|
|||
Get data tracks: |
|||
>>> data = FFCueSplitter('/home/user/other.cue', dry=True) |
|||
>>> data.open_cuefile() |
|||
>>> data.audiotracks # trackdata |
|||
>>> data.cue.meta.data # cd_info |
|||
>>> data.ffmpeg_arguments() |
|||
|
|||
Only processing the track three: |
|||
>>> myfile = FFCueSplitter('/home/user/my_file.cue', |
|||
progress_meter='tqdm') |
|||
>>> f.open_cuefile() |
|||
>>> f.kwargs['tempdir'] = '/tmp/mytempdir' |
|||
>>> f.ffmpeg_arguments() |
|||
>>> f.processing(myfile.arguments[2], myfile.seconds[2]) |
|||
>>> f.move_files_to_outputdir() |
|||
|
|||
For a full meaning of the arguments to pass to the instance, read |
|||
the __init__ docstring of this class. |
|||
|
|||
""" |
|||
def __init__(self, |
|||
filename, |
|||
outputdir=str('.'), |
|||
suffix=str('flac'), |
|||
overwrite=str('ask'), |
|||
ffmpeg_cmd=str('ffmpeg'), |
|||
ffmpeg_loglevel=str('info'), |
|||
ffprobe_cmd=str('ffprobe'), |
|||
ffmpeg_add_params=str(''), |
|||
progress_meter=str('standard'), |
|||
dry=bool(False) |
|||
): |
|||
""" |
|||
------------------ |
|||
Arguments meaning: |
|||
------------------ |
|||
|
|||
filename: |
|||
absolute or relative CUE sheet file |
|||
outputdir: |
|||
absolute or relative pathname to output files |
|||
suffix: |
|||
output format, one of ("wav", "flac", "mp3", "ogg") . |
|||
overwrite: |
|||
overwriting options, one of "ask", "never", "always". |
|||
ffmpeg_cmd: |
|||
an absolute path command of ffmpeg |
|||
ffmpeg_loglevel: |
|||
one of "error", "warning", "info", "verbose", "debug" . |
|||
ffprobe_cmd: |
|||
an absolute path command of ffprobe. |
|||
ffmpeg_add_params: |
|||
additionals parameters of FFmpeg. |
|||
progress_meter: |
|||
one of 'tqdm', 'standard', default is 'standard'. |
|||
dry: |
|||
with `True`, perform the dry run with no changes |
|||
done to filesystem. |
|||
""" |
|||
super().__init__() |
|||
|
|||
self.kwargs = {'filename': os.path.abspath(filename)} |
|||
self.kwargs['dirname'] = os.path.dirname(self.kwargs['filename']) |
|||
if outputdir == '.': |
|||
self.kwargs['outputdir'] = self.kwargs['dirname'] |
|||
else: |
|||
self.kwargs['outputdir'] = os.path.abspath(outputdir) |
|||
self.kwargs['format'] = suffix |
|||
self.kwargs['overwrite'] = overwrite |
|||
self.kwargs['ffmpeg_cmd'] = ffmpeg_cmd |
|||
self.kwargs['ffmpeg_loglevel'] = ffmpeg_loglevel |
|||
self.kwargs['ffprobe_cmd'] = ffprobe_cmd |
|||
self.kwargs['ffmpeg_add_params'] = ffmpeg_add_params |
|||
self.kwargs['progress_meter'] = progress_meter |
|||
self.kwargs['dry'] = dry |
|||
self.kwargs['logtofile'] = os.path.join(self.kwargs['dirname'], |
|||
'ffcuesplitter.log') |
|||
self.kwargs['tempdir'] = '.' |
|||
self.audiotracks = None |
|||
self.probedata = [] |
|||
self.cue_encoding = None # data chardet |
|||
self.cue = None |
|||
self.testpatch = None # set for test cases only |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def move_files_to_outputdir(self): |
|||
""" |
|||
All files are processed in a /temp folder. After the split |
|||
operation is complete, all tracks are moved from /temp folder |
|||
to output folder. Here evaluates what to do if files already |
|||
exists on output folder. |
|||
|
|||
Raises: |
|||
FFCueSplitterError |
|||
Returns: |
|||
None |
|||
|
|||
""" |
|||
outputdir = self.kwargs['outputdir'] |
|||
overwr = self.kwargs['overwrite'] |
|||
|
|||
for track in os.listdir(self.kwargs['tempdir']): |
|||
if os.path.exists(os.path.join(outputdir, track)): |
|||
if overwr in ('n', 'N', 'y', 'Y', 'ask'): |
|||
while True: |
|||
msgdebug(warn=f"File already exists: " |
|||
f"'{os.path.join(outputdir, track)}'") |
|||
overwr = input("Overwrite [Y/n/always/never]? > ") |
|||
if overwr in ('Y', 'y', 'n', 'N', 'always', 'never'): |
|||
break |
|||
msgdebug(err=f"Invalid option '{overwr}'") |
|||
continue |
|||
if overwr == 'never': |
|||
msgdebug(info=("Do not overwrite any files because " |
|||
"you specified 'never' option")) |
|||
return None |
|||
|
|||
if overwr in ('y', 'Y', 'always', 'never', 'ask'): |
|||
if overwr == 'always': |
|||
msgdebug(info=("Overwrite existing file because " |
|||
"you specified the 'always' option")) |
|||
try: |
|||
shutil.move(os.path.join(self.kwargs['tempdir'], track), |
|||
os.path.join(outputdir, track)) |
|||
|
|||
except Exception as error: |
|||
raise FFCueSplitterError(error) from error |
|||
|
|||
return None |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def do_operations(self): |
|||
""" |
|||
Automates the work in a temporary context using tempfile. |
|||
|
|||
Raises: |
|||
FFCueSplitterError |
|||
Returns: |
|||
None |
|||
""" |
|||
with tempfile.TemporaryDirectory(suffix=None, |
|||
prefix='ffcuesplitter_', |
|||
dir=None) as tmpdir: |
|||
self.kwargs['tempdir'] = tmpdir |
|||
self.ffmpeg_arguments() |
|||
|
|||
msgdebug(info=(f"Temporary Target: '{self.kwargs['tempdir']}'")) |
|||
count = 0 |
|||
msgdebug(info="Extracting audio tracks (type Ctrl+c to stop):") |
|||
|
|||
for args, secs, title in zip(self.arguments, |
|||
self.seconds, |
|||
self.audiotracks): |
|||
count += 1 |
|||
msg(f'\nTRACK {count}/{len(self.audiotracks)} ' |
|||
f'>> "{title["TITLE"]}.{self.outsuffix}" ...') |
|||
self.processing(args, secs) |
|||
|
|||
if self.kwargs['dry'] is True: |
|||
return |
|||
|
|||
msg('\n') |
|||
msgdebug(info="...done exctracting") |
|||
msgdebug(info="Move files to: ", |
|||
tail=(f"\033[34m" |
|||
f"'{os.path.abspath(self.kwargs['outputdir'])}'" |
|||
f"\033[0m")) |
|||
try: |
|||
os.makedirs(self.kwargs['outputdir'], |
|||
mode=0o777, exist_ok=True) |
|||
except Exception as error: |
|||
raise FFCueSplitterError(error) from error |
|||
|
|||
self.move_files_to_outputdir() |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def get_track_duration(self, tracks): |
|||
""" |
|||
Gets total duration of the source audio tracks for chunks |
|||
calculation on the progress meter during ffmpeg executions. |
|||
Given a total duration calculates the remains duration |
|||
for the last track as well. |
|||
|
|||
This method is called by `cuefile_parser` method, Do not |
|||
call this method directly. |
|||
|
|||
Raises: |
|||
FFCueSplitterError |
|||
Returns: |
|||
tracks (list), all track data taken from the cue file. |
|||
""" |
|||
if self.testpatch: |
|||
probe = {'format': {'duration': 6.000000}} |
|||
else: |
|||
filename = tracks[0].get('FILE') |
|||
cmd = self.kwargs['ffprobe_cmd'] |
|||
kwargs = {'loglevel': 'error', 'hide_banner': None} |
|||
probe = ffprobe(filename, cmd=cmd, **kwargs) |
|||
self.probedata.append(probe) |
|||
|
|||
time = [] |
|||
for idx in enumerate(tracks): |
|||
if idx[0] != len(tracks) - 1: # minus last |
|||
trk = (tracks[idx[0] + 1]['START'] - |
|||
tracks[idx[0]]['START']) / (44100) |
|||
time.append(trk) |
|||
|
|||
if not time: |
|||
last = (float(probe['format']['duration']) - |
|||
tracks[0]['START'] / 44100) |
|||
else: |
|||
last = float(probe['format']['duration']) - sum(time) |
|||
time.append(last) |
|||
for keydur, remain in zip(tracks, time): |
|||
keydur['DURATION'] = remain |
|||
|
|||
return tracks |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def deflacue_object_handler(self): |
|||
""" |
|||
Handles `deflacue.CueParser` data. |
|||
Raises: |
|||
FFCueSplitterError: if no source audio file found |
|||
Returns: |
|||
'audiotracks' list object |
|||
""" |
|||
self.audiotracks = [] |
|||
cd_info = self.cue.meta.data |
|||
|
|||
def sanitize(val: str) -> str: |
|||
return val.replace('/', '').replace('\\','').replace('"', '') |
|||
|
|||
tracks = self.cue.tracks |
|||
sourcenames = {k: [] for k in [str(x.file.path) for x in tracks]} |
|||
|
|||
for track in enumerate(tracks): |
|||
track_file = track[1].file.path |
|||
|
|||
if not track_file.exists(): |
|||
msgdebug(warn=(f'Source file `{track_file}` is not ' |
|||
f'found. Track is skipped.')) |
|||
|
|||
if str(track_file) in sourcenames: |
|||
sourcenames.pop(str(track_file)) |
|||
if not sourcenames: |
|||
raise FFCueSplitterError('No audio source files ' |
|||
'found!') |
|||
continue |
|||
|
|||
filename = (f"{sanitize(track[1].title)}") |
|||
|
|||
data = {'FILE': str(track_file), **cd_info, **track[1].data} |
|||
data['TITLE'] = filename |
|||
data['START'] = track[1].start |
|||
|
|||
if track[1].end != 0: |
|||
data['END'] = track[1].end |
|||
|
|||
if f"{data['FILE']}" in sourcenames.keys(): |
|||
sourcenames[f'{data["FILE"]}'].append(data) |
|||
|
|||
for val in sourcenames.values(): |
|||
self.audiotracks += self.get_track_duration(val) |
|||
|
|||
return self.audiotracks |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def check_cuefile(self): |
|||
""" |
|||
Cue file check |
|||
""" |
|||
filesuffix = os.path.splitext(self.kwargs['filename'])[1] |
|||
isfile = os.path.isfile(self.kwargs['filename']) |
|||
|
|||
if not isfile or filesuffix not in ('.cue', '.CUE'): |
|||
raise InvalidFileError(f"Invalid CUE sheet file: " |
|||
f"'{self.kwargs['filename']}'") |
|||
# ----------------------------------------------------------------# |
|||
|
|||
def open_cuefile(self, testpatch=None): |
|||
""" |
|||
Read cue file and start file parsing via deflacue package |
|||
""" |
|||
if testpatch: |
|||
self.testpatch = True |
|||
|
|||
self.check_cuefile() |
|||
curdir = os.getcwd() |
|||
os.chdir(self.kwargs['dirname']) |
|||
|
|||
with open(self.kwargs['filename'], 'rb') as file: |
|||
cuebyte = file.read() |
|||
self.cue_encoding = chardet.detect(cuebyte) |
|||
|
|||
parser = CueParser.from_file(self.kwargs['filename'], |
|||
encoding=self.cue_encoding['encoding']) |
|||
self.cue = parser.run() |
|||
self.deflacue_object_handler() |
|||
os.chdir(curdir) |
|||
@ -0,0 +1,41 @@ |
|||
""" |
|||
Name: exceptions.py |
|||
Porpose: defines class Exceptions for ffcuesplitter |
|||
Platform: MacOs, Gnu/Linux, FreeBSD |
|||
Writer: jeanslack <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: February 03 2022 |
|||
Code checker: flake8 and pylint |
|||
#################################################################### |
|||
|
|||
This file is part of FFcuesplitter. |
|||
|
|||
FFcuesplitter is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
FFcuesplitter is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with FFcuesplitter. If not, see <http://www.gnu.org/licenses/>. |
|||
""" |
|||
|
|||
|
|||
class FFMpegError(Exception): |
|||
"""Excepion raised by FFMpeg class""" |
|||
|
|||
|
|||
class FFProbeError(Exception): |
|||
"""Excepion raised by FFProbe class""" |
|||
|
|||
|
|||
class InvalidFileError(Exception): |
|||
"""Exception type raised when CUE file is invalid.""" |
|||
|
|||
|
|||
class FFCueSplitterError(Exception): |
|||
"""Exception raised in all other cases.""" |
|||
@ -0,0 +1,230 @@ |
|||
# -*- coding: UTF-8 -*- |
|||
""" |
|||
Name: ffmpeg.py |
|||
Porpose: builds arguments for FFmpeg processing. |
|||
Compatibility: Python3 |
|||
Platform: all platforms |
|||
Author: Gianluca Pernigotto <jeanlucperni@gmail.com> |
|||
Copyright: (c) 2022/2023 Gianluca Pernigotto <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: February 06 2022 |
|||
Code checker: flake8, pylint |
|||
######################################################## |
|||
|
|||
This file is part of FFcuesplitter. |
|||
|
|||
FFcuesplitter is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
FFcuesplitter is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with FFcuesplitter. If not, see <http://www.gnu.org/licenses/>. |
|||
""" |
|||
import subprocess |
|||
import os |
|||
import sys |
|||
import platform |
|||
from tqdm import tqdm |
|||
from ffcuesplitter.exceptions import FFMpegError |
|||
from ffcuesplitter.str_utils import msg |
|||
from ffcuesplitter.utils import Popen |
|||
|
|||
if not platform.system() == 'Windows': |
|||
import shlex |
|||
|
|||
|
|||
class FFMpeg: |
|||
""" |
|||
FFMpeg is a parent base class interface for FFCueSplitter. |
|||
It represents FFmpeg command and arguments with their |
|||
sub-processing. |
|||
""" |
|||
DATACODECS = {'wav': 'pcm_s16le -ar 44100', |
|||
'flac': 'flac -ar 44100', |
|||
'ogg': 'libvorbis -ar 44100', |
|||
'mp3': 'libmp3lame -ar 44100', |
|||
} |
|||
|
|||
def __init__(self, **kwargs): |
|||
""" |
|||
Constructor |
|||
""" |
|||
self.kwargs = kwargs |
|||
self.audiotracks = kwargs |
|||
self.seconds = None |
|||
self.arguments = None |
|||
self.osplat = platform.system() |
|||
self.outsuffix = None |
|||
# -------------------------------------------------------------# |
|||
|
|||
def codec_setup(self, sourcef): |
|||
""" |
|||
Returns codec arg based on given format |
|||
""" |
|||
if self.kwargs['format'] == 'copy': |
|||
self.outsuffix = os.path.splitext(sourcef)[1].replace('.', '') |
|||
codec = '-c copy' |
|||
|
|||
else: |
|||
self.outsuffix = self.kwargs['format'] |
|||
codec = f'-c:a {FFMpeg.DATACODECS[self.kwargs["format"]]}' |
|||
|
|||
return codec, self.outsuffix |
|||
# -------------------------------------------------------------# |
|||
|
|||
def ffmpeg_arguments(self): |
|||
""" |
|||
Builds `FFmpeg` arguments and calculates time seconds |
|||
for each audio track. |
|||
|
|||
Returns: |
|||
dict(arguments:[...], seconds:[...]) |
|||
""" |
|||
self.arguments = [] |
|||
self.seconds = [] |
|||
|
|||
meters = {'tqdm': '-progress pipe:1 -nostats -nostdin', 'standard': ''} |
|||
|
|||
for track in self.audiotracks: |
|||
codec, suffix = self.codec_setup(track["FILE"]) |
|||
metadata = {'ARTIST': track.get('PERFORMER', ''), |
|||
'ALBUM': track.get('ALBUM', ''), |
|||
'TITLE': track.get('TITLE', ''), |
|||
'TRACK': (str(track['TRACK_NUM']) + '/' + |
|||
str(len(self.audiotracks))), |
|||
'DISCNUMBER': track.get('DISCNUMBER', ''), |
|||
'GENRE': track.get('GENRE', ''), |
|||
'DATE': track.get('DATE', ''), |
|||
'COMMENT': track.get('COMMENT', ''), |
|||
'DISCID': track.get('DISCID', ''), |
|||
} |
|||
cmd = f'"{self.kwargs["ffmpeg_cmd"]}" ' |
|||
cmd += f' -loglevel {self.kwargs["ffmpeg_loglevel"]}' |
|||
cmd += f" {meters[self.kwargs['progress_meter']]}" |
|||
fpath = os.path.join(self.kwargs["dirname"], track["FILE"]) |
|||
cmd += f' -i "{fpath}"' |
|||
cmd += f" -ss {round(track['START'] / 44100, 6)}" # ff to secs |
|||
if 'END' in track: |
|||
cmd += f" -to {round(track['END'] / 44100, 6)}" # ff to secs |
|||
for key, val in metadata.items(): |
|||
cmd += f' -metadata {key}="{val}"' |
|||
cmd += f' {codec}' |
|||
cmd += f" {self.kwargs['ffmpeg_add_params']}" |
|||
cmd += ' -y' |
|||
num = str(track['TRACK_NUM']).rjust(2, '0') |
|||
name = f'{num} - {track["TITLE"]}.{suffix}' |
|||
cmd += f' "{os.path.join(self.kwargs["tempdir"], name)}"' |
|||
self.arguments.append(cmd) |
|||
self.seconds.append(track['DURATION']) |
|||
|
|||
return {'arguments': self.arguments, 'seconds': self.seconds} |
|||
# --------------------------------------------------------------# |
|||
|
|||
def processing(self, arg, secs): |
|||
""" |
|||
Redirect to required processing |
|||
""" |
|||
if self.kwargs['progress_meter'] == 'tqdm': |
|||
cmd = arg if self.osplat == 'Windows' else shlex.split(arg) |
|||
if self.kwargs['dry'] is True: |
|||
msg(cmd) # stdout cmd in dry mode |
|||
return |
|||
self.processing_with_tqdm_progress(cmd, secs) |
|||
|
|||
elif self.kwargs['progress_meter'] == 'standard': |
|||
cmd = arg if self.osplat == 'Windows' else shlex.split(arg) |
|||
if self.kwargs['dry'] is True: |
|||
msg(cmd) # stdout cmd in dry mode |
|||
return |
|||
self.processing_with_standard_progress(cmd) |
|||
# --------------------------------------------------------------# |
|||
|
|||
def processing_with_tqdm_progress(self, cmd, seconds): |
|||
""" |
|||
FFmpeg sub-processing showing a tqdm progress meter |
|||
for each loop. Also writes a log file to the same |
|||
destination folder as the .cue file . |
|||
|
|||
Usage for get seconds elapsed: |
|||
progbar = tqdm(total=round(seconds), unit="s", dynamic_ncols=True) |
|||
progbar.clear() |
|||
previous_s = 0 |
|||
|
|||
s_processed = round(int(output.split('=')[1]) / 1_000_000) |
|||
s_increase = s_processed - previous_s |
|||
progbar.update(s_increase) |
|||
previous_s = s_processed |
|||
|
|||
Raises: |
|||
FFMpegError |
|||
Returns: |
|||
None |
|||
""" |
|||
progbar = tqdm(total=100, |
|||
unit="s", |
|||
dynamic_ncols=True |
|||
) |
|||
progbar.clear() |
|||
|
|||
try: |
|||
with open(self.kwargs['logtofile'], "w", encoding='utf-8') as log: |
|||
log.write(f'\nCOMMAND: {cmd}') |
|||
with Popen(cmd, |
|||
stdout=subprocess.PIPE, |
|||
stderr=log, |
|||
bufsize=1, |
|||
universal_newlines=True) as proc: |
|||
|
|||
for output in proc.stdout: |
|||
if "out_time_ms" in output.strip(): |
|||
s_processed = int(output.split('=')[1]) / 1_000_000 |
|||
percent = s_processed / seconds * 100 |
|||
progbar.update(round(percent) - progbar.n) |
|||
|
|||
if proc.wait(): # error |
|||
progbar.close() |
|||
raise FFMpegError(f"ffmpeg FAILED: See log details: " |
|||
f"'{self.kwargs['logtofile']}'" |
|||
f"\nExit status: {proc.wait()}") |
|||
|
|||
except (OSError, FileNotFoundError) as excepterr: |
|||
progbar.close() |
|||
raise FFMpegError(excepterr) from excepterr |
|||
|
|||
except KeyboardInterrupt: |
|||
# proc.kill() |
|||
progbar.close() |
|||
proc.terminate() |
|||
sys.exit("\n[KeyboardInterrupt] FFmpeg process terminated.") |
|||
|
|||
progbar.close() |
|||
# --------------------------------------------------------------# |
|||
|
|||
def processing_with_standard_progress(self, cmd): |
|||
""" |
|||
FFmpeg sub-processing with stderr output to console. |
|||
The output depending on the ffmpeg loglevel option. |
|||
Raises: |
|||
FFMpegError |
|||
Returns: |
|||
None |
|||
""" |
|||
with open(self.kwargs['logtofile'], "w", encoding='utf-8') as log: |
|||
log.write(f'COMMAND: {cmd}') |
|||
try: |
|||
subprocess.run(cmd, check=True, shell=False) |
|||
|
|||
except FileNotFoundError as err: |
|||
raise FFMpegError(f"{err}") from err |
|||
|
|||
except subprocess.CalledProcessError as err: |
|||
raise FFMpegError(f"ffmpeg FAILED: {err}") from err |
|||
|
|||
except KeyboardInterrupt: |
|||
sys.exit("\n[KeyboardInterrupt]") |
|||
@ -0,0 +1,91 @@ |
|||
# -*- coding: UTF-8 -*- |
|||
""" |
|||
Name: ffprobe.py |
|||
Porpose: simple cross-platform wrap for ffprobe |
|||
Compatibility: Python3 |
|||
Platform: all platforms |
|||
Author: Gianluca Pernigotto <jeanlucperni@gmail.com> |
|||
Copyright: (c) 2022/2023 Gianluca Pernigotto <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: Feb.17.2022 |
|||
Code checker: flake8, pylint |
|||
######################################################## |
|||
|
|||
This file is part of FFcuesplitter. |
|||
|
|||
FFcuesplitter is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
FFcuesplitter is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with FFcuesplitter. If not, see <http://www.gnu.org/licenses/>. |
|||
""" |
|||
import subprocess |
|||
import shlex |
|||
import platform |
|||
import json |
|||
from ffcuesplitter.exceptions import FFProbeError |
|||
from ffcuesplitter.utils import Popen |
|||
|
|||
|
|||
def from_kwargs_to_args(kwargs): |
|||
""" |
|||
Helper function to build command line |
|||
arguments out of dict. |
|||
""" |
|||
args = [] |
|||
for key in sorted(kwargs.keys()): |
|||
val = kwargs[key] |
|||
args.append(f'-{key}') |
|||
if val is not None: |
|||
args.append(f'{val}') |
|||
return args |
|||
|
|||
|
|||
def ffprobe(filename, cmd='ffprobe', **kwargs): |
|||
""" |
|||
Run ffprobe on the specified file and return a |
|||
JSON representation of the output. |
|||
|
|||
Raises: |
|||
:class:`ffcuesplitter.FFProbeError`: if ffprobe |
|||
returns a non-zero exit code; |
|||
`ffcuesplitter.FFProbeError` from `OSError`, |
|||
`FileNotFoundError` if a generic error. |
|||
|
|||
Usage: |
|||
ffprobe(filename, |
|||
cmd='/usr/bin oi/ffprobe', |
|||
loglevel='error', |
|||
hide_banner=None, |
|||
etc, |
|||
) |
|||
""" |
|||
args = (f'"{cmd}" -show_format -show_streams -of json ' |
|||
f'{" ".join(from_kwargs_to_args(kwargs))} ' |
|||
f'"{filename}"' |
|||
) |
|||
args = shlex.split(args) if platform.system() != 'Windows' else args |
|||
|
|||
try: |
|||
with Popen(args, |
|||
stdout=subprocess.PIPE, |
|||
stderr=subprocess.PIPE, |
|||
universal_newlines=True, |
|||
) as proc: |
|||
output, error = proc.communicate() |
|||
|
|||
if proc.returncode != 0: |
|||
raise FFProbeError(f'ffprobe: {error}') |
|||
|
|||
except (OSError, FileNotFoundError) as excepterr: |
|||
raise FFProbeError(excepterr) from excepterr |
|||
|
|||
else: |
|||
return json.loads(output) |
|||
@ -0,0 +1,64 @@ |
|||
# -*- coding: utf-8 -*- |
|||
""" |
|||
Name: str_utils.py (module) |
|||
Porpose: module for cosmetic output console in ANSI sequences |
|||
Writer: Gianluca Pernigoto <jeanlucperni@gmail.com> |
|||
Copyright: (c) 2022 Gianluca Pernigoto <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: Jan 10 2022 |
|||
Code checker: flake8, pylint |
|||
""" |
|||
|
|||
|
|||
def msgdebug(head='', info=None, warn=None, err=None, tail=''): |
|||
""" |
|||
Print debug messages: |
|||
``head`` can be used for additionals custom string to use |
|||
at the beginning of the string. |
|||
``tail`` can be used for additionals custom string to use |
|||
at the end of the string. |
|||
``info`` print in blue color, ``warn`` print in yellow color |
|||
``err`` print in red color. |
|||
""" |
|||
if info: |
|||
print(f"{head}\033[32;1mINFO:\033[0m {info}{tail}") |
|||
elif warn: |
|||
print(f"{head}\033[33;1mWARNING:\033[0m {warn}{tail}") |
|||
elif err: |
|||
print(f"{head}\033[31;1mERROR:\033[0m {err}{tail}") |
|||
|
|||
|
|||
def msgcolor(head='', orange=None, green=None, azure=None, tail=''): |
|||
""" |
|||
Print information messages by explicitly |
|||
choosing the name of the color to be displayed: |
|||
``head`` can be used for additionals custom string to use |
|||
at the beginning of the string. |
|||
``tail`` can be used for additionals custom string to use |
|||
at the end of the string. |
|||
""" |
|||
if orange: |
|||
print(f"{head}\033[33;1m{orange}\033[0m{tail}") |
|||
|
|||
elif green: |
|||
print(f"{head}\033[32;1m{green}\033[0m{tail}") |
|||
|
|||
elif azure: |
|||
print(f"{head}\033[34;1m{azure}\033[0m{tail}") |
|||
|
|||
|
|||
def msgend(done=None, abort=None): |
|||
""" |
|||
Print status messages at the end of the tasks |
|||
""" |
|||
if done: |
|||
print("\033[1m..Finished!\033[0m\n") |
|||
elif abort: |
|||
print("\033[1m..Abort!\033[0m\n") |
|||
|
|||
|
|||
def msg(message): |
|||
""" |
|||
Print any string messages |
|||
""" |
|||
print(message) |
|||
@ -0,0 +1,86 @@ |
|||
""" |
|||
Name: utils.py |
|||
Porpose: utils used by FFcuesplitter |
|||
Platform: MacOs, Gnu/Linux, FreeBSD |
|||
Writer: jeanslack <jeanlucperni@gmail.com> |
|||
license: GPL3 |
|||
Rev: January 16 2022 |
|||
Code checker: flake8 and pylint |
|||
#################################################################### |
|||
|
|||
This file is part of FFcuesplitter. |
|||
|
|||
FFcuesplitter is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
FFcuesplitter is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with FFcuesplitter. If not, see <http://www.gnu.org/licenses/>. |
|||
""" |
|||
import subprocess |
|||
import platform |
|||
import datetime |
|||
|
|||
|
|||
def pairwise(iterable): |
|||
""" |
|||
Return a zip object from iterable. |
|||
This function is used by run method. |
|||
---- |
|||
USE: |
|||
|
|||
after splitting ffmpeg's progress strings such as: |
|||
output = "frame= 1178 fps=155 q=29.0 size= 2072kB time=00:00:39.02 |
|||
bitrate= 435.0kbits/s speed=5.15x " |
|||
in a list as: |
|||
iterable = ['frame', '1178', 'fps', '155', 'q', '29.0', 'size', '2072kB', |
|||
'time', '00:00:39.02', 'bitrate', '435.0kbits/s', speed', |
|||
'5.15x'] |
|||
for x, y in pairwise(iterable): |
|||
(x,y) |
|||
|
|||
<https://stackoverflow.com/questions/5389507/iterating-over-every- |
|||
two-elements-in-a-list> |
|||
|
|||
""" |
|||
itobj = iter(iterable) # list_iterator object |
|||
return zip(itobj, itobj) # zip object pairs from list iterable object |
|||
# ------------------------------------------------------------------------ |
|||
|
|||
|
|||
def frames_to_seconds(frames): |
|||
""" |
|||
Converts frames (10407600) to seconds (236.0) and then |
|||
converts them to a time format string (0:03:56) using datetime. |
|||
""" |
|||
secs = frames / 44100 |
|||
return str(datetime.timedelta(seconds=secs)) |
|||
# ------------------------------------------------------------------------ |
|||
|
|||
|
|||
class Popen(subprocess.Popen): |
|||
""" |
|||
Inherit subprocess.Popen class to set _startupinfo. |
|||
This avoids displaying a console window on MS-Windows |
|||
using GUI's . |
|||
""" |
|||
if platform.system() == 'Windows': |
|||
_startupinfo = subprocess.STARTUPINFO() |
|||
_startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW |
|||
else: |
|||
_startupinfo = None |
|||
|
|||
def __init__(self, *args, **kwargs): |
|||
"""Constructor |
|||
""" |
|||
super().__init__(*args, **kwargs, startupinfo=self._startupinfo) |
|||
|
|||
# def communicate_or_kill(self, *args, **kwargs): |
|||
# return process_communicate_or_kill(self, *args, **kwargs) |
|||
# ------------------------------------------------------------------------ |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue