music_album_creation.audio_segmentation.album_segmentation.AudioSegmenter.segment() - Code Metrics - Inspection of "fix some CWEs and add dev scripts" - boromir674/music-album-creator - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — dev ( 3a28f6...0451e7 )

by Konstantinos

created 2023-03-19 21:03 UTC

AudioSegmenter.segment() B

↳ Parent: music_album_creation.audio_segmentation.album_segmentation

Complexity

Conditions

Size

Total Lines	40
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	11
CRAP Score	6.8667

Importance

Changes

Metric	Value
cc	5
eloc	28
nop	6
dl	0
loc	40
ccs	11
cts	19
cp	0.5789
crap	6.8667
rs	8.7413
c	0
b	0
f	0

import logging
import os
import tempfile
import time
from typing import Optional

from music_album_creation.ffmpeg import FFMPEG
from music_album_creation.tracks_parsing import StringParser

from .data import SegmentationInformation

logger = logging.getLogger(__name__)


ffmpeg = FFMPEG(os.environ.get('MUSIC_FFMPEG', 'ffmpeg'))


class AudioSegmenter(object):
    def __init__(self, target_directory=tempfile.gettempdir()):
        self._dir = target_directory

    @property
    def target_directory(self):
        """The directory path that will serve as the destination for storing created tracks"""
        return self._dir

    @target_directory.setter
    def target_directory(self, directory_path):
        self._dir = directory_path

    def _trans(self, track_info):
        return [os.path.join(self._dir, '{}.mp4'.format(track_info[0]))] + track_info[1:]

    def segment(
        self, album_file, data, supress_stdout=True, supress_stderr=True, sleep_seconds=0
    ):
        """

        :param album_file:
        :param data:
        :param supress_stdout:
        :param supress_stderr:
        :param float sleep_seconds:
        :return:
        """
        exit_code = 0
        i = 0
        while exit_code == 0 and i < len(data) - 1:
            time.sleep(sleep_seconds)
            result = self._segment(
                album_file,
                *self._trans(list(data[i])),
                supress_stdout=supress_stdout,
                supress_stderr=supress_stderr,
            )
            i += 1
        if result.exit_code != 0:

            logger.error("Fmmpeg exit code: %s", result.exit_code)
            logger.error("Ffmpeg st out: %s", result.stdout)
            logger.error("Fmmpeg stderr: %s", result.stderr)
            raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
        result = self._segment(
            album_file,
            *self._trans(list(data[-1])),
            supress_stdout=supress_stdout,
            supress_stderr=supress_stderr,
        )
        if result.exit_code != 0:
            logger.error("Fmmpeg exit code: %s", result.exit_code)
            logger.error("Ffmpeg st out: %s", result.stdout)
            logger.error("Fmmpeg stderr: %s", result.stderr)
            raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
        return [os.path.join(self._dir, '{}.mp4'.format(list(x)[0])) for x in data]

    def segment_from_list(
        self, album_file, data, supress_stdout=True, supress_stderr=True, sleep_seconds=0
    ):
        """
        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
        :param str album_file:
        :param list data: list of lists. Each inner list must have 2 elements: track name and starting timestamp in hh:mm:ss
        :param bool supress_stdout:
        :param bool supress_stderr:
        :param bool verbose:
        :param float sleep_seconds:
        :return: full paths to audio tracks
        """
        # if not re.search('0:00', data[0][1]):
        #     raise NotStartingFromZeroTimestampError("First track ({}) is supposed to have a 0:00 timestamp. Instead {} found".format(data[0][0], data[0][1]))

        exit_code = 0
        data = StringParser().convert_tracks_data(data, album_file, target_directory=self._dir)
        audio_file_paths = [x[0] for x in data]
        i = 0
        while exit_code == 0 and i < len(data) - 1:
            time.sleep(sleep_seconds)
            result = self._segment(
                album_file,
                *data[i],
                supress_stdout=supress_stdout,
                supress_stderr=supress_stderr,
            )
            if result.exit_code != 0:
                logger.error("Fmmpeg exit code: %s", result.exit_code)
                logger.error("Ffmpeg st out: %s", result.stdout)
                logger.error("Fmmpeg stderr: %s", result.stderr)
                raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
            i += 1
        # if exit_code != 0:
        #     raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
        result = self._segment(
            album_file, *data[-1], supress_stdout=supress_stdout, supress_stderr=supress_stderr
        )
        if result.exit_code != 0:
            logger.error("Fmmpeg exit code: %s", result.exit_code)
            logger.error("Ffmpeg st out: %s", result.stdout)
            logger.error("Fmmpeg stderr: %s", result.stderr)
            raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
        # if exit_code != 0:
        #     raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
        return audio_file_paths

    def segment_from_file(
        self,
        album_file,
        tracks_file,
        hhmmss_type,
        supress_stdout=True,
        supress_stderr=True,
        sleep_seconds=0,
    ):
        """
        Assumes the hhmmss are starting timestamps. 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
        :param str album_file:
        :param str tracks_file:
        :param str hhmmss_type:
        :param bool supress_stdout:
        :param bool supress_stderr:
        :param float sleep_seconds:
        :return:
        """
        with open(tracks_file, 'r') as f:
            segmentation_info = SegmentationInformation.from_multiline(
                f.read().strip(), hhmmss_type
            )
        return self.segment(
            album_file,
            segmentation_info,
            supress_stdout=supress_stdout,
            supress_stderr=supress_stderr,
            sleep_seconds=sleep_seconds,
        )

    def _segment(self, *args, **kwargs):
        album_file = args[0]
        track_file = args[1]
        # supress_stdout = kwargs['supress_stdout']
        # supress_stderr = kwargs['supress_stderr']

        start = args[2]  # starting timestamp
        end: Optional[str] = None  # end timestamp
        # if it is the last segment then end timestamp is the end of the album
        # and in that case the client code need to supply 3 arguments (not 4)
        if 3 < len(args):
            end = args[3]

        # args = ['ffmpeg', '-y', '-i', '-acodec', 'copy', '-ss']
        # self._args = args[:3] + ['{}'.format(album_file)] + args[3:] + [start] + (lambda: ['-to', str(end)] if end else [])() + ['{}'.format(track_file)]
        self._args = (
            '-y',
            '-i',
            str(album_file),  # youtube downloaded audio steam (ie local mp4 file)
            '-acodec',
            'copy',
            '-ss',
            start,
            *list((lambda: ['-to', str(end)] if end else [])()),
            str(track_file),
        )
        logger.info("Segmenting: ffmpeg '{}'".format(' '.join(self._args)))
        return ffmpeg(
            *self._args,
        )
        # return subprocess.check_call(self._args, **self.__std_parameters(supress_stdout, supress_stderr))
        # ro = subprocess.run(self._args, **self.__std_parameters(supress_stdout, supress_stderr))
        # return ro.returncode

    # @classmethod
    # def __std_parameters(cls, std_out_flag, std_error_flag):
    #     """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
    #     If an input flag is False then the stream will be normally outputted; ie at the terminal"""
    #     return {v: subprocess.PIPE for k, v in zip([std_out_flag, std_error_flag], ['stdout', 'stderr']) if k}


class FfmpegCommandError(Exception):
    pass


if __name__ == '__main__':
    import sys

    if len(sys.argv) < 3:
        print('Usage: python3 Album_segmentation.py AUDIO_FILE TRACKS_FILE')
        sys.exit(1)
    try:
        audio_segmenter = AudioSegmenter()
        audio_segmenter.segment_from_file(
            sys.argv[1],
            sys.argv[2],
            supress_stdout=True,
            supress_stderr=True,
            verbose=True,
            sleep_seconds=0.45,
        )
    except Exception as e:
        print(e)
        sys.exit(1)


1	1	import logging
2	1	import os
3	1	import tempfile
4	1	import time
5	1	from typing import Optional
6
7	1	from music_album_creation.ffmpeg import FFMPEG
8		from music_album_creation.tracks_parsing import StringParser
9	1
10		from .data import SegmentationInformation
11	1
12		logger = logging.getLogger(__name__)
13
14	1
15		ffmpeg = FFMPEG(os.environ.get('MUSIC_FFMPEG', 'ffmpeg'))
16	1
17	1
18		class AudioSegmenter(object):
19	1	def __init__(self, target_directory=tempfile.gettempdir()):
20		self._dir = target_directory
21
22	1	@property
23		def target_directory(self):
24	1	"""The directory path that will serve as the destination for storing created tracks"""
25		return self._dir
26	1
27		@target_directory.setter
28	1	def target_directory(self, directory_path):
29	1	self._dir = directory_path
30
31	1	def _trans(self, track_info):
32		return [os.path.join(self._dir, '{}.mp4'.format(track_info[0]))] + track_info[1:]
33
34		def segment(
35		self, album_file, data, supress_stdout=True, supress_stderr=True, sleep_seconds=0
36		):
37		"""
38
39		:param album_file:
40		:param data:
41	1	:param supress_stdout:
42	1	:param supress_stderr:
43	1	:param float sleep_seconds:
44	1	:return:
45	1	"""
46	1	exit_code = 0
47	1	i = 0
48		while exit_code == 0 and i < len(data) - 1:
49	1	time.sleep(sleep_seconds)
50	1	result = self._segment(
51		album_file,
52	1	*self._trans(list(data[i])),
53		supress_stdout=supress_stdout,
54	1	supress_stderr=supress_stderr,
55		)
56		i += 1
57		if result.exit_code != 0:
		0 ignored issues – show introduced 2023-03-19 21:08 UTC by Report Bug Copy Issue Report The variable `result` does not seem to be defined in case the `while` loop on line `48` is not entered. Are you sure this can never be the case? Loading history...
58		logger.error("Fmmpeg exit code: %s", result.exit_code)
59		logger.error("Ffmpeg st out: %s", result.stdout)
60		logger.error("Fmmpeg stderr: %s", result.stderr)
61		raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
62		result = self._segment(
63		album_file,
64		*self._trans(list(data[-1])),
65		supress_stdout=supress_stdout,
66		supress_stderr=supress_stderr,
67		)
68		if result.exit_code != 0:
69		logger.error("Fmmpeg exit code: %s", result.exit_code)
70		logger.error("Ffmpeg st out: %s", result.stdout)
71		logger.error("Fmmpeg stderr: %s", result.stderr)
72		raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
73		return [os.path.join(self._dir, '{}.mp4'.format(list(x)[0])) for x in data]
74
75		def segment_from_list(
76		self, album_file, data, supress_stdout=True, supress_stderr=True, sleep_seconds=0
77		):
78		"""
79		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
80		:param str album_file:
81		:param list data: list of lists. Each inner list must have 2 elements: track name and starting timestamp in hh:mm:ss
82		:param bool supress_stdout:
83	1	:param bool supress_stderr:
84		:param bool verbose:
85		:param float sleep_seconds:
86		:return: full paths to audio tracks
87		"""
88		# if not re.search('0:00', data[0][1]):
89		# raise NotStartingFromZeroTimestampError("First track ({}) is supposed to have a 0:00 timestamp. Instead {} found".format(data[0][0], data[0][1]))
90
91		exit_code = 0
92		data = StringParser().convert_tracks_data(data, album_file, target_directory=self._dir)
93		audio_file_paths = [x[0] for x in data]
94	1	i = 0
95	1	while exit_code == 0 and i < len(data) - 1:
96	1	time.sleep(sleep_seconds)
97		result = self._segment(
98	1	album_file,
99	1	*data[i],
100	1	supress_stdout=supress_stdout,
101	1	supress_stderr=supress_stderr,
102	1	)
103		if result.exit_code != 0:
104	1	logger.error("Fmmpeg exit code: %s", result.exit_code)
105	1	logger.error("Ffmpeg st out: %s", result.stdout)
106	1	logger.error("Fmmpeg stderr: %s", result.stderr)
107	1	raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
108	1	i += 1
109	1	# if exit_code != 0:
110	1	# raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
111	1	result = self._segment(
112		album_file, *data[-1], supress_stdout=supress_stdout, supress_stderr=supress_stderr
113		)
114		if result.exit_code != 0:
115	1	logger.error("Fmmpeg exit code: %s", result.exit_code)
116		logger.error("Ffmpeg st out: %s", result.stdout)
117		logger.error("Fmmpeg stderr: %s", result.stderr)
118		raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
119	1	# if exit_code != 0:
120		# raise FfmpegCommandError("Command '{}' failed".format(' '.join(self._args)))
121		return audio_file_paths
122
123		def segment_from_file(
124		self,
125	1	album_file,
126		tracks_file,
127		hhmmss_type,
128		supress_stdout=True,
129		supress_stderr=True,
130		sleep_seconds=0,
131		):
132		"""
133		Assumes the hhmmss are starting timestamps. 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
134		:param str album_file:
135		:param str tracks_file:
136		:param str hhmmss_type:
137		:param bool supress_stdout:
138		:param bool supress_stderr:
139		:param float sleep_seconds:
140		:return:
141		"""
142		with open(tracks_file, 'r') as f:
143		segmentation_info = SegmentationInformation.from_multiline(
144		f.read().strip(), hhmmss_type
145		)
146		return self.segment(
147		album_file,
148		segmentation_info,
149		supress_stdout=supress_stdout,
150		supress_stderr=supress_stderr,
151		sleep_seconds=sleep_seconds,
152		)
153
154		def _segment(self, args, *kwargs):
155		album_file = args[0]
156		track_file = args[1]
157		# supress_stdout = kwargs['supress_stdout']
158		# supress_stderr = kwargs['supress_stderr']
159
160		start = args[2] # starting timestamp
161		end: Optional[str] = None # end timestamp
162		# if it is the last segment then end timestamp is the end of the album
163		# and in that case the client code need to supply 3 arguments (not 4)
164		if 3 < len(args):
165		end = args[3]
166
167		# args = ['ffmpeg', '-y', '-i', '-acodec', 'copy', '-ss']
168		# self._args = args[:3] + ['{}'.format(album_file)] + args[3:] + [start] + (lambda: ['-to', str(end)] if end else [])() + ['{}'.format(track_file)]
169		self._args = (
170		'-y',
171		'-i',
172		str(album_file), # youtube downloaded audio steam (ie local mp4 file)
173		'-acodec',
174		'copy',
175		'-ss',
176		start,
177		*list((lambda: ['-to', str(end)] if end else [])()),
178		str(track_file),
179		)
180		logger.info("Segmenting: ffmpeg '{}'".format(' '.join(self._args)))
181		return ffmpeg(
182		*self._args,
183		)
184		# return subprocess.check_call(self._args, **self.__std_parameters(supress_stdout, supress_stderr))
185		# ro = subprocess.run(self._args, **self.__std_parameters(supress_stdout, supress_stderr))
186		# return ro.returncode
187
188		# @classmethod
189		# def __std_parameters(cls, std_out_flag, std_error_flag):
190		# """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
191		# If an input flag is False then the stream will be normally outputted; ie at the terminal"""
192		# return {v: subprocess.PIPE for k, v in zip([std_out_flag, std_error_flag], ['stdout', 'stderr']) if k}
193
194
195		class FfmpegCommandError(Exception):
196		pass
197
198
199		if __name__ == '__main__':
200		import sys
201
202		if len(sys.argv) < 3:
203		print('Usage: python3 Album_segmentation.py AUDIO_FILE TRACKS_FILE')
204		sys.exit(1)
205		try:
206		audio_segmenter = AudioSegmenter()
207		audio_segmenter.segment_from_file(
208		sys.argv[1],
209		sys.argv[2],
210		supress_stdout=True,
211		supress_stderr=True,
212		verbose=True,
213		sleep_seconds=0.45,
214		)
215		except Exception as e:
216		print(e)
217		sys.exit(1)
218

boromir674 / music-album-creator

Push — dev ( 3a28f6...0451e7 )

AudioSegmenter.segment() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like