base

class WorkflowImageData:

    def __init__(
        self,
        parent_metadata: ImageParentMetadata,
        workflow_root_ancestor_metadata: Optional[ImageParentMetadata] = None,
        image_reference: Optional[str] = None,
        base64_image: Optional[str] = None,
        numpy_image: Optional[np.ndarray] = None,
        video_metadata: Optional[VideoMetadata] = None,
    ):
        if not base64_image and numpy_image is None and not image_reference:
            raise ValueError("Could not initialise empty `WorkflowImageData`.")
        self._parent_metadata = parent_metadata
        self._workflow_root_ancestor_metadata = (
            workflow_root_ancestor_metadata
            if workflow_root_ancestor_metadata
            else self._parent_metadata
        )
        self._image_reference = image_reference
        self._base64_image = base64_image
        self._numpy_image = numpy_image
        self._video_metadata = video_metadata

    @classmethod
    def copy_and_replace(
        cls, origin_image_data: "WorkflowImageData", **kwargs
    ) -> "WorkflowImageData":
        """
        Creates new instance of `WorkflowImageData` with updated property.

        Properties are passed by kwargs, supported properties are:
        * parent_metadata
        * workflow_root_ancestor_metadata
        * image_reference
        * base64_image
        * numpy_image
        * video_metadata

        When more than one from ["numpy_image", "base64_image", "image_reference"] args are
        given, they MUST be compliant.
        """
        parent_metadata = origin_image_data._parent_metadata
        workflow_root_ancestor_metadata = (
            origin_image_data._workflow_root_ancestor_metadata
        )
        image_reference = origin_image_data._image_reference
        base64_image = origin_image_data._base64_image
        numpy_image = origin_image_data._numpy_image
        video_metadata = origin_image_data._video_metadata
        if any(k in kwargs for k in ["numpy_image", "base64_image", "image_reference"]):
            numpy_image = kwargs.get("numpy_image")
            base64_image = kwargs.get("base64_image")
            image_reference = kwargs.get("image_reference")
        if "parent_metadata" in kwargs:
            if workflow_root_ancestor_metadata is parent_metadata:
                workflow_root_ancestor_metadata = kwargs["parent_metadata"]
            parent_metadata = kwargs["parent_metadata"]
        if "workflow_root_ancestor_metadata" in kwargs:
            if parent_metadata is workflow_root_ancestor_metadata:
                parent_metadata = kwargs["workflow_root_ancestor_metadata"]
            workflow_root_ancestor_metadata = kwargs["workflow_root_ancestor_metadata"]
        if "video_metadata" in kwargs:
            video_metadata = kwargs["video_metadata"]
        return cls(
            parent_metadata=parent_metadata,
            workflow_root_ancestor_metadata=workflow_root_ancestor_metadata,
            image_reference=image_reference,
            base64_image=base64_image,
            numpy_image=numpy_image,
            video_metadata=video_metadata,
        )

    @classmethod
    def create_crop(
        cls,
        origin_image_data: "WorkflowImageData",
        crop_identifier: str,
        cropped_image: np.ndarray,
        offset_x: int,
        offset_y: int,
        preserve_video_metadata: bool = False,
    ) -> "WorkflowImageData":
        """
        Creates new instance of `WorkflowImageData` being a crop of original image,
        making adjustment to all metadata.
        """
        parent_metadata = ImageParentMetadata(
            parent_id=crop_identifier,
            origin_coordinates=OriginCoordinatesSystem(
                left_top_x=offset_x,
                left_top_y=offset_y,
                origin_width=origin_image_data.numpy_image.shape[1],
                origin_height=origin_image_data.numpy_image.shape[0],
            ),
        )
        workflow_root_ancestor_coordinates = replace(
            origin_image_data.workflow_root_ancestor_metadata.origin_coordinates,
            left_top_x=origin_image_data.workflow_root_ancestor_metadata.origin_coordinates.left_top_x
            + offset_x,
            left_top_y=origin_image_data.workflow_root_ancestor_metadata.origin_coordinates.left_top_y
            + offset_y,
        )
        workflow_root_ancestor_metadata = ImageParentMetadata(
            parent_id=origin_image_data.workflow_root_ancestor_metadata.parent_id,
            origin_coordinates=workflow_root_ancestor_coordinates,
        )
        video_metadata = None
        if preserve_video_metadata and origin_image_data._video_metadata is not None:
            video_metadata = copy(origin_image_data._video_metadata)
            video_metadata.video_identifier = (
                f"{video_metadata.video_identifier} | crop: {crop_identifier}"
            )
        return WorkflowImageData(
            parent_metadata=parent_metadata,
            workflow_root_ancestor_metadata=workflow_root_ancestor_metadata,
            numpy_image=cropped_image,
            video_metadata=video_metadata,
        )

    @property
    def parent_metadata(self) -> ImageParentMetadata:
        if self._parent_metadata.origin_coordinates is None:
            numpy_image = self.numpy_image
            origin_coordinates = OriginCoordinatesSystem(
                left_top_y=0,
                left_top_x=0,
                origin_width=numpy_image.shape[1],
                origin_height=numpy_image.shape[0],
            )
            self._parent_metadata = replace(
                self._parent_metadata, origin_coordinates=origin_coordinates
            )
        return self._parent_metadata

    @property
    def workflow_root_ancestor_metadata(self) -> ImageParentMetadata:
        if self._workflow_root_ancestor_metadata.origin_coordinates is None:
            numpy_image = self.numpy_image
            origin_coordinates = OriginCoordinatesSystem(
                left_top_y=0,
                left_top_x=0,
                origin_width=numpy_image.shape[1],
                origin_height=numpy_image.shape[0],
            )
            self._workflow_root_ancestor_metadata = replace(
                self._workflow_root_ancestor_metadata,
                origin_coordinates=origin_coordinates,
            )
        return self._workflow_root_ancestor_metadata

    @property
    def numpy_image(self) -> np.ndarray:
        if self._numpy_image is not None:
            return self._numpy_image
        if self._base64_image:
            self._numpy_image = attempt_loading_image_from_string(self._base64_image)[0]
            return self._numpy_image
        if self._image_reference.startswith(
            "http://"
        ) or self._image_reference.startswith("https://"):
            self._numpy_image = load_image_from_url(value=self._image_reference)
        else:
            self._numpy_image = cv2.imread(self._image_reference)
        return self._numpy_image

    @property
    def base64_image(self) -> str:
        if self._base64_image is not None:
            return self._base64_image
        numpy_image = self.numpy_image
        self._base64_image = base64.b64encode(
            encode_image_to_jpeg_bytes(numpy_image, jpeg_quality=95)
        ).decode("ascii")
        return self._base64_image

    @property
    def video_metadata(self) -> VideoMetadata:
        if self._video_metadata is not None:
            return self._video_metadata
        return VideoMetadata(
            video_identifier=self.parent_metadata.parent_id,
            frame_number=0,
            frame_timestamp=datetime.now(),
            fps=30,
            comes_from_video_file=None,
        )

    def to_inference_format(self, numpy_preferred: bool = False) -> Dict[str, Any]:
        if numpy_preferred:
            return {"type": "numpy_object", "value": self.numpy_image}
        if self._image_reference:
            if self._image_reference.startswith(
                "http://"
            ) or self._image_reference.startswith("https://"):
                return {"type": "url", "value": self._image_reference}
            return {"type": "file", "value": self._image_reference}
        if self._base64_image:
            return {"type": "base64", "value": self.base64_image}
        return {"type": "numpy_object", "value": self.numpy_image}