'Refactored by Sourcery'

Author: Sourcery AI

ffmpeg_streaming/_clouds.py

@@ -78,7 +78,7 @@ def download(self, filename=None, **options):
         try:
             with filename as f:
                 self.s3.download_fileobj(bucket_name, key, f)
-            logging.info("The " + filename.name + " file was downloaded")
+            logging.info(f'The {filename.name} file was downloaded')
         except self.err as e:
             logging.error(e)
             raise RuntimeError(e)
@@ -174,7 +174,7 @@ def download(self, filename=None, **options):
 
         try:
             self.block_blob_service.get_blob_to_path(container, blob, filename)
-            logging.info("The " + filename + " file was downloaded")
+            logging.info(f'The {filename} file was downloaded')
         except:
             error = "An error occurred while downloading the file"
             logging.error(error)

ffmpeg_streaming/_command_builder.py

@@ -40,7 +40,7 @@ def _get_audio_bitrate(rep, index: int = None):
     @TODO: add documentation
     """
     if rep.bitrate.audio_ is not None and rep.bitrate.audio_ != 0:
-        opt = 'b:a' if index is None else 'b:a:' + str(index)
+        opt = 'b:a' if index is None else f'b:a:{index}'
         return {opt: rep.bitrate.audio}
 
     return {}
@@ -52,9 +52,10 @@ def _get_dash_stream(key, rep):
     """
     args = {
         'map': 0,
-        's:v:' + str(key): rep.size,
-        'b:v:' + str(key): rep.bitrate.calc_video()
+        f's:v:{str(key)}': rep.size,
+        f'b:v:{str(key)}': rep.bitrate.calc_video(),
     }
+
     args.update(_get_audio_bitrate(rep, key))
     args.update(rep.options)
     return cnv_options_to_args(args)

ffmpeg_streaming/_hls_helper.py

@@ -64,8 +64,8 @@ def update_suffix(self):
         @TODO: add documentation
         """
         unique = uuid.uuid4()
-        self.path = self.c_path + "-" + str(unique)
-        self.url = self.c_url + "-" + str(unique)
+        self.path = f'{self.c_path}-{str(unique)}'
+        self.url = f'{self.c_url}-{str(unique)}'
 
     def rotate_key(self, line: str):
         """
@@ -88,11 +88,12 @@ def sub_info(rep, sub_path) -> list:
     """
     tag = '#EXT-X-MEDIA:'
     info = [
-        f'TYPE=SUBTITLES',
-        f'GROUP-ID="subs"',
-        f'NAME="subtitles"',
-        f'URI="'+sub_path+'"'
+        'TYPE=SUBTITLES',
+        'GROUP-ID="subs"',
+        'NAME="subtitles"',
+        f'URI="{sub_path}' + '"',
     ]
+
     return [tag + ",".join(info)]
 
 def stream_info(rep, sub_exists) -> list:
@@ -106,7 +107,7 @@ def stream_info(rep, sub_exists) -> list:
         f'NAME="{rep.size.height}"'
     ]
     if sub_exists:
-        info.append(f'SUBTITLES="subs"')
+        info.append('SUBTITLES="subs"')
     custom = rep.options.pop('stream_info', [])
 
     return [tag + ",".join(info + custom)]
@@ -133,8 +134,9 @@ def _content(self) -> str:
         content = ['#EXTM3U'] + self._get_version() + self.media.options.get('description', [])
 
         for rep in self.media.reps:
-            sub_exists=os.path.isfile(os.path.dirname(self.media.output_)+'/'+self.sub_path(rep)[0])
-            if(sub_exists):
+            if sub_exists := os.path.isfile(
+                os.path.dirname(self.media.output_) + '/' + self.sub_path(rep)[0]
+            ):
                 content += sub_info(rep,self.sub_path(rep)[0]) + stream_info(rep,sub_exists) + self.stream_path(rep)
             else:
                 content += stream_info(rep,sub_exists) + self.stream_path(rep)
@@ -146,7 +148,7 @@ def _get_version(self) -> list:
         @TODO: add documentation
         """
         version = "7" if self.media.options.get('hls_segment_type', '') == 'fmp4' else "3"
-        return ['#EXT-X-VERSION:' + version]
+        return [f'#EXT-X-VERSION:{version}']
 
     def stream_path(self, rep):
         """

ffmpeg_streaming/_input.py

@@ -26,22 +26,17 @@ def __init__(self, video, options):
         self.video = video
 
     def _linux(self):
-        is_screen = self.options.pop('screen', False)
-        if is_screen:
-            cap = 'x11grab'
-        else:
-            cap = 'v4l2'
-
+        cap = 'x11grab' if (is_screen := self.options.pop('screen', False)) else 'v4l2'
         return {
             'f': cap,
             'i': self.video
         }
 
     def _windows(self):
-        self.video = 'video=' + str(self.video)
+        self.video = f'video={str(self.video)}'
         windows_audio = self.options.pop('windows_audio', None)
         if windows_audio is not None:
-            self.video = self.video + ':audio=' + str(windows_audio)
+            self.video = f'{self.video}:audio={str(windows_audio)}'
 
         return {
             'f': 'dshow',
@@ -59,7 +54,7 @@ def _unknown():
         raise OSError("Unreported OS!")
 
     def __iter__(self):
-        yield from getattr(self, '_' + get_os())().items()
+        yield from getattr(self, f'_{get_os()}')().items()
 
 
 def get_from_cloud(_cloud: Clouds, options: dict):
@@ -69,10 +64,7 @@ def get_from_cloud(_cloud: Clouds, options: dict):
     global cloud
     if cloud is None:
         save_to = options.pop('save_to', None)
-        cloud = {
-            'i': _cloud.download(save_to, **options),
-            'is_tmp': True if save_to is None else False
-        }
+        cloud = {'i': _cloud.download(save_to, **options), 'is_tmp': save_to is None}
 
     return cloud
 

ffmpeg_streaming/_media.py

@@ -115,9 +115,7 @@ def run(self, ffmpeg_bin, monitor: callable = None, **options):
         """
         @TODO: add documentation
         """
-        async_run = options.pop('async_run', True)
-
-        if async_run:
+        if async_run := options.pop('async_run', True):
             asyncio.run(self.async_run(ffmpeg_bin, monitor, **options))
         else:
             self._run(ffmpeg_bin, monitor, **options)
@@ -150,7 +148,12 @@ 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))
+        _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'):
@@ -215,7 +218,12 @@ 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))
+        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):
@@ -244,7 +252,7 @@ def __init__(self, _inputs):
         self.inputs = _inputs
 
         first_input = dict(copy.copy(_inputs.inputs[0]))
-        self.input = first_input.get('i', None)
+        self.input = first_input.get('i')
         self.input_temp = first_input.get('is_tmp', False)
 
     def hls(self, _format: Format, **hls_options):

ffmpeg_streaming/_reperesentation.py

@@ -30,10 +30,7 @@ def min_bitrate(bitrate: int) -> int:
     """
     @TODO: add documentation
     """
-    if bitrate < MINIMUM_BITRATE:
-        return MINIMUM_BITRATE
-
-    return bitrate
+    return max(bitrate, MINIMUM_BITRATE)
 
 
 def reduce_bitrate(bitrate: Bitrate, divide: int) -> Bitrate:
@@ -68,7 +65,7 @@ def __init__(self, original_size: Size, original_bitrate: Bitrate, _format: Form
         self._format = _format
         self.heights = heights if heights is not None else [2160, 1440, 1080, 720, 480, 360, 240, 144]
         self.bitrate = bitrate
-        self.is_default = True if heights is not None and bitrate is not None else False
+        self.is_default = heights is not None and bitrate is not None
 
         if heights is not None and bitrate is not None and len(heights) != len(bitrate):
             raise ValueError("The length of heights list must the same as length of bitrate")
@@ -90,14 +87,13 @@ def __next__(self):
         """
         @TODO: add documentation
         """
-        if self.index < len(self.heights):
-            height = self.heights[self.index]
-            width = self.original_size.ratio.calculate_width(height, self._format.multiply())
-            self.index += 1
-
-            return Representation(Size(width, height), cal_bitrate(self.bitrate, self.original_bitrate, self.index))
-        else:
+        if self.index >= len(self.heights):
             raise StopIteration
+        height = self.heights[self.index]
+        width = self.original_size.ratio.calculate_width(height, self._format.multiply())
+        self.index += 1
+
+        return Representation(Size(width, height), cal_bitrate(self.bitrate, self.original_bitrate, self.index))
 
 
 __all__ = [

ffmpeg_streaming/_utiles.py

@@ -36,7 +36,6 @@ def mkdir(dirname: str) -> None:
         os.makedirs(dirname)
     except OSError as exc:
         logging.info(exc)
-        pass
 
 
 def rm(path: str) -> None:
@@ -47,7 +46,6 @@ def rm(path: str) -> None:
         os.remove(path)
     except OSError as exc:
         logging.info(exc)
-        pass
 
 
 def clean_args(args: list) -> list:
@@ -105,16 +103,14 @@ def get_os():
     """
     @TODO: add documentation
     """
-    if platform == "linux" or platform == "linux2":
-        os_name = 'linux'
+    if platform in ["linux", "linux2"]:
+        return 'linux'
     elif platform == "darwin":
-        os_name = 'os_x'
-    elif platform == "win32" or platform == "Windows":
-        os_name = 'windows'
+        return 'os_x'
+    elif platform in ["win32", "Windows"]:
+        return 'windows'
     else:
-        os_name = 'unknown'
-
-    return os_name
+        return 'unknown'
 
 
 def cnv_options_to_args(options: dict):