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
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.workspace_id = auth.workspace_id
        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):
        return (
            f"Asset(name: {self._info['name']}, asset_id: {self.asset_id}, type: {self._info['type']}, "
            f"source: {self._info['source']}, createdAt: {self._info['createdAt']}, "
            f"size: {self._info['size']})"
        )

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

    @property
    def source(self) -> dict:
        """
        Gets the source of the Asset. One of `TASKING`, `ORDER`, `BLOCK`, `UPLOAD`.
        """
        source = self.info["source"]
        logger.info(f"Asset source is {source}")
        return source

    def _get_download_url(self) -> str:
        url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/assets/{self.asset_id}/downloadUrl"
        response_json = self.auth._request(request_type="GET", url=url)
        download_url = response_json["data"]["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 final result which is in TAR or ZIP archive format 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_results_from_gcs(
                download_url=download_url,
                output_directory=output_directory,
            )
        else:
            out_filepaths = download_results_from_gcs_without_unpacking(
                download_url=download_url,
                output_directory=output_directory,
            )

        self.results = out_filepaths
        return out_filepaths

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 final result which is in TAR or ZIP archive format will be unpacked.

 True

Returns:

Type Description
List[str]

List of the downloaded asset filepaths.
Source code in up42/asset.py 
 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
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 final result which is in TAR or ZIP archive format 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_results_from_gcs(
            download_url=download_url,
            output_directory=output_directory,
        )
    else:
        out_filepaths = download_results_from_gcs_without_unpacking(
            download_url=download_url,
            output_directory=output_directory,
        )

    self.results = out_filepaths
    return out_filepaths

info() property

Gets and updates the asset metadata information.

Source code in up42/asset.py 
42
43
44
45
46
47
48
49
50
@property
def info(self) -> dict:
    """
    Gets and updates the asset metadata information.
    """
    url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/assets/{self.asset_id}"
    response_json = self.auth._request(request_type="GET", url=url)
    self._info = response_json["data"]
    return self._info

source() property

Gets the source of the Asset. One of TASKING, ORDER, BLOCK, UPLOAD.

Source code in up42/asset.py 
52
53
54
55
56
57
58
59
@property
def source(self) -> dict:
    """
    Gets the source of the Asset. One of `TASKING`, `ORDER`, `BLOCK`, `UPLOAD`.
    """
    source = self.info["source"]
    logger.info(f"Asset source is {source}")
    return source