Completed
Push — dev ( 232b11...6092be )
by Konstantinos
04:15 queued 01:32
created

AudioSegmenter.target_directory()   A

Complexity

Conditions 1

Size

Total Lines 4
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 2
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 3
nop 2
dl 0
loc 4
ccs 2
cts 2
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
#!/usr/bin/python3
2
3 1
import logging
4 1
import subprocess
5 1
import tempfile
6 1
import time
7
8 1
from music_album_creation.tracks_parsing import StringParser
9
10 1
logger = logging.getLogger(__name__)
11
12
13 1
class AudioSegmenter:
14
15 1
    args = ['ffmpeg', '-i', '-acodec', 'copy', '-ss']
16
17 1
    def __init__(self, target_directory=tempfile.gettempdir()):
18 1
        self._dir = target_directory
19
20 1
    @property
21
    def target_directory(self):
22
        """The directory path that will serve as the destination for storing created tracks"""
23 1
        return self._dir
24
25 1
    @target_directory.setter
26
    def target_directory(self, directory_path):
27 1
        self._dir = directory_path
28
29 1
    def segment_from_list(self, album_file, data, supress_stdout=True, supress_stderr=True, verbose=False, sleep_seconds=0):
30
        """
31
        Given an album audio file and data structure with tracks information, segments the audio file into audio tracks which get stored in the 'self.target_directory' folder.\n
32
        :param str album_file:
33
        :param list data: list of lists. Each inner list must have 2 elements: track name and starting timestamp in hh:mm:ss
34
        :param bool supress_stdout:
35
        :param bool supress_stderr:
36
        :param bool verbose:
37
        :param float sleep_seconds:
38
        :return: full paths to audio tracks
39
        """
40
        # if not re.search('0:00', data[0][1]):
41
        #     raise NotStartingFromZeroTimestampError("First track ({}) is supposed to have a 0:00 timestamp. Instead {} found".format(data[0][0], data[0][1]))
42
43 1
        exit_code = 0
44 1
        data = StringParser.convert_tracks_data(data, album_file, target_directory=self._dir)
45 1
        audio_file_paths = [x[0] for x in data]
46 1
        i = 0
47 1
        while exit_code == 0 and i < len(data) - 1:
48 1
            time.sleep(sleep_seconds)
49 1
            exit_code = self._segment(album_file, *data[i], supress_stdout=supress_stdout, supress_stderr=supress_stderr)
50 1
            i += 1
51 1
        if exit_code != 0:
52
            raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
53 1
        exit_code = self._segment(album_file, *data[-1], supress_stdout=supress_stdout, supress_stderr=supress_stderr)
54 1
        if exit_code != 0:
55
            raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
56 1
        return audio_file_paths
57
58 1
    def segment_from_file(self, album_file, tracks_file, supress_stdout=True, supress_stderr=True, verbose=False, sleep_seconds=0.45):
59
        """
60
        Given an album audio file and a file with track information, segments the audio file into audio tracks which get stored in the 'self.target_directory' folder.\n
61
        :param str album_file:
62
        :param str tracks_file:
63
        :param bool supress_stdout:
64
        :param bool supress_stderr:
65
        :param bool verbose:
66
        :param float sleep_seconds:
67
        :return:
68
        """
69 1
        with open(tracks_file, 'r') as f:
70 1
            list_of_lists = StringParser.parse_hhmmss_string(f.read().strip())
71 1
        self.segment_from_list(album_file, list_of_lists, supress_stdout=supress_stdout, supress_stderr=supress_stderr, verbose=verbose, sleep_seconds=sleep_seconds)
72
73
74 1
    def _segment(self, *args, supress_stdout=True, supress_stderr=True, verbose=False):
75 1
        album_file = args[0]
76 1
        track_file = args[1]
77
78 1
        start = args[2]
79 1
        end = None
80 1
        if 3 < len(args):
81 1
            end = args[3]
82 1
        self._args = self.args[:2] + ['{}'.format(album_file)] + self.args[2:] + [start] + (lambda: ['-to', str(end)] if end else [])() + ['{}'.format(track_file)]
83 1
        logger.info("Segmenting: '{}'".format(' '.join(self._args)))
84 1
        ro = subprocess.run(self._args, **self.__std_parameters(supress_stdout, supress_stderr))
85 1
        return ro.returncode
86
87 1
    @classmethod
88
    def __std_parameters(cls, std_out_flag, std_error_flag):
89
        """If an input flag is True then the stream  (either 'out' or 'err') can be obtained ie obj = subprocess.run(..); str(obj.stdout, encoding='utf-8')).\n
90
        If an input flag is False then the stream will be normally outputted; ie at the terminal"""
91 1
        return {v: subprocess.PIPE for k, v in zip([std_out_flag, std_error_flag], ['stdout', 'stderr']) if k}
92
93
94
class FfmpegCommandError(Exception): pass
95
96
97 1
if __name__ == '__main__':
98
    import sys
99
100
    if len(sys.argv) < 3:
101
        print('Usage: python3 Album_segmentation.py AUDIO_FILE TRACKS_FILE')
102
        sys.exit(1)
103
    try:
104
        audio_segmenter = AudioSegmenter()
105
        audio_segmenter.segment_from_file(sys.argv[1], sys.argv[2], supress_stdout=True, supress_stderr=True, verbose=True, sleep_seconds=0.45)
106
    except Exception as e:
107
        print(e)
108
        sys.exit(1)
109