ffmpeg_streaming._media.Media.stream2file() - Code Metrics - Inspection of "Sourcery refactored master branch" - aminyazdanpanah/python-ffmpeg-video-streaming - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#74)

unknown

created 2022-02-10 12:33 UTC

ffmpeg_streaming._media.Media.stream2file() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	5
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	3
dl	0
loc	5
rs	10
c	0
b	0
f	0

"""
ffmpeg_streaming.media
~~~~~~~~~~~~

Media object to build a stream objects


:copyright: (c) 2020 by Amin Yazdanpanah.
:website: https://www.aminyazdanpanah.com
:email: [email protected]
:license: MIT, see LICENSE for more details.
"""
import abc
import os
import shutil
import tempfile
import atexit
import asyncio
import copy

from ._clouds import CloudManager
from ._command_builder import command_builder
from ._format import Format
from ._hls_helper import HLSKeyInfoFile, HLSMasterPlaylist
from ._process import Process
from ._reperesentation import Representation, AutoRep
from ._utiles import mkdir, rm
from .ffprobe import FFProbe


class Save(abc.ABC):
    def __init__(self, media, _format: Format, **options):
        """
        @TODO: add documentation
        """
        atexit.register(self.finish_up)

        self.output_ = ''
        self.key = None
        self.media = media
        self.format = _format
        self.options = options
        self.pipe = None
        self.output_temp = False

    def finish_up(self):
        """
        @TODO: add documentation
        """
        if self.media.input_temp:
            rm(self.media.input)
        if self.output_temp:
            self.clouds.transfer('upload_directory', os.path.dirname(self.output_))
            if self.output:
                shutil.move(os.path.dirname(self.output_), os.path.dirname(str(self.output)))
            else:
                shutil.rmtree(os.path.dirname(str(self.output_)), ignore_errors=True)

    @abc.abstractmethod
    def set_up(self):
        pass

    def __getattr__(self, name):
        def method(*args, **kwargs):
            if name in ['save', 'package']:
                self.output(*args, **kwargs)
            else:
                raise AttributeError("The object has no attribute {}".format(name))

        return method

    def output(self, output: str = None, clouds: CloudManager = None,
               run_command: bool = True, ffmpeg_bin: str = 'ffmpeg', monitor: callable = None, **options):
        """
        @TODO: add documentation
        """
        if output is None and clouds is None:
            self.output_ = self.media.input
        elif clouds is not None:
            self.output_temp = True
            setattr(self, 'clouds', clouds)
            setattr(self, 'output', output)
            if clouds.filename is None:
                clouds.filename = os.path.basename(output if output is not None else self.media.input)
            self.output_ = os.path.join(tempfile.mkdtemp(prefix='ffmpeg_streaming_'), clouds.filename)
        else:
            mkdir(os.path.dirname(output))
            self.output_ = output

        self.set_up()

        if run_command:
            self.run(ffmpeg_bin, monitor, **options)

    def probe(self, ffprobe_bin='ffprobe') -> FFProbe:
        """
        @TODO: add documentation
        """
        return FFProbe(self.media.input, ffprobe_bin)

    def _run(self, ffmpeg_bin, monitor: callable = None, **options):
        """
        @TODO: add documentation
        """
        with Process(self, command_builder(ffmpeg_bin, self), monitor, **options) as process:
            self.pipe, err = process.run()

    async def async_run(self, ffmpeg_bin, monitor: callable = None, **options):
        """
        @TODO: add documentation
        """
        self._run(ffmpeg_bin, monitor, **options)

    def run(self, ffmpeg_bin, monitor: callable = None, **options):
        """
        @TODO: add documentation
        """
        if async_run := options.pop('async_run', True):
            asyncio.run(self.async_run(ffmpeg_bin, monitor, **options))
        else:
            self._run(ffmpeg_bin, monitor, **options)


class Streaming(Save, abc.ABC):
    def __init__(self, media, _format: Format, **options):
        """
        @TODO: add documentation
        """
        self.reps = list
        super(Streaming, self).__init__(media, _format, **options)

    def representations(self, *reps: Representation):
        self.reps = list(reps)

    def auto_generate_representations(self, heights=None, bitrate=None, ffprobe_bin='ffprobe', include_original=True,
                                      ascending_sort=False):
        """
        @TODO: add documentation
        """
        probe = self.probe(ffprobe_bin)
        self.reps = AutoRep(probe.video_size, probe.bitrate, self.format, heights, bitrate, include_original)

        if ascending_sort:
            self.reps = sorted(self.reps, key=lambda rep: rep.bitrate.overall_)

    def add_filter(self, *_filter: str):
        """
        @TODO: add documentation
        """
        _filters = self.options.pop('filter_complex', None)
        _filters = (
            f'{_filters},' + ",".join(list(_filter))
            if _filters is not None
            else ",".join(list(_filter))
        )

        self.options.update({'filter_complex': _filters})

    def watermarking(self, path, _filter='overlay=10:10'):
        """
        @TODO: add documentation
        """
        self.media.inputs.input(path)
        self.add_filter(_filter)


class DASH(Streaming):
    def set_up(self):
        pass

    def generate_hls_playlist(self):
        self.options.update({'hls_playlist': 1})


class HLS(Streaming):
    KEY_INFO_FILE_PATH = None
    PERIODIC_RE_KEY_FLAG = 'periodic_rekey'
    MASTER_PLAYLIST_IS_SAVED = False

    def set_up(self):
        """
        @TODO: add documentation
        """
        if not HLS.MASTER_PLAYLIST_IS_SAVED:
            self.save_master_playlist()

    def encryption(self, path: str, url: str, key_rotation_period: int = 0, needle: str = ".ts' for writing", length: int = 16):
        """
        @TODO: add documentation
        """
        with tempfile.NamedTemporaryFile(mode='w', suffix='_py_ff_vi_st.tmp', delete=False) as key_info:
            HLS.KEY_INFO_FILE_PATH = key_info.name

        key_info_file = HLSKeyInfoFile(HLS.KEY_INFO_FILE_PATH, path, url, key_rotation_period, needle, length)
        self.options.update({'hls_key_info_file': str(key_info_file)})

        if key_rotation_period > 0:
            setattr(self, 'key_rotation', key_info_file)
            self.flags(HLS.PERIODIC_RE_KEY_FLAG)

    def fragmented_mp4(self):
        """
        @TODO: add documentation
        """
        self.options.update({'hls_segment_type': 'fmp4'})

    def save_master_playlist(self, path=None):
        """
        @TODO: add documentation
        """
        if path is not None:
            HLS.MASTER_PLAYLIST_IS_SAVED = True

        HLSMasterPlaylist.generate(self, path)

    def flags(self, *flags: str):
        """
        @TODO: add documentation
        """
        hls_flags = self.options.pop('hls_flags', None)
        hls_flags = (
            f'{hls_flags}+' + "+".join(list(flags))
            if hls_flags is not None
            else "+".join(list(flags))
        )

        self.options.update({'hls_flags': hls_flags})

    def finish_up(self):
        """
        @TODO: add documentation
        """
        if HLS.KEY_INFO_FILE_PATH is not None:
            rm(HLS.KEY_INFO_FILE_PATH)

        super(HLS, self).finish_up()


class Stream2File(Save):
    """
    @TODO: add documentation
    """
    def set_up(self):
        pass


class Media(object):
    def __init__(self, _inputs):
        """
        @TODO: add documentation
        """
        self.inputs = _inputs

        first_input = dict(copy.deepcopy(_inputs.inputs[0]))
        self.input = first_input.get('i')
        self.input_temp = first_input.get('is_tmp', False)

    def hls(self, _format: Format, **hls_options):
        """
        @TODO: add documentation
        """
        return HLS(self, _format, **hls_options)

    def dash(self, _format: Format, **dash_options):
        """
        @TODO: add documentation
        """
        return DASH(self, _format, **dash_options)

    def stream2file(self, _format: Format, **options):
        """
        @TODO: add documentation
        """
        return Stream2File(self, _format, **options)


1			"""
2			ffmpeg_streaming.media
3			~~~~~~~~~~~~
4
5			Media object to build a stream objects
6
7
8			:copyright: (c) 2020 by Amin Yazdanpanah.
9			:website: https://www.aminyazdanpanah.com
10			:email: [email protected]
11			:license: MIT, see LICENSE for more details.
12			"""
13			import abc
14			import os
15			import shutil
16			import tempfile
17			import atexit
18			import asyncio
19			import copy
20
21			from ._clouds import CloudManager
22			from ._command_builder import command_builder
23			from ._format import Format
24			from ._hls_helper import HLSKeyInfoFile, HLSMasterPlaylist
25			from ._process import Process
26			from ._reperesentation import Representation, AutoRep
27			from ._utiles import mkdir, rm
28			from .ffprobe import FFProbe
29
30
31			class Save(abc.ABC):
32			def __init__(self, media, _format: Format, **options):
33			"""
34			@TODO: add documentation
35			"""
36			atexit.register(self.finish_up)
37
38			self.output_ = ''
39			self.key = None
40			self.media = media
41			self.format = _format
42			self.options = options
43			self.pipe = None
44			self.output_temp = False
45
46			def finish_up(self):
47			"""
48			@TODO: add documentation
49			"""
50			if self.media.input_temp:
51			rm(self.media.input)
52			if self.output_temp:
53			self.clouds.transfer('upload_directory', os.path.dirname(self.output_))
54			if self.output:
55			shutil.move(os.path.dirname(self.output_), os.path.dirname(str(self.output)))
56			else:
57			shutil.rmtree(os.path.dirname(str(self.output_)), ignore_errors=True)
58
59			@abc.abstractmethod
60			def set_up(self):
61			pass
62
63			def __getattr__(self, name):
64			def method(args, *kwargs):
65			if name in ['save', 'package']:
66			self.output(args, *kwargs)
67			else:
68			raise AttributeError("The object has no attribute {}".format(name))
69
70			return method
71
72			def output(self, output: str = None, clouds: CloudManager = None,
73			run_command: bool = True, ffmpeg_bin: str = 'ffmpeg', monitor: callable = None, **options):
74			"""
75			@TODO: add documentation
76			"""
77			if output is None and clouds is None:
78			self.output_ = self.media.input
79			elif clouds is not None:
80			self.output_temp = True
81			setattr(self, 'clouds', clouds)
82			setattr(self, 'output', output)
83			if clouds.filename is None:
84			clouds.filename = os.path.basename(output if output is not None else self.media.input)
85			self.output_ = os.path.join(tempfile.mkdtemp(prefix='ffmpeg_streaming_'), clouds.filename)
86			else:
87			mkdir(os.path.dirname(output))
88			self.output_ = output
89
90			self.set_up()
91
92			if run_command:
93			self.run(ffmpeg_bin, monitor, **options)
94
95			def probe(self, ffprobe_bin='ffprobe') -> FFProbe:
96			"""
97			@TODO: add documentation
98			"""
99			return FFProbe(self.media.input, ffprobe_bin)
100
101			def _run(self, ffmpeg_bin, monitor: callable = None, **options):
102			"""
103			@TODO: add documentation
104			"""
105			with Process(self, command_builder(ffmpeg_bin, self), monitor, **options) as process:
106			self.pipe, err = process.run()
107
108			async def async_run(self, ffmpeg_bin, monitor: callable = None, **options):
109			"""
110			@TODO: add documentation
111			"""
112			self._run(ffmpeg_bin, monitor, **options)
113
114			def run(self, ffmpeg_bin, monitor: callable = None, **options):
115			"""
116			@TODO: add documentation
117			"""
118			if async_run := options.pop('async_run', True):
119			asyncio.run(self.async_run(ffmpeg_bin, monitor, **options))
120			else:
121			self._run(ffmpeg_bin, monitor, **options)
122
123
124			class Streaming(Save, abc.ABC):
125			def __init__(self, media, _format: Format, **options):
126			"""
127			@TODO: add documentation
128			"""
129			self.reps = list
130			super(Streaming, self).__init__(media, _format, **options)
131
132			def representations(self, *reps: Representation):
133			self.reps = list(reps)
134
135			def auto_generate_representations(self, heights=None, bitrate=None, ffprobe_bin='ffprobe', include_original=True,
136			ascending_sort=False):
137			"""
138			@TODO: add documentation
139			"""
140			probe = self.probe(ffprobe_bin)
141			self.reps = AutoRep(probe.video_size, probe.bitrate, self.format, heights, bitrate, include_original)
142
143			if ascending_sort:
144			self.reps = sorted(self.reps, key=lambda rep: rep.bitrate.overall_)
145
146			def add_filter(self, *_filter: str):
147			"""
148			@TODO: add documentation
149			"""
150			_filters = self.options.pop('filter_complex', None)
151			_filters = (
152			f'{_filters},' + ",".join(list(_filter))
153			if _filters is not None
154			else ",".join(list(_filter))
155			)
156
157			self.options.update({'filter_complex': _filters})
158
159			def watermarking(self, path, _filter='overlay=10:10'):
160			"""
161			@TODO: add documentation
162			"""
163			self.media.inputs.input(path)
164			self.add_filter(_filter)
165
166
167			class DASH(Streaming):
168			def set_up(self):
169			pass
170
171			def generate_hls_playlist(self):
172			self.options.update({'hls_playlist': 1})
173
174
175			class HLS(Streaming):
176			KEY_INFO_FILE_PATH = None
177			PERIODIC_RE_KEY_FLAG = 'periodic_rekey'
178			MASTER_PLAYLIST_IS_SAVED = False
179
180			def set_up(self):
181			"""
182			@TODO: add documentation
183			"""
184			if not HLS.MASTER_PLAYLIST_IS_SAVED:
185			self.save_master_playlist()
186
187			def encryption(self, path: str, url: str, key_rotation_period: int = 0, needle: str = ".ts' for writing", length: int = 16):
188			"""
189			@TODO: add documentation
190			"""
191			with tempfile.NamedTemporaryFile(mode='w', suffix='_py_ff_vi_st.tmp', delete=False) as key_info:
192			HLS.KEY_INFO_FILE_PATH = key_info.name
193
194			key_info_file = HLSKeyInfoFile(HLS.KEY_INFO_FILE_PATH, path, url, key_rotation_period, needle, length)
195			self.options.update({'hls_key_info_file': str(key_info_file)})
196
197			if key_rotation_period > 0:
198			setattr(self, 'key_rotation', key_info_file)
199			self.flags(HLS.PERIODIC_RE_KEY_FLAG)
200
201			def fragmented_mp4(self):
202			"""
203			@TODO: add documentation
204			"""
205			self.options.update({'hls_segment_type': 'fmp4'})
206
207			def save_master_playlist(self, path=None):
208			"""
209			@TODO: add documentation
210			"""
211			if path is not None:
212			HLS.MASTER_PLAYLIST_IS_SAVED = True
213
214			HLSMasterPlaylist.generate(self, path)
215
216			def flags(self, *flags: str):
217			"""
218			@TODO: add documentation
219			"""
220			hls_flags = self.options.pop('hls_flags', None)
221			hls_flags = (
222			f'{hls_flags}+' + "+".join(list(flags))
223			if hls_flags is not None
224			else "+".join(list(flags))
225			)
226
227			self.options.update({'hls_flags': hls_flags})
228
229			def finish_up(self):
230			"""
231			@TODO: add documentation
232			"""
233			if HLS.KEY_INFO_FILE_PATH is not None:
234			rm(HLS.KEY_INFO_FILE_PATH)
235
236			super(HLS, self).finish_up()
237
238
239			class Stream2File(Save):
240			"""
241			@TODO: add documentation
242			"""
243			def set_up(self):
244			pass
245
246
247			class Media(object):
248			def __init__(self, _inputs):
249			"""
250			@TODO: add documentation
251			"""
252			self.inputs = _inputs
253
254			first_input = dict(copy.deepcopy(_inputs.inputs[0]))
255			self.input = first_input.get('i')
256			self.input_temp = first_input.get('is_tmp', False)
257
258			def hls(self, _format: Format, **hls_options):
259			"""
260			@TODO: add documentation
261			"""
262			return HLS(self, _format, **hls_options)
263
264			def dash(self, _format: Format, **dash_options):
265			"""
266			@TODO: add documentation
267			"""
268			return DASH(self, _format, **dash_options)
269
270			def stream2file(self, _format: Format, **options):
271			"""
272			@TODO: add documentation
273			"""
274			return Stream2File(self, _format, **options)
275

aminyazdanpanah / python-ffmpeg-video-streaming

Pull Request — master (#74)

ffmpeg_streaming._media.Media.stream2file() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like