ffmpeg_streaming._media.Save.async_run() - Code Metrics - Inspection of "Add time left to monitor callback method" - aminyazdanpanah/python-ffmpeg-video-streaming - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( db420a...23049e )

by Amin

created 2020-07-10 12:45 UTC

ffmpeg_streaming._media.Save.async_run() A

↳ Parent: ffmpeg_streaming._media

Complexity

Conditions

Size

Total Lines	5
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	4
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
        """
        async_run = options.pop('async_run', True)

        if async_run:
            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 = _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 = 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', None)
        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			async_run = options.pop('async_run', True)
119
120			if async_run:
121			asyncio.run(self.async_run(ffmpeg_bin, monitor, **options))
122			else:
123			self._run(ffmpeg_bin, monitor, **options)
124
125
126			class Streaming(Save, abc.ABC):
127			def __init__(self, media, _format: Format, **options):
128			"""
129			@TODO: add documentation
130			"""
131			self.reps = list
132			super(Streaming, self).__init__(media, _format, **options)
133
134			def representations(self, *reps: Representation):
135			self.reps = list(reps)
136
137			def auto_generate_representations(self, heights=None, bitrate=None, ffprobe_bin='ffprobe', include_original=True,
138			ascending_sort=False):
139			"""
140			@TODO: add documentation
141			"""
142			probe = self.probe(ffprobe_bin)
143			self.reps = AutoRep(probe.video_size, probe.bitrate, self.format, heights, bitrate, include_original)
144
145			if ascending_sort:
146			self.reps = sorted(self.reps, key=lambda rep: rep.bitrate.overall_)
147
148			def add_filter(self, *_filter: str):
149			"""
150			@TODO: add documentation
151			"""
152			_filters = self.options.pop('filter_complex', None)
153			_filters = _filters + "," + ",".join(list(_filter)) if _filters is not None else ",".join(list(_filter))
154			self.options.update({'filter_complex': _filters})
155
156			def watermarking(self, path, _filter='overlay=10:10'):
157			"""
158			@TODO: add documentation
159			"""
160			self.media.inputs.input(path)
161			self.add_filter(_filter)
162
163
164			class DASH(Streaming):
165			def set_up(self):
166			pass
167
168			def generate_hls_playlist(self):
169			self.options.update({'hls_playlist': 1})
170
171
172			class HLS(Streaming):
173			KEY_INFO_FILE_PATH = None
174			PERIODIC_RE_KEY_FLAG = 'periodic_rekey'
175			MASTER_PLAYLIST_IS_SAVED = False
176
177			def set_up(self):
178			"""
179			@TODO: add documentation
180			"""
181			if not HLS.MASTER_PLAYLIST_IS_SAVED:
182			self.save_master_playlist()
183
184			def encryption(self, path: str, url: str, key_rotation_period: int = 0, needle: str = ".ts' for writing", length: int = 16):
185			"""
186			@TODO: add documentation
187			"""
188			with tempfile.NamedTemporaryFile(mode='w', suffix='_py_ff_vi_st.tmp', delete=False) as key_info:
189			HLS.KEY_INFO_FILE_PATH = key_info.name
190
191			key_info_file = HLSKeyInfoFile(HLS.KEY_INFO_FILE_PATH, path, url, key_rotation_period, needle, length)
192			self.options.update({'hls_key_info_file': str(key_info_file)})
193
194			if key_rotation_period > 0:
195			setattr(self, 'key_rotation', key_info_file)
196			self.flags(HLS.PERIODIC_RE_KEY_FLAG)
197
198			def fragmented_mp4(self):
199			"""
200			@TODO: add documentation
201			"""
202			self.options.update({'hls_segment_type': 'fmp4'})
203
204			def save_master_playlist(self, path=None):
205			"""
206			@TODO: add documentation
207			"""
208			if path is not None:
209			HLS.MASTER_PLAYLIST_IS_SAVED = True
210
211			HLSMasterPlaylist.generate(self, path)
212
213			def flags(self, *flags: str):
214			"""
215			@TODO: add documentation
216			"""
217			hls_flags = self.options.pop('hls_flags', None)
218			hls_flags = hls_flags + "+" + "+".join(list(flags)) if hls_flags is not None else "+".join(list(flags))
219			self.options.update({'hls_flags': hls_flags})
220
221			def finish_up(self):
222			"""
223			@TODO: add documentation
224			"""
225			if HLS.KEY_INFO_FILE_PATH is not None:
226			rm(HLS.KEY_INFO_FILE_PATH)
227
228			super(HLS, self).finish_up()
229
230
231			class Stream2File(Save):
232			"""
233			@TODO: add documentation
234			"""
235			def set_up(self):
236			pass
237
238
239			class Media(object):
240			def __init__(self, _inputs):
241			"""
242			@TODO: add documentation
243			"""
244			self.inputs = _inputs
245
246			first_input = dict(copy.deepcopy(_inputs.inputs[0]))
247			self.input = first_input.get('i', None)
248			self.input_temp = first_input.get('is_tmp', False)
249
250			def hls(self, _format: Format, **hls_options):
251			"""
252			@TODO: add documentation
253			"""
254			return HLS(self, _format, **hls_options)
255
256			def dash(self, _format: Format, **dash_options):
257			"""
258			@TODO: add documentation
259			"""
260			return DASH(self, _format, **dash_options)
261
262			def stream2file(self, _format: Format, **options):
263			"""
264			@TODO: add documentation
265			"""
266			return Stream2File(self, _format, **options)
267

aminyazdanpanah / python-ffmpeg-video-streaming

Push — master ( db420a...23049e )

ffmpeg_streaming._media.Save.async_run() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like