Skip to content

image

Strategy class for image/jpg.

ImageDataParseStrategy

Parse strategy for images.

Registers strategies:

  • ("mediaType", "image/jpg")
  • ("mediaType", "image/jpeg")
  • ("mediaType", "image/jp2")
  • ("mediaType", "image/png")
  • ("mediaType", "image/gif")
  • ("mediaType", "image/tiff")
  • ("mediaType", "image/eps")
Source code in oteapi/strategies/parse/image.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@dataclass
class ImageDataParseStrategy:
    """Parse strategy for images.

    **Registers strategies**:

    - `("mediaType", "image/jpg")`
    - `("mediaType", "image/jpeg")`
    - `("mediaType", "image/jp2")`
    - `("mediaType", "image/png")`
    - `("mediaType", "image/gif")`
    - `("mediaType", "image/tiff")`
    - `("mediaType", "image/eps")`

    """

    parse_config: ImageParserResourceConfig

    def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
        """Initialize strategy."""
        return SessionUpdate()

    def get(
        self, session: "Optional[Dict[str, Any]]" = None
    ) -> SessionUpdateImageParse:
        """Execute the strategy."""
        if session:
            self._use_filters(session)
        session = session if session else {}

        mime_format = self.parse_config.mediaType.split("/")[1]
        image_format = self._map_mime_to_format()

        # Retrieve image file
        download_config = ResourceConfig(**self.parse_config.dict())
        del download_config.configuration
        downloader = create_strategy("download", download_config)
        session.update(downloader.initialize(session))
        cache_key = downloader.get(session).get("key", "")

        cache = DataCache(self.parse_config.configuration.datacache_config)

        # Treat image according to filter values
        with cache.getfile(cache_key, suffix=mime_format) as filename:
            image = Image.open(filename, formats=[image_format]).crop(
                self.parse_config.configuration.crop
            )

        if image_format == "GIF":
            if image.info.get("version", b"").startswith(b"GIF"):
                image.info.update(
                    {"version": image.info.get("version", b"")[len(b"GIF") :]}
                )

        # Return parsed and treated image
        image_content = BytesIO()
        image.save(image_content, format=image_format)
        return SessionUpdateImageParse(
            content=image_content.getvalue(),
            format=image_format,
            cropped=bool(self.parse_config.configuration.crop),
        )

    def _use_filters(self, session: "Dict[str, Any]") -> None:
        """Update `config` according to filter values found in the session."""
        if "imagecrop" in session and not self.parse_config.configuration.crop:
            # Use CropFilter available in session
            self.parse_config.configuration.crop = session["imagecrop"]

    def _map_mime_to_format(self) -> str:
        """Return mapped Pillow-understandable format."""
        image_format = self.parse_config.mediaType.split("/")[1]
        return {
            "jpg": "JPEG",
            "jp2": "JPEG2000",
        }.get(image_format, image_format.upper())

_map_mime_to_format(self)

Return mapped Pillow-understandable format.

Source code in oteapi/strategies/parse/image.py 
139
140
141
142
143
144
145
def _map_mime_to_format(self) -> str:
    """Return mapped Pillow-understandable format."""
    image_format = self.parse_config.mediaType.split("/")[1]
    return {
        "jpg": "JPEG",
        "jp2": "JPEG2000",
    }.get(image_format, image_format.upper())

_use_filters(self, session)

Update config according to filter values found in the session.

Source code in oteapi/strategies/parse/image.py 
133
134
135
136
137
def _use_filters(self, session: "Dict[str, Any]") -> None:
    """Update `config` according to filter values found in the session."""
    if "imagecrop" in session and not self.parse_config.configuration.crop:
        # Use CropFilter available in session
        self.parse_config.configuration.crop = session["imagecrop"]

get(self, session=None)

Execute the strategy.

Source code in oteapi/strategies/parse/image.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get(
    self, session: "Optional[Dict[str, Any]]" = None
) -> SessionUpdateImageParse:
    """Execute the strategy."""
    if session:
        self._use_filters(session)
    session = session if session else {}

    mime_format = self.parse_config.mediaType.split("/")[1]
    image_format = self._map_mime_to_format()

    # Retrieve image file
    download_config = ResourceConfig(**self.parse_config.dict())
    del download_config.configuration
    downloader = create_strategy("download", download_config)
    session.update(downloader.initialize(session))
    cache_key = downloader.get(session).get("key", "")

    cache = DataCache(self.parse_config.configuration.datacache_config)

    # Treat image according to filter values
    with cache.getfile(cache_key, suffix=mime_format) as filename:
        image = Image.open(filename, formats=[image_format]).crop(
            self.parse_config.configuration.crop
        )

    if image_format == "GIF":
        if image.info.get("version", b"").startswith(b"GIF"):
            image.info.update(
                {"version": image.info.get("version", b"")[len(b"GIF") :]}
            )

    # Return parsed and treated image
    image_content = BytesIO()
    image.save(image_content, format=image_format)
    return SessionUpdateImageParse(
        content=image_content.getvalue(),
        format=image_format,
        cropped=bool(self.parse_config.configuration.crop),
    )

initialize(self, session=None)

Initialize strategy.

Source code in oteapi/strategies/parse/image.py 
88
89
90
def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
    """Initialize strategy."""
    return SessionUpdate()

ImageParserConfig

Bases: AttrDict

Configuration data model for ImageDataParseStrategy.

Source code in oteapi/strategies/parse/image.py 
19
20
21
22
23
24
25
26
27
28
29
30
class ImageParserConfig(AttrDict):
    """Configuration data model for
    [`ImageDataParseStrategy`][oteapi.strategies.parse.image.ImageDataParseStrategy]."""

    crop: Optional[Tuple[int, int, int, int]] = Field(
        None,
        description="Box cropping parameters.",
    )
    datacache_config: Optional[DataCacheConfig] = Field(
        None,
        description="Configuration options for the local data cache.",
    )

ImageParserResourceConfig

Bases: ResourceConfig

Image parse strategy resource config.

Source code in oteapi/strategies/parse/image.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class ImageParserResourceConfig(ResourceConfig):
    """Image parse strategy resource config."""

    configuration: ImageParserConfig = Field(
        ImageParserConfig(), description="Image parse strategy-specific configuration."
    )
    mediaType: str = Field(
        ...,
        description=(
            "The media type of the distribution as defined by IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]"
            ".\n\nUsage: This property *SHOULD* be used when the media"
            " type of the distribution is defined in IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]."
        ),
    )

SessionUpdateImageParse

Bases: SessionUpdate

Configuration model for ImageParse.

Source code in oteapi/strategies/parse/image.py 
62
63
64
65
66
67
class SessionUpdateImageParse(SessionUpdate):
    """Configuration model for ImageParse."""

    content: bytes = Field(..., description="Parsed output from ImageParse.")
    format: SupportedFormat = Field(..., description="The format of the parsed image.")
    cropped: bool = Field(..., description="Whether or not the image has been cropped.")

SupportedFormat

Bases: Enum

Supported formats for ImageDataParseStrategy.

Source code in oteapi/strategies/parse/image.py 
51
52
53
54
55
56
57
58
59
class SupportedFormat(Enum):
    """Supported formats for `ImageDataParseStrategy`."""

    JPEG = "JPEG"
    JPEG2000 = "JPEG2000"
    PNG = "PNG"
    GIF = "GIF"
    TIFF = "TIFF"
    EPS = "EPS"
Back to top