Skip to content

Asset

The Asset class enables access to the UP42 assets in the storage. Assets are results of orders or results of jobs with download blocks.

Use an existing asset: 

asset = up42.initialize_asset(asset_id="8c2dfb4d-bd35-435f-8667-48aea0dce2da")

Source code in up42/asset.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 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
135
136
137
138
139
140
141
142
143
144
145
146
class Asset:
    """
    The Asset class enables access to the UP42 assets in the storage. Assets are results
    of orders or results of jobs with download blocks.

    Use an existing asset:
    ```python
    asset = up42.initialize_asset(asset_id="8c2dfb4d-bd35-435f-8667-48aea0dce2da")
    ```
    """

    def __init__(self, auth: Auth, asset_id: str, asset_info: Optional[dict] = None):
        self.auth = auth
        self.asset_id = asset_id
        self.results: Union[List[str], None] = None
        if asset_info is not None:
            self._info = asset_info
        else:
            self._info = self.info

    def __repr__(self):
        representation = (
            f"Asset(name: {self._info['name']}, asset_id: {self.asset_id}, createdAt: {self._info['createdAt']}, "
            f"size: {self._info['size']})"
        )
        if "source" in self._info:
            representation += f", source: {self._info['source']}"
        if "contentType" in self._info:
            representation += f", contentType: {self._info['contentType']}"
        return representation

    @property
    def info(self) -> dict:
        """
        Gets and updates the asset metadata information.
        """
        url = f"{self.auth._endpoint()}/v2/assets/{self.asset_id}/metadata"
        response_json = self.auth._request(request_type="GET", url=url)
        self._info = response_json
        return self._info

    @property
    def stac_info(self) -> Union[dict, None]:
        """
        Gets the storage STAC information for the asset as a FeatureCollection.

        One asset can contain multiple STAC items (e.g. the pan- and multispectral images).
        """
        stac_search_parameters = {
            "max_items": 50,
            "limit": 50,
            "filter": {
                "op": "=",
                "args": [{"property": "asset_id"}, self.asset_id],
            },
        }
        url = f"{self.auth._endpoint()}/v2/assets/stac/search"
        stac_results = self.auth._request(
            request_type="POST", url=url, data=stac_search_parameters
        )
        stac_results.pop("links", None)
        if not stac_results["features"]:
            logger.info(
                "No STAC metadata information available for this asset's items!"
            )
        return stac_results

    def update_metadata(
        self, title: str = None, tags: List[str] = None, **kwargs
    ) -> dict:
        """
        Update the metadata of the asset.

        Args:
            title: The title string to be assigned to the asset.
            tags: A list of tag strings to be assigned to the asset.

        Returns:
            The updated asset metadata information
        """
        url = f"{self.auth._endpoint()}/v2/assets/{self.asset_id}/metadata"
        body_update = {"title": title, "tags": tags, **kwargs}
        response_json = self.auth._request(
            request_type="POST", url=url, data=body_update
        )
        self._info = response_json
        return self._info

    def _get_download_url(self) -> str:
        url = f"{self.auth._endpoint()}/v2/assets/{self.asset_id}/download-url"
        response_json = self.auth._request(request_type="POST", url=url)
        download_url = response_json["url"]
        return download_url

    def download(
        self, output_directory: Union[str, Path, None] = None, unpacking: bool = True
    ) -> List[str]:
        """
        Downloads the asset. Unpacking the downloaded file will happen as default.

        Args:
            output_directory: The file output directory, defaults to the current working
                directory.
            unpacking: By default the download TGZ/TAR or ZIP archive file will be unpacked.

        Returns:
            List of the downloaded asset filepaths.
        """
        logger.info(f"Downloading asset {self.asset_id}")

        if output_directory is None:
            output_directory = (
                Path.cwd() / f"project_{self.auth.project_id}/asset_{self.asset_id}"
            )
        else:
            output_directory = Path(output_directory)
        output_directory.mkdir(parents=True, exist_ok=True)
        logger.info(f"Download directory: {str(output_directory)}")

        download_url = self._get_download_url()
        if unpacking:
            out_filepaths = download_from_gcs_unpack(
                download_url=download_url,
                output_directory=output_directory,
            )
        else:
            out_filepaths = download_gcs_not_unpack(
                download_url=download_url,
                output_directory=output_directory,
            )

        self.results = out_filepaths
        return out_filepaths

Attributes

info: dict property

Gets and updates the asset metadata information.

stac_info: Union[dict, None] property

Gets the storage STAC information for the asset as a FeatureCollection.

One asset can contain multiple STAC items (e.g. the pan- and multispectral images).

Functions

download(output_directory=None, unpacking=True)

Downloads the asset. Unpacking the downloaded file will happen as default.

Parameters:

Name Type Description Default
output_directory Union[str, Path, None]

The file output directory, defaults to the current working directory.

 None
unpacking bool

By default the download TGZ/TAR or ZIP archive file will be unpacked.

 True

Returns:

Type Description
List[str]

List of the downloaded asset filepaths.
Source code in up42/asset.py 
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
146
def download(
    self, output_directory: Union[str, Path, None] = None, unpacking: bool = True
) -> List[str]:
    """
    Downloads the asset. Unpacking the downloaded file will happen as default.

    Args:
        output_directory: The file output directory, defaults to the current working
            directory.
        unpacking: By default the download TGZ/TAR or ZIP archive file will be unpacked.

    Returns:
        List of the downloaded asset filepaths.
    """
    logger.info(f"Downloading asset {self.asset_id}")

    if output_directory is None:
        output_directory = (
            Path.cwd() / f"project_{self.auth.project_id}/asset_{self.asset_id}"
        )
    else:
        output_directory = Path(output_directory)
    output_directory.mkdir(parents=True, exist_ok=True)
    logger.info(f"Download directory: {str(output_directory)}")

    download_url = self._get_download_url()
    if unpacking:
        out_filepaths = download_from_gcs_unpack(
            download_url=download_url,
            output_directory=output_directory,
        )
    else:
        out_filepaths = download_gcs_not_unpack(
            download_url=download_url,
            output_directory=output_directory,
        )

    self.results = out_filepaths
    return out_filepaths

update_metadata(title=None, tags=None, **kwargs)

Update the metadata of the asset.

Parameters:

Name Type Description Default
title str

The title string to be assigned to the asset.

 None
tags List[str]

A list of tag strings to be assigned to the asset.

 None

Returns:

Type Description
dict

The updated asset metadata information
Source code in up42/asset.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def update_metadata(
    self, title: str = None, tags: List[str] = None, **kwargs
) -> dict:
    """
    Update the metadata of the asset.

    Args:
        title: The title string to be assigned to the asset.
        tags: A list of tag strings to be assigned to the asset.

    Returns:
        The updated asset metadata information
    """
    url = f"{self.auth._endpoint()}/v2/assets/{self.asset_id}/metadata"
    body_update = {"title": title, "tags": tags, **kwargs}
    response_json = self.auth._request(
        request_type="POST", url=url, data=body_update
    )
    self._info = response_json
    return self._info