Skip to content

parse_image

Strategy class for parsing an image to a DLite instance.

LOGGER = logging.getLogger('oteapi_dlite.strategies') module-attribute

DLiteImageConfig

Bases: ImageParserConfig

Configuration for DLite image parser.

Source code in oteapi_dlite/strategies/parse_image.py 
31
32
33
34
35
36
37
38
39
class DLiteImageConfig(ImageParserConfig):
    """Configuration for DLite image parser."""

    image_label: Annotated[
        str,
        Field(
            description="Label to assign to the image in the collection.",
        ),
    ] = "image"

image_label: Annotated[str, Field(description='Label to assign to the image in the collection.')] = 'image' class-attribute instance-attribute

DLiteImageParseStrategy

Parse strategy for image files.

Registers strategies:

  • ("mediaType", "image/vnd.dlite-gif")
  • ("mediaType", "image/vnd.dlite-jpeg")
  • ("mediaType", "image/vnd.dlite-jpg")
  • ("mediaType", "image/vnd.dlite-jp2")
  • ("mediaType", "image/vnd.dlite-png")
  • ("mediaType", "image/vnd.dlite-tiff")
Source code in oteapi_dlite/strategies/parse_image.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 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
@dataclass
class DLiteImageParseStrategy:
    """Parse strategy for image files.

    **Registers strategies**:

    - `("mediaType", "image/vnd.dlite-gif")`
    - `("mediaType", "image/vnd.dlite-jpeg")`
    - `("mediaType", "image/vnd.dlite-jpg")`
    - `("mediaType", "image/vnd.dlite-jp2")`
    - `("mediaType", "image/vnd.dlite-png")`
    - `("mediaType", "image/vnd.dlite-tiff")`

    """

    parse_config: DLiteImageResourceConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific
        endpoint of the OTE-API Services.  It assumes that the image to
        parse is stored in a data cache, and can be retrieved via a key
        that is supplied in either the session (highest priority)
        or in the parser configuration (lowest priority).

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            DLite instance.
        """
        config = self.parse_config.configuration

        # Configuration for ImageDataParseStrategy in oteapi-core
        conf = self.parse_config.model_dump()
        conf["configuration"] = ImageParserConfig(
            **config.model_dump(), extra="ignore"
        )
        conf["mediaType"] = "image/" + conf["mediaType"].split("-")[-1]
        core_config = ImageParserResourceConfig(**conf)

        parse_strategy_session = ImageDataParseStrategy(core_config).initialize(
            session
        )
        output = ImageDataParseStrategy(core_config).get(parse_strategy_session)

        cache = DataCache()
        data = cache.get(output["image_key"])
        if isinstance(data, bytes):
            data = np.asarray(
                Image.frombytes(
                    data=data,
                    mode=output["image_mode"],
                    size=output["image_size"],
                )
            )
        if not isinstance(data, np.ndarray):
            raise TypeError(
                "Expected image data to be a numpy array, instead it was "
                f"{type(data)}."
            )

        meta = get_meta("http://onto-ns.com/meta/1.0/Image")
        inst = meta(dimensions=data.shape)
        inst["data"] = data

        LOGGER.info("session: %s", session)

        coll = get_collection(session)
        coll.add(config.image_label, inst)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

parse_config: DLiteImageResourceConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services. It assumes that the image to parse is stored in a data cache, and can be retrieved via a key that is supplied in either the session (highest priority) or in the parser configuration (lowest priority).

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteSessionUpdate

DLite instance.
Source code in oteapi_dlite/strategies/parse_image.py 
 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
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific
    endpoint of the OTE-API Services.  It assumes that the image to
    parse is stored in a data cache, and can be retrieved via a key
    that is supplied in either the session (highest priority)
    or in the parser configuration (lowest priority).

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        DLite instance.
    """
    config = self.parse_config.configuration

    # Configuration for ImageDataParseStrategy in oteapi-core
    conf = self.parse_config.model_dump()
    conf["configuration"] = ImageParserConfig(
        **config.model_dump(), extra="ignore"
    )
    conf["mediaType"] = "image/" + conf["mediaType"].split("-")[-1]
    core_config = ImageParserResourceConfig(**conf)

    parse_strategy_session = ImageDataParseStrategy(core_config).initialize(
        session
    )
    output = ImageDataParseStrategy(core_config).get(parse_strategy_session)

    cache = DataCache()
    data = cache.get(output["image_key"])
    if isinstance(data, bytes):
        data = np.asarray(
            Image.frombytes(
                data=data,
                mode=output["image_mode"],
                size=output["image_size"],
            )
        )
    if not isinstance(data, np.ndarray):
        raise TypeError(
            "Expected image data to be a numpy array, instead it was "
            f"{type(data)}."
        )

    meta = get_meta("http://onto-ns.com/meta/1.0/Image")
    inst = meta(dimensions=data.shape)
    inst["data"] = data

    LOGGER.info("session: %s", session)

    coll = get_collection(session)
    coll.add(config.image_label, inst)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/parse_image.py 
70
71
72
73
74
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteImageResourceConfig

Bases: ResourceConfig

Resource config for DLite image parser.

Source code in oteapi_dlite/strategies/parse_image.py 
42
43
44
45
46
47
48
49
50
class DLiteImageResourceConfig(ResourceConfig):
    """Resource config for DLite image parser."""

    configuration: Annotated[
        DLiteImageConfig,
        Field(
            description="Image parse strategy-specific configuration.",
        ),
    ] = DLiteImageConfig()

configuration: Annotated[DLiteImageConfig, Field(description='Image parse strategy-specific configuration.')] = DLiteImageConfig() class-attribute instance-attribute