Black

keras_fsl/dataframe/operators/__init__.py

@@ -4,7 +4,7 @@
 
 
 __all__ = [
-    'NaiveMaxProba',
-    'RandomAssignment',
-    'ToKShotDataset',
+    "NaiveMaxProba",
+    "RandomAssignment",
+    "ToKShotDataset",
 ]

keras_fsl/dataframe/operators/to_k_shot_dataset.py

@@ -9,7 +9,7 @@ class ToKShotDataset(AbstractOperator):
     Create tf.data.Dataset with random groups of k_shot consecutive images with the same label
     """
 
-    def __init__(self, k_shot, preprocessing, label_column='label_one_hot'):
+    def __init__(self, k_shot, preprocessing, label_column="label_one_hot"):
         """
 
         Args:
@@ -29,16 +29,12 @@ def load_img(annotation):
         Returns:
             dict: the input dict with an extra image key.
         """
-        return (
-            {
-                'image': tf.io.decode_and_crop_jpeg(
-                    tf.io.read_file(annotation['image_name']),
-                    crop_window=annotation['crop_window'],
-                    channels=3,
-                ),
-                **annotation,
-            }
-        )
+        return {
+            "image": tf.io.decode_and_crop_jpeg(
+                tf.io.read_file(annotation["image_name"]), crop_window=annotation["crop_window"], channels=3,
+            ),
+            **annotation,
+        }
 
     def repeat_k_shot(self, index):
         return tf.data.Dataset.from_tensors(index).repeat(self.k_shot)
@@ -48,33 +44,32 @@ def to_dataset(self, group):
         Transform a pd.DataFrame into a tf.data.Dataset and load images
         """
         return (
-            tf.data.Dataset.from_tensor_slices(group.to_dict('list'))
+            tf.data.Dataset.from_tensor_slices(group.to_dict("list"))
             .map(self.load_img, num_parallel_calls=tf.data.experimental.AUTOTUNE)
             .cache()
             .shuffle(buffer_size=len(group), reshuffle_each_iteration=True)
             .repeat()
         )
 
     def __call__(self, input_dataframe):
-        return (
-            tf.data.experimental.choose_from_datasets(
-                datasets=(
-                    input_dataframe
-                    .assign(
-                        label_one_hot=lambda df: pd.get_dummies(df.label).values.tolist(),
-                        crop_window=lambda df: df[["crop_y", "crop_x", "crop_height", "crop_width"]].values.tolist(),
-                    )
-                    .groupby('label')
-                    .apply(self.to_dataset)
-                ),
-                choice_dataset=(
-                    tf.data.Dataset.range(len(input_dataframe.label.unique()))
-                    .shuffle(buffer_size=len(input_dataframe.label.unique()), reshuffle_each_iteration=True)
-                    .flat_map(self.repeat_k_shot)
-                ),
-            )
-            .map(
-                lambda annotation: (self.preprocessing(annotation['image']), tf.cast(annotation[self.label_column], tf.float32)),
-                num_parallel_calls=tf.data.experimental.AUTOTUNE,
-            )
+        return tf.data.experimental.choose_from_datasets(
+            datasets=(
+                input_dataframe.assign(
+                    label_one_hot=lambda df: pd.get_dummies(df.label).values.tolist(),
+                    crop_window=lambda df: df[["crop_y", "crop_x", "crop_height", "crop_width"]].values.tolist(),
+                )
+                .groupby("label")
+                .apply(self.to_dataset)
+            ),
+            choice_dataset=(
+                tf.data.Dataset.range(len(input_dataframe.label.unique()))
+                .shuffle(buffer_size=len(input_dataframe.label.unique()), reshuffle_each_iteration=True)
+                .flat_map(self.repeat_k_shot)
+            ),
+        ).map(
+            lambda annotation: (
+                self.preprocessing(annotation["image"]),
+                tf.cast(annotation[self.label_column], tf.float32),
+            ),
+            num_parallel_calls=tf.data.experimental.AUTOTUNE,
         )

keras_fsl/datasets/omniglot.py

@@ -4,35 +4,31 @@
 import pandas as pd
 from tensorflow.keras.utils.data_utils import get_file
 
-BASE_PATH = 'https://raw.githubusercontent.com/brendenlake/omniglot/master/python'
+BASE_PATH = "https://raw.githubusercontent.com/brendenlake/omniglot/master/python"
 
 
 def load_dataframe(dataset_name):
     dataset_path = get_file(
-        f'{dataset_name}.zip',
-        origin=f'{BASE_PATH}/{dataset_name}.zip',
+        f"{dataset_name}.zip",
+        origin=f"{BASE_PATH}/{dataset_name}.zip",
         extract=True,
-        cache_subdir=Path('datasets') / 'omniglot'
+        cache_subdir=Path("datasets") / "omniglot",
     )
     dataset_dir = os.path.splitext(dataset_path)[0]
-    dataset = pd.DataFrame(columns=['image_name', 'alphabet', 'label'])
+    dataset = pd.DataFrame(columns=["image_name", "alphabet", "label"])
     for root, _, files in os.walk(dataset_dir):
         if files:
             alphabet, label = Path(root).relative_to(dataset_dir).parts
             root = Path(root)
             image_names = [root / file for file in files]
-            dataset = (
-                dataset
-                .append(
-                    pd.DataFrame({'image_name': image_names, 'alphabet': alphabet, 'label': label}),
-                    ignore_index=True,
-                )
+            dataset = dataset.append(
+                pd.DataFrame({"image_name": image_names, "alphabet": alphabet, "label": label}), ignore_index=True,
             )
 
     return dataset
 
 
 def load_data():
-    train_set = load_dataframe('images_background')
-    test_set = load_dataframe('images_evaluation')
+    train_set = load_dataframe("images_background")
+    test_set = load_dataframe("images_evaluation")
     return train_set, test_set

keras_fsl/losses/yolo_loss.py

@@ -24,12 +24,13 @@ def _yolo_loss(y_true, y_pred):
 
         for image, pred in zip(y_true, y_pred):
             loss_objectness.assign_add(
-                tf.math.reduce_sum(tf.keras.backend.binary_crossentropy(tf.zeros_like(y_pred[..., 4]), y_pred[..., 4])))
+                tf.math.reduce_sum(tf.keras.backend.binary_crossentropy(tf.zeros_like(y_pred[..., 4]), y_pred[..., 4]))
+            )
             for box in image:
                 if box[4] < 1:
                     continue
-                height_width_min = tf.minimum(box[2:4], anchors[['height', 'width']].values)
-                height_width_max = tf.maximum(box[2:4], anchors[['height', 'width']].values)
+                height_width_min = tf.minimum(box[2:4], anchors[["height", "width"]].values)
+                height_width_max = tf.maximum(box[2:4], anchors[["height", "width"]].values)
                 intersection = tf.reduce_prod(height_width_min, axis=-1)
                 union = tf.reduce_prod(height_width_max, axis=-1)
                 iou = intersection / union
@@ -46,7 +47,9 @@ def _yolo_loss(y_true, y_pred):
                         loss_objectness.assign_add(tf.keras.backend.binary_crossentropy(box[4], selected_pred[4]))
                         loss_coordinates.assign_add(tf.norm(box[:2] - selected_pred[:2], ord=2))
                         loss_box.assign_add(tf.norm(box[2:4] - selected_pred[2:4], ord=2))
-                        loss_classes.assign_add(tf.reduce_sum(tf.keras.backend.binary_crossentropy(box[5:], selected_pred[5:-1])))
+                        loss_classes.assign_add(
+                            tf.reduce_sum(tf.keras.backend.binary_crossentropy(box[5:], selected_pred[5:-1]))
+                        )
 
         return loss_coordinates + loss_box + loss_objectness + loss_classes
 

keras_fsl/models/activations/__init__.py

@@ -2,6 +2,6 @@
 from .yolo_coordinates import YoloCoordinates
 
 __all__ = [
-    'YoloBox',
-    'YoloCoordinates',
+    "YoloBox",
+    "YoloCoordinates",
 ]

keras_fsl/models/activations/yolo_box.py

@@ -15,9 +15,13 @@ def YoloBox(anchor):
         anchor (Union[pandas.Series, collections.namedtuple]): with key width and height. Note that given a tensor with shape
             (batch_size, i, j, channels), i is related to height and j to width
   """
-    return Sequential([
-        Activation('exponential'),
-        Lambda(lambda input_, anchor_=anchor: (
-            input_ * tf.convert_to_tensor([anchor_.height, anchor_.width], dtype=tf.float32)
-        )),
-    ])
+    return Sequential(
+        [
+            Activation("exponential"),
+            Lambda(
+                lambda input_, anchor_=anchor: (
+                    input_ * tf.convert_to_tensor([anchor_.height, anchor_.width], dtype=tf.float32)
+                )
+            ),
+        ]
+    )

keras_fsl/models/activations/yolo_coordinates.py

@@ -27,8 +27,13 @@ def YoloCoordinates():
     Activation function for the box center coordinates regression. Coordinates are relative to the image dimension, ie. between 0
         and 1
     """
-    return Sequential([
-        Activation('sigmoid'),
-        Lambda(lambda input_: input_ + tf.cast(tf.expand_dims(build_grid_coordinates(tf.shape(input_)[1:3]), 0), input_.dtype)),
-        Lambda(lambda input_: input_ / tf.cast(tf.shape(input_)[1:3], input_.dtype)),
-    ])
+    return Sequential(
+        [
+            Activation("sigmoid"),
+            Lambda(
+                lambda input_: input_
+                + tf.cast(tf.expand_dims(build_grid_coordinates(tf.shape(input_)[1:3]), 0), input_.dtype)
+            ),
+            Lambda(lambda input_: input_ / tf.cast(tf.shape(input_)[1:3], input_.dtype)),
+        ]
+    )

keras_fsl/models/branch_models/darknet.py

@@ -14,11 +14,7 @@ def conv_2d(*args, **kwargs):
 
 @wraps(Conv2D)
 def conv_block(*args, **kwargs):
-    return Sequential([
-        conv_2d(*args, **kwargs, use_bias=False),
-        BatchNormalization(),
-        LeakyReLU(alpha=0.1),
-    ])
+    return Sequential([conv_2d(*args, **kwargs, use_bias=False), BatchNormalization(), LeakyReLU(alpha=0.1),])
 
 
 def residual_block(input_shape, num_filters, num_blocks):