text: from __future__ import division
  from collections import defaultdict
  
  import random
  
  import numpy as np
  
  from albumentations.augmentations.keypoints_utils import KeypointsProcessor
  from albumentations.core.serialization import SerializableMeta, get_shortest_class_fullname
  from albumentations.core.six import add_metaclass
  from albumentations.core.transforms_interface import DualTransform
  from albumentations.core.utils import format_args, Params
  from albumentations.augmentations.bbox_utils import BboxProcessor
  from albumentations.core.serialization import SERIALIZABLE_REGISTRY, instantiate_lambda
  
  __all__ = ["Compose", "OneOf", "OneOrOther", "BboxParams", "KeypointParams", "ReplayCompose", "Sequential"]
  
  
  REPR_INDENT_STEP = 2
  
  
  class Transforms:
      def __init__(self, transforms):
          self.transforms = transforms
          self.start_end = self._find_dual_start_end(transforms)
  
      def _find_dual_start_end(self, transforms):
          dual_start_end = None
          last_dual = None
          for idx, transform in enumerate(transforms):
              if isinstance(transform, DualTransform):
                  last_dual = idx
                  if dual_start_end is None:
                      dual_start_end = [idx]
              if isinstance(transform, BaseCompose):
                  inside = self._find_dual_start_end(transform)
                  if inside is not None:
                      last_dual = idx
                      if dual_start_end is None:
                          dual_start_end = [idx]
          if dual_start_end is not None:
              dual_start_end.append(last_dual)
          return dual_start_end
  
      def get_always_apply(self, transforms):
          new_transforms = []
          for transform in transforms:
              if isinstance(transform, BaseCompose):
                  new_transforms.extend(self.get_always_apply(transform))
              elif transform.always_apply:
                  new_transforms.append(transform)
          return Transforms(new_transforms)
  
      def __getitem__(self, item):
          return self.transforms[item]
  
  
  def set_always_apply(transforms):
      for t in transforms:
          t.always_apply = True
  
  
  @add_metaclass(SerializableMeta)
  class BaseCompose:
      def __init__(self, transforms, p):
          self.transforms = Transforms(transforms)
          self.p = p
  
          self.replay_mode = False
          self.applied_in_replay = False
  
      def __getitem__(self, item):
          return self.transforms[item]
  
      def __repr__(self):
          return self.indented_repr()
  
      def indented_repr(self, indent=REPR_INDENT_STEP):
          args = {k: v for k, v in self._to_dict().items() if not (k.startswith("__") or k == "transforms")}
          repr_string = self.__class__.__name__ + "(["
          for t in self.transforms:
              repr_string += "\n"
              if hasattr(t, "indented_repr"):
                  t_repr = t.indented_repr(indent + REPR_INDENT_STEP)
              else:
                  t_repr = repr(t)
              repr_string += " " * indent + t_repr + ","
          repr_string += "\n" + " " * (indent - REPR_INDENT_STEP) + "], {args})".format(args=format_args(args))
          return repr_string
  
      @classmethod
      def get_class_fullname(cls):
          return get_shortest_class_fullname(cls)
  
      def _to_dict(self):
          return {
              "__class_fullname__": self.get_class_fullname(),
              "p": self.p,
              "transforms": [t._to_dict() for t in self.transforms],  # skipcq: PYL-W0212
          }
  
      def get_dict_with_id(self):
          return {
              "__class_fullname__": self.get_class_fullname(),
              "id": id(self),
              "params": None,
              "transforms": [t.get_dict_with_id() for t in self.transforms],
          }
  
      def add_targets(self, additional_targets):
          if additional_targets:
              for t in self.transforms:
                  t.add_targets(additional_targets)
  
      def set_deterministic(self, flag, save_key="replay"):
          for t in self.transforms:
              t.set_deterministic(flag, save_key)
  
  
  class Compose(BaseCompose):
      """Compose transforms and handle all transformations regarding bounding boxes
  
      Args:
          transforms (list): list of transformations to compose.
          bbox_params (BboxParams): Parameters for bounding boxes transforms
          keypoint_params (KeypointParams): Parameters for keypoints transforms
          additional_targets (dict): Dict with keys - new target name, values - old target name. ex: {'image2': 'image'}
          p (float): probability of applying all list of transforms. Default: 1.0.
      """
  
      def __init__(self, transforms, bbox_params=None, keypoint_params=None, additional_targets=None, p=1.0):
          super(Compose, self).__init__([t for t in transforms if t is not None], p)
  
          self.processors = {}
          if bbox_params:
              if isinstance(bbox_params, dict):
                  params = BboxParams(**bbox_params)
              elif isinstance(bbox_params, BboxParams):
                  params = bbox_params
              else:
                  raise ValueError("unknown format of bbox_params, please use `dict` or `BboxParams`")
              self.processors["bboxes"] = BboxProcessor(params, additional_targets)
  
          if keypoint_params:
              if isinstance(keypoint_params, dict):
                  params = KeypointParams(**keypoint_params)
              elif isinstance(keypoint_params, KeypointParams):
                  params = keypoint_params
              else:
                  raise ValueError("unknown format of keypoint_params, please use `dict` or `KeypointParams`")
              self.processors["keypoints"] = KeypointsProcessor(params, additional_targets)
  
          if additional_targets is None:
              additional_targets = {}
  
          self.additional_targets = additional_targets
  
          for proc in self.processors.values():
              proc.ensure_transforms_valid(self.transforms)
  
          self.add_targets(additional_targets)
  
      def __call__(self, *args, force_apply=False, **data):
          if args:
              raise KeyError("You have to pass data to augmentations as named arguments, for example: aug(image=image)")
          self._check_args(**data)
          assert isinstance(force_apply, (bool, int)), "force_apply must have bool or int type"
          need_to_run = force_apply or random.random() < self.p
          for p in self.processors.values():
              p.ensure_data_valid(data)
          transforms = self.transforms if need_to_run else self.transforms.get_always_apply(self.transforms)
          dual_start_end = transforms.start_end if self.processors else None
          check_each_transform = any(
              getattr(item.params, "check_each_transform", False) for item in self.processors.values()
          )
  
          for idx, t in enumerate(transforms):
              if dual_start_end is not None and idx == dual_start_end[0]:
                  for p in self.processors.values():
                      p.preprocess(data)
  
              data = t(force_apply=force_apply, **data)
  
              if dual_start_end is not None and idx == dual_start_end[1]:
                  for p in self.processors.values():
                      p.postprocess(data)
              elif check_each_transform and isinstance(t, DualTransform):
                  rows, cols = data["image"].shape[:2]
                  for p in self.processors.values():
                      if not getattr(p.params, "check_each_transform", False):
                          continue
  
                      for data_name in p.data_fields:
                          data[data_name] = p.filter(data[data_name], rows, cols)
  
          return data
  
      def _to_dict(self):
          dictionary = super(Compose, self)._to_dict()
          bbox_processor = self.processors.get("bboxes")
          keypoints_processor = self.processors.get("keypoints")
          dictionary.update(
              {
                  "bbox_params": bbox_processor.params._to_dict() if bbox_processor else None,  # skipcq: PYL-W0212
                  "keypoint_params": keypoints_processor.params._to_dict()  # skipcq: PYL-W0212
                  if keypoints_processor
                  else None,
                  "additional_targets": self.additional_targets,
              }
          )
          return dictionary
  
      def get_dict_with_id(self):
          dictionary = super().get_dict_with_id()
          bbox_processor = self.processors.get("bboxes")
          keypoints_processor = self.processors.get("keypoints")
          dictionary.update(
              {
                  "bbox_params": bbox_processor.params._to_dict() if bbox_processor else None,  # skipcq: PYL-W0212
                  "keypoint_params": keypoints_processor.params._to_dict()  # skipcq: PYL-W0212
                  if keypoints_processor
                  else None,
                  "additional_targets": self.additional_targets,
                  "params": None,
              }
          )
          return dictionary
  
      def _check_args(self, **kwargs):
          checked_single = ["image", "mask"]
          checked_multi = ["masks"]
          # ["bboxes", "keypoints"] could be almost any type, no need to check them
          for data_name, data in kwargs.items():
              internal_data_name = self.additional_targets.get(data_name, data_name)
              if internal_data_name in checked_single:
                  if not isinstance(data, np.ndarray):
                      raise TypeError("{} must be numpy array type".format(data_name))
              if internal_data_name in checked_multi:
                  if data:
                      if not isinstance(data[0], np.ndarray):
                          raise TypeError("{} must be list of numpy arrays".format(data_name))
  
  
  class OneOf(BaseCompose):
      """Select one of transforms to apply. Selected transform will be called with `force_apply=True`.
      Transforms probabilities will be normalized to one 1, so in this case transforms probabilities works as weights.
  
      Args:
          transforms (list): list of transformations to compose.
          p (float): probability of applying selected transform. Default: 0.5.
      """
  
      def __init__(self, transforms, p=0.5):
          super(OneOf, self).__init__(transforms, p)
          transforms_ps = [t.p for t in transforms]
          s = sum(transforms_ps)
          self.transforms_ps = [t / s for t in transforms_ps]
  
      def __call__(self, force_apply=False, **data):
          if self.replay_mode:
              for t in self.transforms:
                  data = t(**data)
              return data
  
          if self.transforms_ps and (force_apply or random.random() < self.p):
              random_state = np.random.RandomState(random.randint(0, 2 ** 32 - 1))
              t = random_state.choice(self.transforms.transforms, p=self.transforms_ps)
              data = t(force_apply=True, **data)
          return data
  
  
  class OneOrOther(BaseCompose):
      """Select one or another transform to apply. Selected transform will be called with `force_apply=True`."""
  
      def __init__(self, first=None, second=None, transforms=None, p=0.5):
          if transforms is None:
              transforms = [first, second]
          super(OneOrOther, self).__init__(transforms, p)
  
      def __call__(self, force_apply=False, **data):
          if self.replay_mode:
              for t in self.transforms:
                  data = t(**data)
              return data
  
          if random.random() < self.p:
              return self.transforms[0](force_apply=True, **data)
  
          return self.transforms[-1](force_apply=True, **data)
  
  
  class PerChannel(BaseCompose):
      """Apply transformations per-channel
  
      Args:
          transforms (list): list of transformations to compose.
          channels (list): channels to apply the transform to. Pass None to apply to all.
                           Default: None (apply to all)
          p (float): probability of applying the transform. Default: 0.5.
      """
  
      def __init__(self, transforms, channels=None, p=0.5):
          super(PerChannel, self).__init__(transforms, p)
          self.channels = channels
  
      def __call__(self, force_apply=False, **data):
          if force_apply or random.random() < self.p:
  
              image = data["image"]
  
              # Expand mono images to have a single channel
              if len(image.shape) == 2:
                  image = np.expand_dims(image, -1)
  
              if self.channels is None:
                  self.channels = range(image.shape[2])
  
              for c in self.channels:
                  for t in self.transforms:
                      image[:, :, c] = t(image=image[:, :, c])["image"]
  
              data["image"] = image
  
          return data
  
  
  class ReplayCompose(Compose):
      def __init__(
          self, transforms, bbox_params=None, keypoint_params=None, additional_targets=None, p=1.0, save_key="replay"
      ):
          super(ReplayCompose, self).__init__(transforms, bbox_params, keypoint_params, additional_targets, p)
          self.set_deterministic(True, save_key=save_key)
          self.save_key = save_key
  
      def __call__(self, force_apply=False, **kwargs):
          kwargs[self.save_key] = defaultdict(dict)
          result = super(ReplayCompose, self).__call__(force_apply=force_apply, **kwargs)
          serialized = self.get_dict_with_id()
          self.fill_with_params(serialized, result[self.save_key])
          self.fill_applied(serialized)
          result[self.save_key] = serialized
          return result
  
      @staticmethod
      def replay(saved_augmentations, **kwargs):
          augs = ReplayCompose._restore_for_replay(saved_augmentations)
          return augs(force_apply=True, **kwargs)
  
      @staticmethod
      def _restore_for_replay(transform_dict, lambda_transforms=None):
          """
          Args:
              transform (dict): A dictionary with serialized transform pipeline.
              lambda_transforms (dict): A dictionary that contains lambda transforms, that
              is instances of the Lambda class.
                  This dictionary is required when you are restoring a pipeline that contains lambda transforms. Keys
                  in that dictionary should be named same as `name` arguments in respective lambda transforms from
                  a serialized pipeline.
          """
          transform = transform_dict
          applied = transform["applied"]
          params = transform["params"]
          lmbd = instantiate_lambda(transform, lambda_transforms)
          if lmbd:
              transform = lmbd
          else:
              name = transform["__class_fullname__"]
              args = {k: v for k, v in transform.items() if k not in ["__class_fullname__", "applied", "params"]}
              cls = SERIALIZABLE_REGISTRY[name]
              if "transforms" in args:
                  args["transforms"] = [
                      ReplayCompose._restore_for_replay(t, lambda_transforms=lambda_transforms)
                      for t in args["transforms"]
                  ]
              transform = cls(**args)
  
          transform.params = params
          transform.replay_mode = True
          transform.applied_in_replay = applied
          return transform
  
      def fill_with_params(self, serialized, all_params):
          params = all_params.get(serialized.get("id"))
          serialized["params"] = params
          del serialized["id"]
          for transform in serialized.get("transforms", []):
              self.fill_with_params(transform, all_params)
  
      def fill_applied(self, serialized):
          if "transforms" in serialized:
              applied = [self.fill_applied(t) for t in serialized["transforms"]]
              serialized["applied"] = any(applied)
          else:
              serialized["applied"] = serialized.get("params") is not None
          return serialized["applied"]
  
      def _to_dict(self):
          dictionary = super(ReplayCompose, self)._to_dict()
          dictionary.update({"save_key": self.save_key})
          return dictionary
  
  
  class BboxParams(Params):
      """
      Parameters of bounding boxes
  
      Args:
          format (str): format of bounding boxes. Should be 'coco', 'pascal_voc', 'albumentations' or 'yolo'.
  
              The `coco` format
                  `[x_min, y_min, width, height]`, e.g. [97, 12, 150, 200].
              The `pascal_voc` format
                  `[x_min, y_min, x_max, y_max]`, e.g. [97, 12, 247, 212].
              The `albumentations` format
                  is like `pascal_voc`, but normalized,
                  in other words: [x_min, y_min, x_max, y_max]`, e.g. [0.2, 0.3, 0.4, 0.5].
              The `yolo` format
                  `[x, y, width, height]`, e.g. [0.1, 0.2, 0.3, 0.4];
                  `x`, `y` - normalized bbox center; `width`, `height` - normalized bbox width and height.
          label_fields (list): list of fields that are joined with boxes, e.g labels.
              Should be same type as boxes.
          min_area (float): minimum area of a bounding box. All bounding boxes whose
              visible area in pixels is less than this value will be removed. Default: 0.0.
          min_visibility (float): minimum fraction of area for a bounding box
              to remain this box in list. Default: 0.0.
          check_each_transform (bool): if `True`, then bboxes will be checked after each dual transform.
              Default: `True`
      """
  
      def __init__(self, format, label_fields=None, min_area=0.0, min_visibility=0.0, check_each_transform=True):
          super(BboxParams, self).__init__(format, label_fields)
          self.min_area = min_area
          self.min_visibility = min_visibility
          self.check_each_transform = check_each_transform
  
      def _to_dict(self):
          data = super(BboxParams, self)._to_dict()
          data.update(
              {
                  "min_area": self.min_area,
                  "min_visibility": self.min_visibility,
                  "check_each_transform": self.check_each_transform,
              }
          )
          return data
  
  
  class KeypointParams(Params):
      """
      Parameters of keypoints
  
      Args:
          format (str): format of keypoints. Should be 'xy', 'yx', 'xya', 'xys', 'xyas', 'xysa'.
  
              x - X coordinate,
  
              y - Y coordinate
  
              s - Keypoint scale
  
              a - Keypoint orientation in radians or degrees (depending on KeypointParams.angle_in_degrees)
          label_fields (list): list of fields that are joined with keypoints, e.g labels.
              Should be same type as keypoints.
          remove_invisible (bool): to remove invisible points after transform or not
          angle_in_degrees (bool): angle in degrees or radians in 'xya', 'xyas', 'xysa' keypoints
          check_each_transform (bool): if `True`, then keypoints will be checked after each dual transform.
              Default: `True`
      """
  
      def __init__(
          self,
          format,  # skipcq: PYL-W0622
          label_fields=None,
          remove_invisible=True,
          angle_in_degrees=True,
          check_each_transform=True,
      ):
          super(KeypointParams, self).__init__(format, label_fields)
          self.remove_invisible = remove_invisible
          self.angle_in_degrees = angle_in_degrees
          self.check_each_transform = check_each_transform
  
      def _to_dict(self):
          data = super(KeypointParams, self)._to_dict()
          data.update(
              {
                  "remove_invisible": self.remove_invisible,
                  "angle_in_degrees": self.angle_in_degrees,
                  "check_each_transform": self.check_each_transform,
              }
          )
          return data
  
  
  class Sequential(BaseCompose):
      """Sequentially applies all transforms to targets.
  
      Note:
          This transform is not intended to be a replacement for `Compose`. Instead, it should be used inside `Compose`
          the same way `OneOf` or `OneOrOther` are used. For instance, you can combine `OneOf` with `Sequential` to
          create an augmentation pipeline that contains multiple sequences of augmentations and applies one randomly
          chose sequence to input data (see the `Example` section for an example definition of such pipeline).
  
      Example:
          >>> import albumentations as A
          >>> transform = A.Compose([
          >>>    A.OneOf([
          >>>        A.Sequential([
          >>>            A.HorizontalFlip(p=0.5),
          >>>            A.ShiftScaleRotate(p=0.5),
          >>>        ]),
          >>>        A.Sequential([
          >>>            A.VerticalFlip(p=0.5),
          >>>            A.RandomBrightnessContrast(p=0.5),
          >>>        ]),
          >>>    ], p=1)
          >>> ])
      """
  
      def __init__(self, transforms, p=0.5):
          super().__init__(transforms, p)
  
      def __call__(self, **data):
          for t in self.transforms:
              data = t(**data)
          return data
  