[fix] ruff formatting?

Author: chyomin06

compressai_vision/codecs/encdec_utils/png_yuv.py

@@ -89,9 +89,9 @@ def __call__(self, input: Dict, file_prefix: str):
             images_in_folder = len(list(parent.glob(ext)))
             nb_frames = input["last_frame"] - input["frame_skip"]
 
-            assert images_in_folder == nb_frames, (
-                f"input folder contains {images_in_folder} images, {nb_frames} were expected"
-            )
+            assert (
+                images_in_folder == nb_frames
+            ), f"input folder contains {images_in_folder} images, {nb_frames} were expected"
 
             input_info = [
                 "-pattern_type",
@@ -122,9 +122,9 @@ def __call__(self, input: Dict, file_prefix: str):
 
         # Use existing YUV (if found and indicated for use):
         if self.use_yuv:
-            assert yuv_file is not None, (
-                "Parameter 'use_yuv' set True but YUV file not found."
-            )
+            assert (
+                yuv_file is not None
+            ), "Parameter 'use_yuv' set True but YUV file not found."
             size = yuv_file.stat().st_size
             bytes_per_luma_sample = {"yuv420p": 1.5}[chroma_format]
             bytes_per_sample = (input_bitdepth + 7) >> 3
@@ -135,9 +135,9 @@ def __call__(self, input: Dict, file_prefix: str):
                 * bytes_per_sample
                 * nb_frames
             )
-            assert size == expected_size, (
-                f"YUV found for input but expected size of {expected_size} bytes differs from actual size of {size} bytes"
-            )
+            assert (
+                size == expected_size
+            ), f"YUV found for input but expected size of {expected_size} bytes differs from actual size of {size} bytes"
             shutil.copy(yuv_file, yuv_in_path)
             print(f"Using pre-existing YUV file: {yuv_file}")
             return (yuv_in_path, nb_frames, frame_width, frame_height, file_prefix)
@@ -204,9 +204,9 @@ def __call__(
         frame_width = video_info["width"]
         frame_height = video_info["height"]
 
-        assert "420" in video_info["format"].value, (
-            f"Only support yuv420, but got {video_info['format']}"
-        )
+        assert (
+            "420" in video_info["format"].value
+        ), f"Only support yuv420, but got {video_info['format']}"
         pix_fmt_suffix = "10le" if video_info["bitdepth"] == 10 else ""
         chroma_format = "yuv420p"
 

compressai_vision/codecs/sic_sfu2022.py

@@ -151,13 +151,13 @@ def __init__(self, device: str, **kwargs):
         # root_url = "https://dspub.blob.core.windows.net/compressai/sic_sfu2022"
 
         self.target_tlayer = int(kwargs["target_task_layer"])
-        assert self.num_tasks == 2 or self.num_tasks == 3, (
-            f"SIC_SFU2023 supports only 2 or 3 task layers, but got {self.num_tasks}"
-        )
-        assert self.target_tlayer < self.num_tasks, (
-            f"target task layer must be lower than the number of tasks, \
+        assert (
+            self.num_tasks == 2 or self.num_tasks == 3
+        ), f"SIC_SFU2023 supports only 2 or 3 task layers, but got {self.num_tasks}"
+        assert (
+            self.target_tlayer < self.num_tasks
+        ), f"target task layer must be lower than the number of tasks, \
             but got {self.target_tlayer} < {self.num_tasks}"
-        )
 
         self.trg_vmodel = self.vmodels[self.target_tlayer]
 
@@ -572,9 +572,9 @@ def compress(self, x, target_layer=0, feature_only=False):
                 "models (the entropy coder is run sequentially on CPU)."
             )
 
-        assert target_layer < self.NUM_LAYERS, (
-            f"Got the target layer {target_layer}, but should be less than {self.NUM_LAYERS}"
-        )
+        assert (
+            target_layer < self.NUM_LAYERS
+        ), f"Got the target layer {target_layer}, but should be less than {self.NUM_LAYERS}"
 
         y = self.g_a(x)
         z = self.h_a(y)

compressai_vision/codecs/std_codecs.py

@@ -480,9 +480,9 @@ def encode(
                 for partial in list_of_bitstreams:
                     Path(partial).unlink()
 
-        assert Path(bitstream_path).is_file(), (
-            f"bitstream {bitstream_path} was not created"
-        )
+        assert Path(
+            bitstream_path
+        ).is_file(), f"bitstream {bitstream_path} was not created"
 
         if not remote_inference:
             inner_codec_bitstream = load_bitstream(bitstream_path)
@@ -601,12 +601,12 @@ def decode(
                 for file_path in sorted(Path(dec_path).glob(f"*{file_prefix}*.png")):
                     rec_frames.append(str(file_path))
 
-                assert file_prefix in rec_frames[0], (
-                    f"Can't find a correct filename with {file_prefix} in {dec_path}"
-                )
-                assert len(rec_frames) == 1, (
-                    f"Number of retrieved file must be 1, but got {len(rec_frames)}"
-                )
+                assert (
+                    file_prefix in rec_frames[0]
+                ), f"Can't find a correct filename with {file_prefix} in {dec_path}"
+                assert (
+                    len(rec_frames) == 1
+                ), f"Number of retrieved file must be 1, but got {len(rec_frames)}"
 
             conversion_time = 0
             output = {"file_names": rec_frames}

compressai_vision/codecs/utils.py

@@ -257,9 +257,9 @@ def reshape_frame_to_feature_pyramid(
         """reshape a frame of channels into the feature pyramid"""
 
         assert isinstance(x, (Tensor, Dict))
-        assert packing_all_in_one is True, (
-            "packing_all_in_one = False is not supported yet"
-        )
+        assert (
+            packing_all_in_one is True
+        ), "packing_all_in_one = False is not supported yet"
 
         top_y = 0
         tiled_frames = {}

compressai_vision/datasets/image.py

@@ -246,9 +246,9 @@ def __init__(self, root, dataset_name, imgs_folder, **kwargs):
         if kwargs["linear_mapper"] is True:
             mapper = LinearMapper()
         else:
-            assert kwargs["cfg"] is not None, (
-                "A proper mapper information via cfg must be provided"
-            )
+            assert (
+                kwargs["cfg"] is not None
+            ), "A proper mapper information via cfg must be provided"
             mapper = DatasetMapper(kwargs["cfg"], False)
 
         self.mapDataset = MapDataset(_dataset, mapper)

compressai_vision/datasets/utils.py

@@ -140,9 +140,9 @@ def __call__(self, dataset_dict):
         # Read image
         org_img = cv2.imread(dataset_dict["file_name"])  # return img in BGR by default
 
-        assert len(org_img.shape) == 3, (
-            f"detect an input image with 2 chs, {dataset_dict['file_name']}"
-        )
+        assert (
+            len(org_img.shape) == 3
+        ), f"detect an input image with 2 chs, {dataset_dict['file_name']}"
 
         img_h, img_w, _ = org_img.shape
 
@@ -221,9 +221,9 @@ def __call__(self, dataset_dict):
         # Read image
         org_img = cv2.imread(dataset_dict["file_name"])  # return img in BGR by default
 
-        assert len(org_img.shape) == 3, (
-            f"detect an input image with 2 chs, {dataset_dict['file_name']}"
-        )
+        assert (
+            len(org_img.shape) == 3
+        ), f"detect an input image with 2 chs, {dataset_dict['file_name']}"
 
         dataset_dict["height"], dataset_dict["width"], _ = org_img.shape
 

compressai_vision/evaluators/evaluators.py

@@ -725,9 +725,9 @@ def digest_summary(summary):
         return ret
 
     def mot_eval(self):
-        assert len(self.dataset) == len(self._predictions), (
-            "Total number of frames are mismatch"
-        )
+        assert len(self.dataset) == len(
+            self._predictions
+        ), "Total number of frames are mismatch"
 
         # skip the very first frame
         for gt_frame in self.dataset[1:]:
@@ -818,9 +818,9 @@ def __init__(
         assert self.seqinfo_path is not None, "Sequence Information must be provided"
 
     def mot_eval(self):
-        assert len(self.dataset) == len(self._predictions), (
-            "Total number of frames are mismatch"
-        )
+        assert len(self.dataset) == len(
+            self._predictions
+        ), "Total number of frames are mismatch"
 
         self._save_all_eval_info(self._predictions)
         _pd_pd = self._format_pd_in_motchallenge(self._predictions)
@@ -874,9 +874,9 @@ def __init__(
         )
 
     def mot_eval(self):
-        assert len(self.dataset) == len(self._predictions), (
-            "Total number of frames are mismatch"
-        )
+        assert len(self.dataset) == len(
+            self._predictions
+        ), "Total number of frames are mismatch"
 
         self._save_all_eval_info(self._predictions)
         _pd_pd = self._format_pd_in_motchallenge(self._predictions)

compressai_vision/model_wrappers/detectron2.py

@@ -387,9 +387,9 @@ def __init__(self, img_size: list):
         proposals, _ = self.proposal_generator(cdummy, x, None)
         results, _ = self.roi_heads(cdummy, x, proposals, None)
 
-        assert not torch.jit.is_scripting(), (
-            "Scripting is not supported for postprocess."
-        )
+        assert (
+            not torch.jit.is_scripting()
+        ), "Scripting is not supported for postprocess."
         return self.model._postprocess(
             results,
             [
@@ -423,9 +423,9 @@ def __init__(self, img_size: list):
         proposals, _ = self.proposal_generator(cdummy, x, None)
         results, _ = self.roi_heads(cdummy, x, proposals, None)
 
-        assert not torch.jit.is_scripting(), (
-            "Scripting is not supported for postprocess."
-        )
+        assert (
+            not torch.jit.is_scripting()
+        ), "Scripting is not supported for postprocess."
         return self.model._postprocess(
             results,
             [
@@ -461,9 +461,9 @@ def __init__(self, img_size: list):
         proposals, _ = self.proposal_generator(cdummy, fptensors, None)
         results, _ = self.roi_heads(cdummy, fptensors, proposals, None)
 
-        assert not torch.jit.is_scripting(), (
-            "Scripting is not supported for postprocess."
-        )
+        assert (
+            not torch.jit.is_scripting()
+        ), "Scripting is not supported for postprocess."
 
         return self.model._postprocess(
             results,
@@ -483,9 +483,9 @@ def deeper_features_for_accuracy_proxy(self, x: Dict):
         d = {}
         for e, ft in enumerate(x["data"].values()):
             nft = ft.contiguous().to(self.device)
-            assert nft.dim() == 3 or nft.dim() == 4, (
-                f"Input feature tensor dimension is supposed to be 3 or 4, but got {nft.dim()}"
-            )
+            assert (
+                nft.dim() == 3 or nft.dim() == 4
+            ), f"Input feature tensor dimension is supposed to be 3 or 4, but got {nft.dim()}"
             d[e] = nft.unsqueeze(0) if nft.dim() == 3 else nft
 
         class dummy:
@@ -580,9 +580,9 @@ def __init__(self, img_size: list):
         proposals, _ = self.proposal_generator(cdummy, x, None)
         results, _ = self.roi_heads(cdummy, x, proposals, None)
 
-        assert not torch.jit.is_scripting(), (
-            "Scripting is not supported for postprocess."
-        )
+        assert (
+            not torch.jit.is_scripting()
+        ), "Scripting is not supported for postprocess."
 
         processed_results = []
         for sem_seg_result, detector_result, input_per_image, image_size in zip(

compressai_vision/model_wrappers/jde.py

@@ -243,9 +243,9 @@ def _jde_process(self, pred, org_img_size: tuple, input_img_size: tuple):
         # removed tracks from the current frame
         current_removed_tracks = []
 
-        assert pred.size(1) == 54264, (
-            f"Default number of proposals by JDE must be 54264, but got {pred.size(1)}"
-        )
+        assert (
+            pred.size(1) == 54264
+        ), f"Default number of proposals by JDE must be 54264, but got {pred.size(1)}"
 
         selected_pred = pred[:, pred[0, :, 4] > self.model_configs["conf_thres"]]
         # only check the objects detected with confidence greater than .5

compressai_vision/pipelines/base.py

@@ -175,16 +175,14 @@ def _update_codec_configs_at_pipeline_level(self, total_num_frames):
         self._codec_skip_n_frames = self.configs["codec"]["skip_n_frames"]
         n_frames_to_be_encoded = self.configs["codec"]["n_frames_to_be_encoded"]
 
-        assert self._codec_skip_n_frames < total_num_frames, (
-            f"Number of skip frames {self._codec_skip_n_frames} must be less than total number of frames {total_num_frames}"
-        )
+        assert (
+            self._codec_skip_n_frames < total_num_frames
+        ), f"Number of skip frames {self._codec_skip_n_frames} must be less than total number of frames {total_num_frames}"
 
         if n_frames_to_be_encoded == -1:
             n_frames_to_be_encoded = total_num_frames
 
-        assert n_frames_to_be_encoded, (
-            f"Number of frames to be encoded must be greater than 0, but got {n_frames_to_be_encoded}"
-        )
+        assert n_frames_to_be_encoded, f"Number of frames to be encoded must be greater than 0, but got {n_frames_to_be_encoded}"
 
         if (self._codec_skip_n_frames + n_frames_to_be_encoded) > total_num_frames:
             self.logger.warning(
@@ -202,9 +200,11 @@ def _update_codec_configs_at_pipeline_level(self, total_num_frames):
             self._codec_skip_n_frames > 0
             or self._codec_n_frames_to_be_encoded != total_num_frames
         ):
-            assert self.configs["codec"]["encode_only"], (
-                "Encoding part of a sequence is only available when `codec.encode_only' is True"
-            )
+            assert self.configs[
+                "codec"
+            ][
+                "encode_only"
+            ], "Encoding part of a sequence is only available when `codec.encode_only' is True"
 
         self._codec_end_frame_idx = (
             self._codec_skip_n_frames + self._codec_n_frames_to_be_encoded
@@ -219,18 +219,18 @@ def _prep_features_to_dump(features, n_bits, datacatalog_name):
         if n_bits == -1:
             data_features = features["data"]
         elif n_bits >= 8:
-            assert n_bits == 8 or n_bits == 16, (
-                "currently it only supports dumping features in 8 bits or 16 bits"
-            )
-            assert datacatalog_name in list(MIN_MAX_DATASET.keys()), (
-                f"{datacatalog_name} does not exist in the pre-computed minimum and maximum tables"
-            )
+            assert (
+                n_bits == 8 or n_bits == 16
+            ), "currently it only supports dumping features in 8 bits or 16 bits"
+            assert (
+                datacatalog_name in list(MIN_MAX_DATASET.keys())
+            ), f"{datacatalog_name} does not exist in the pre-computed minimum and maximum tables"
             minv, maxv = MIN_MAX_DATASET[datacatalog_name]
             data_features = {}
             for key, data in features["data"].items():
-                assert data.min() >= minv and data.max() <= maxv, (
-                    f"{data.min()} should be greater than {minv} and {data.max()} should be less than {maxv}"
-                )
+                assert (
+                    data.min() >= minv and data.max() <= maxv
+                ), f"{data.min()} should be greater than {minv} and {data.max()} should be less than {maxv}"
                 out, _ = min_max_normalization(data, minv, maxv, bitdepth=n_bits)
 
                 if n_bits <= 8:
@@ -258,12 +258,12 @@ def _post_process_loaded_features(features, n_bits, datacatalog_name):
         if n_bits == -1:
             assert "data" in features
         elif n_bits >= 8:
-            assert n_bits == 8 or n_bits == 16, (
-                "currently it only supports dumping features in 8 bits or 16 bits"
-            )
-            assert datacatalog_name in list(MIN_MAX_DATASET.keys()), (
-                f"{datacatalog_name} does not exist in the pre-computed minimum and maximum tables"
-            )
+            assert (
+                n_bits == 8 or n_bits == 16
+            ), "currently it only supports dumping features in 8 bits or 16 bits"
+            assert (
+                datacatalog_name in list(MIN_MAX_DATASET.keys())
+            ), f"{datacatalog_name} does not exist in the pre-computed minimum and maximum tables"
             minv, maxv = MIN_MAX_DATASET[datacatalog_name]
             data_features = {}
             for key, data in features["data"].items():