Skip to content

up42

up42 is the base library module imported to Python. It provides the elementary functionality that is not bound to a specific class of the UP42 structure (Project > Workflow > Job etc.). From up42 you can initialize other existing objects, get information about UP42 data & processing blocks, read or draw vector data, and adjust the SDK settings.

To import the UP42 library:

import up42

initialize_project()

Returns the correct Project object (has to exist on UP42).

Source code in up42/__init__.py
58
59
60
61
62
63
64
65
66
def initialize_project() -> "Project":
    """
    Returns the correct Project object (has to exist on UP42).
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    project = Project(auth=_auth, project_id=str(_auth.project_id))
    logger.info(f"Initialized {project}")
    return project

initialize_catalog()

Returns a Catalog object for using the catalog search.

Source code in up42/__init__.py
69
70
71
72
73
74
75
def initialize_catalog() -> "Catalog":
    """
    Returns a Catalog object for using the catalog search.
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    return Catalog(auth=_auth)

initialize_workflow(workflow_id)

Returns a Workflow object (has to exist on UP42).

Parameters:

Name Type Description Default
workflow_id str

The UP42 workflow_id

required
Source code in up42/__init__.py
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def initialize_workflow(workflow_id: str) -> "Workflow":
    """
    Returns a Workflow object (has to exist on UP42).

    Args:
        workflow_id: The UP42 workflow_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    workflow = Workflow(
        auth=_auth, workflow_id=workflow_id, project_id=str(_auth.project_id)
    )
    logger.info(f"Initialized {workflow}")
    return workflow

initialize_job(job_id)

Returns a Job object (has to exist on UP42).

Parameters:

Name Type Description Default
job_id str

The UP42 job_id

required
Source code in up42/__init__.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def initialize_job(job_id: str) -> "Job":
    """
    Returns a Job object (has to exist on UP42).

    Args:
        job_id: The UP42 job_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    job = Job(auth=_auth, job_id=job_id, project_id=str(_auth.project_id))
    logger.info(f"Initialized {job}")
    return job

initialize_jobtask(jobtask_id, job_id)

Returns a JobTask object (has to exist on UP42).

Parameters:

Name Type Description Default
jobtask_id str

The UP42 jobtask_id

required
job_id str

The UP42 job_id

required
Source code in up42/__init__.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def initialize_jobtask(jobtask_id: str, job_id: str) -> "JobTask":
    """
    Returns a JobTask object (has to exist on UP42).

    Args:
        jobtask_id: The UP42 jobtask_id
        job_id: The UP42 job_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    jobtask = JobTask(
        auth=_auth,
        jobtask_id=jobtask_id,
        job_id=job_id,
        project_id=str(_auth.project_id),
    )
    logger.info(f"Initialized {jobtask}")
    return jobtask

initialize_jobcollection(job_ids)

Returns a JobCollection object (the referenced jobs have to exist on UP42).

Parameters:

Name Type Description Default
job_ids List[str]

List of UP42 job_ids

required
Source code in up42/__init__.py
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def initialize_jobcollection(job_ids: List[str]) -> "JobCollection":
    """
    Returns a JobCollection object (the referenced jobs have to exist on UP42).

    Args:
        job_ids: List of UP42 job_ids
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    jobs = [
        Job(auth=_auth, job_id=job_id, project_id=str(_auth.project_id))
        for job_id in job_ids
    ]
    jobcollection = JobCollection(
        auth=_auth, project_id=str(_auth.project_id), jobs=jobs
    )
    logger.info(f"Initialized {jobcollection}")
    return jobcollection

initialize_storage()

Returns a Storage object to list orders and assets.

Source code in up42/__init__.py
148
149
150
151
152
153
154
def initialize_storage() -> "Storage":
    """
    Returns a Storage object to list orders and assets.
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    return Storage(auth=_auth)

initialize_order(order_id)

Returns an Order object (has to exist on UP42).

Parameters:

Name Type Description Default
order_id str

The UP42 order_id

required
Source code in up42/__init__.py
157
158
159
160
161
162
163
164
165
166
167
168
def initialize_order(order_id: str) -> "Order":
    """
    Returns an Order object (has to exist on UP42).

    Args:
        order_id: The UP42 order_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    order = Order(auth=_auth, order_id=order_id)
    logger.info(f"Initialized {order}")
    return order

initialize_asset(asset_id)

Returns an Asset object (has to exist on UP42).

Parameters:

Name Type Description Default
asset_id str

The UP42 asset_id

required
Source code in up42/__init__.py
171
172
173
174
175
176
177
178
179
180
181
182
def initialize_asset(asset_id: str) -> "Asset":
    """
    Returns an Asset object (has to exist on UP42).

    Args:
        asset_id: The UP42 asset_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    asset = Asset(auth=_auth, asset_id=asset_id)
    logger.info(f"Initialized {asset}")
    return asset

settings(log=True)

Configures settings about logging etc. when using the up42-py package.

Parameters:

Name Type Description Default
log bool

Activates/deactivates logging, default True is activated logging.

True
Source code in up42/__init__.py
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def settings(log: bool = True) -> None:
    """
    Configures settings about logging etc. when using the up42-py package.

    Args:
        log: Activates/deactivates logging, default True is activated logging.
    """
    if log:
        logger.info(
            "Logging enabled (default) - use up42.settings(log=False) to disable."
        )
    else:
        logger.info("Logging disabled - use up42.settings(log=True) to reactivate.")

    for name in logging.root.manager.loggerDict:
        setattr(logging.getLogger(name), "disabled", not log)

initialize_webhook(webhook_id)

Returns a Webhook object (has to exist on UP42).

Parameters:

Name Type Description Default
webhook_id str

The UP42 webhook_id

required
Source code in up42/__init__.py
282
283
284
285
286
287
288
289
290
291
292
293
def initialize_webhook(webhook_id: str) -> Webhook:
    """
    Returns a Webhook object (has to exist on UP42).

    Args:
        webhook_id: The UP42 webhook_id
    """
    if _auth is None:
        raise RuntimeError("Not authenticated, call up42.authenticate() first")
    webhook = Webhook(auth=_auth, webhook_id=webhook_id)
    logger.info(f"Initialized {webhook}")
    return webhook

get_webhooks(return_json=False)

Gets all registered webhooks for this workspace.

Parameters:

Name Type Description Default
return_json bool

If true returns the webhooks information as json instead of webhook class objects.

False

Returns:

Type Description
List[Webhook]

A list of the registered webhooks for this workspace.

Source code in up42/__init__.py
296
297
298
299
300
301
302
303
304
305
306
307
def get_webhooks(return_json: bool = False) -> List[Webhook]:
    """
    Gets all registered webhooks for this workspace.

    Args:
        return_json: If true returns the webhooks information as json instead of webhook class objects.

    Returns:
        A list of the registered webhooks for this workspace.
    """
    webhooks = Webhooks(auth=_auth).get_webhooks(return_json=return_json)
    return webhooks

create_webhook(name, url, events, active=False, secret=None)

Registers a new webhook in the system.

Parameters:

Name Type Description Default
name str

Webhook name

required
url str

Unique URL where the webhook will send the message (HTTPS required)

required
events List[str]

List of event types (order status / job task status)

required
active bool

Webhook status.

False
secret Optional[str]

String that acts as signature to the https request sent to the url.

None

Returns:

Type Description

A dict with details of the registered webhook.

Source code in up42/__init__.py
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
def create_webhook(
    name: str,
    url: str,
    events: List[str],
    active: bool = False,
    secret: Optional[str] = None,
):
    """
    Registers a new webhook in the system.

    Args:
        name: Webhook name
        url: Unique URL where the webhook will send the message (HTTPS required)
        events: List of event types (order status / job task status)
        active: Webhook status.
        secret: String that acts as signature to the https request sent to the url.

    Returns:
        A dict with details of the registered webhook.
    """
    webhook = Webhooks(auth=_auth).create_webhook(
        name=name, url=url, events=events, active=active, secret=secret
    )
    return webhook

get_webhook_events()

Gets all available webhook events.

Returns:

Type Description
dict

A dict of the available webhook events.

Source code in up42/__init__.py
336
337
338
339
340
341
342
343
344
def get_webhook_events() -> dict:
    """
    Gets all available webhook events.

    Returns:
        A dict of the available webhook events.
    """
    webhook_events = Webhooks(auth=_auth).get_webhook_events()
    return webhook_events
Source code in up42/tools.py
 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
class Tools:
    def __init__(self, auth=None):
        """
        The tools class contains the base functionality that is not bound to a specific
        higher level UP42 object.
        """
        if auth:
            self.auth = auth
        self.quicklooks = None
        self.results = None

    # pylint: disable=no-self-use
    def read_vector_file(
        self, filename: str = "aoi.geojson", as_dataframe: bool = False
    ) -> Union[dict, GeoDataFrame]:
        """
        Reads vector files (geojson, shapefile, kml, wkt) to a feature collection,
        for use as the aoi geometry in the workflow input parameters
        (see get_input_parameters).

        Example aoi fiels are provided, e.g. example/data/aoi_Berlin.geojson

        Args:
            filename: File path of the vector file.
            as_dataframe: Return type, default FeatureCollection, GeoDataFrame if True.

        Returns:
            Feature Collection
        """
        suffix = Path(filename).suffix

        if suffix == ".kml":
            gpd.io.file.fiona.drvsupport.supported_drivers["KML"] = "rw"
            df = gpd.read_file(filename, driver="KML")
        elif suffix == ".wkt":
            with open(filename) as wkt_file:
                wkt = wkt_file.read()
                df = pd.DataFrame({"geometry": [wkt]})
                df["geometry"] = df["geometry"].apply(shapely.wkt.loads)
                df = GeoDataFrame(df, geometry="geometry", crs=4326)
        else:
            df = gpd.read_file(filename)

        if df.crs.to_string() != "EPSG:4326":
            df = df.to_crs(epsg=4326)
        if as_dataframe:
            return df
        else:
            return df.__geo_interface__

    def get_example_aoi(
        self, location: str = "Berlin", as_dataframe: bool = False
    ) -> Union[dict, GeoDataFrame]:
        """
        Gets predefined, small, rectangular example aoi for the selected location.

        Args:
            location: Location, one of Berlin, Washington.
            as_dataframe: Returns a dataframe instead of dict FeatureColletions
                (default).

        Returns:
            Feature collection json with the selected aoi.
        """
        logger.info(f"Getting small example aoi in location '{location}'.")
        if location == "Berlin":
            example_aoi = self.read_vector_file(
                f"{str(Path(__file__).resolve().parent)}/data/aoi_berlin.geojson"
            )
        elif location == "Washington":
            example_aoi = self.read_vector_file(
                f"{str(Path(__file__).resolve().parent)}/data/aoi_washington.geojson"
            )
        else:
            raise ValueError(
                "Please select one of 'Berlin' or 'Washington' as the location!"
            )

        if as_dataframe:
            df = GeoDataFrame.from_features(example_aoi, crs=4326)
            return df
        else:
            return example_aoi

    @staticmethod
    def draw_aoi():
        """
        Displays an interactive map to draw an aoi by hand, returns the folium object if
        not run in a Jupyter notebook.

        Export the drawn aoi via the export button, then read the geometries via
        read_aoi_file().
        """
        m = folium_base_map(layer_control=True)
        DrawFoliumOverride(
            export=True,
            filename="aoi.geojson",
            position="topleft",
            draw_options={
                "rectangle": {"repeatMode": False, "showArea": True},
                "polygon": {"showArea": True, "allowIntersection": False},
                "polyline": False,
                "circle": False,
                "marker": False,
                "circlemarker": False,
            },
            edit_options={"polygon": {"allowIntersection": False}},
        ).add_to(m)
        return m

    @check_auth
    def get_blocks(
        self,
        block_type: Optional[str] = None,
        basic: bool = True,
        as_dataframe: bool = False,
    ) -> Union[List[Dict], dict]:
        """
        Gets a list of all public blocks on the marketplace. Can not access custom blocks.

        Args:
            block_type: Optionally filters to "data" or "processing" blocks, default None.
            basic: Optionally returns simple version {block_id : block_name}
            as_dataframe: Returns a dataframe instead of json (default).

        Returns:
            A list of the public blocks and their metadata. Optional a simpler version
            dict.
        """
        try:
            block_type = block_type.lower()  # type: ignore
        except AttributeError:
            pass
        url = f"{self.auth._endpoint()}/blocks"
        response_json = self.auth._request(request_type="GET", url=url)
        public_blocks_json = response_json["data"]

        if block_type == "data":
            logger.info("Getting only data blocks.")
            blocks_json = [
                block for block in public_blocks_json if block["type"] == "DATA"
            ]
        elif block_type == "processing":
            logger.info("Getting only processing blocks.")
            blocks_json = [
                block for block in public_blocks_json if block["type"] == "PROCESSING"
            ]
        else:
            blocks_json = public_blocks_json

        if basic:
            logger.info(
                "Getting blocks name and id, use basic=False for all block details."
            )
            blocks_basic = {block["name"]: block["id"] for block in blocks_json}
            if as_dataframe:
                return pd.DataFrame.from_dict(blocks_basic, orient="index")
            else:
                return blocks_basic

        else:
            if as_dataframe:
                return pd.DataFrame(blocks_json)
            else:
                return blocks_json

    @check_auth
    def get_block_details(self, block_id: str, as_dataframe: bool = False) -> dict:
        """
        Gets the detailed information about a specific public block from
        the server, includes all manifest.json and marketplace.json contents.
        Can not access custom blocks.

        Args:
            block_id: The block id.
            as_dataframe: Returns a dataframe instead of json (default).

        Returns:
            A dict of the block details metadata for the specific block.
        """
        url = f"{self.auth._endpoint()}/blocks/{block_id}"  # public blocks
        response_json = self.auth._request(request_type="GET", url=url)
        details_json = response_json["data"]

        if as_dataframe:
            return pd.DataFrame.from_dict(details_json, orient="index").transpose()
        else:
            return details_json

    @check_auth
    def get_block_coverage(self, block_id: str) -> dict:
        # pylint: disable=unused-argument
        """
        Gets the spatial coverage of a data/processing block as
        url or GeoJson Feature Collection.

        Args:
            block_id: The block id.

        Returns:
            A dict of the spatial coverage for the specific block.
        """
        url = f"{self.auth._endpoint()}/blocks/{block_id}/coverage"
        response_json = self.auth._request(request_type="GET", url=url)
        details_json = response_json["data"]
        response_coverage = requests.get(details_json["url"]).json()
        return response_coverage

    @check_auth
    def get_credits_balance(self) -> dict:
        """
        Display the overall credits available in your account.

        Returns:
            A dict with the balance of credits available in your account.
        """
        endpoint_url = f"{self.auth._endpoint()}/accounts/me/credits/balance"
        response_json = self.auth._request(request_type="GET", url=endpoint_url)
        details_json = response_json["data"]
        return details_json

    @check_auth
    def get_credits_history(
        self,
        start_date: Optional[Union[str, datetime]] = None,
        end_date: Optional[Union[str, datetime]] = None,
    ) -> Dict[str, Union[str, int, Dict]]:
        """
        Display the overall credits history consumed in your account.
        The consumption history will be returned for all workspace_ids on your
        account.
        Args:
            start_date: The start date for the credit consumption search e.g.
                2021-12-01. Default start_date None uses 2000-01-01.
            end_date: The end date for the credit consumption search e.g.
                2021-12-31. Default end_date None uses current date.

        Returns:
            A dict with the information of the credit consumption records for
            all the users linked by the account_id.
            (see https://docs.up42.com/developers/api#operation/getHistory for
            output description)
        """
        if start_date is None:
            start_date = "2000-01-01"
        if end_date is None:
            tomorrow_date = date.today() + timedelta(days=1)
            tomorrow_datetime = datetime(
                year=tomorrow_date.year,
                month=tomorrow_date.month,
                day=tomorrow_date.day,
            )
            end_date = tomorrow_datetime.strftime("%Y-%m-%d")

        [start_formatted_date, end_formatted_date] = format_time_period(
            start_date=start_date, end_date=end_date
        ).split("/")
        search_parameters = dict(
            {
                "from": start_formatted_date,
                "to": end_formatted_date,
                "size": 2000,  # 2000 is the maximum page size for this call
                "page": 0,
            }
        )
        endpoint_url = f"{self.auth._endpoint()}/accounts/me/credits/history"
        response_json: dict = self.auth._request(
            request_type="GET", url=endpoint_url, querystring=search_parameters
        )
        isLastPage = response_json["data"]["last"]
        credit_history = response_json["data"]["content"].copy()
        result = dict(response_json["data"])
        del result["content"]
        while not isLastPage:
            search_parameters["page"] += 1
            response_json = self.auth._request(
                request_type="GET", url=endpoint_url, querystring=search_parameters
            )
            isLastPage = response_json["data"]["last"]
            credit_history.extend(response_json["data"]["content"].copy())
        result["content"] = credit_history
        return result

    @check_auth
    def validate_manifest(self, path_or_json: Union[str, Path, dict]) -> dict:
        """
        Validates a block manifest json.

        The block manifest is required to build a custom block on UP42 and contains
        the metadata about the block as well as block input and output capabilities.
        Also see the
        [manifest chapter in the UP42 documentation](https://docs.up42.com/reference/block-manifest.html).

        Args:
            path_or_json: The input manifest, either a filepath or json string, see example.

        Returns:
            A dictionary with the validation results and potential validation errors.
        """
        if isinstance(path_or_json, (str, Path)):
            with open(path_or_json) as src:
                manifest_json = json.load(src)
        else:
            manifest_json = path_or_json
        url = f"{self.auth._endpoint()}/validate-schema/block"
        response_json = self.auth._request(
            request_type="POST", url=url, data=manifest_json
        )
        logger.info("The manifest is valid.")
        return response_json["data"]

read_vector_file(filename='aoi.geojson', as_dataframe=False)

Reads vector files (geojson, shapefile, kml, wkt) to a feature collection, for use as the aoi geometry in the workflow input parameters (see get_input_parameters).

Example aoi fiels are provided, e.g. example/data/aoi_Berlin.geojson

Parameters:

Name Type Description Default
filename str

File path of the vector file.

'aoi.geojson'
as_dataframe bool

Return type, default FeatureCollection, GeoDataFrame if True.

False

Returns:

Type Description
Union[dict, GeoDataFrame]

Feature Collection

Source code in up42/tools.py
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
def read_vector_file(
    self, filename: str = "aoi.geojson", as_dataframe: bool = False
) -> Union[dict, GeoDataFrame]:
    """
    Reads vector files (geojson, shapefile, kml, wkt) to a feature collection,
    for use as the aoi geometry in the workflow input parameters
    (see get_input_parameters).

    Example aoi fiels are provided, e.g. example/data/aoi_Berlin.geojson

    Args:
        filename: File path of the vector file.
        as_dataframe: Return type, default FeatureCollection, GeoDataFrame if True.

    Returns:
        Feature Collection
    """
    suffix = Path(filename).suffix

    if suffix == ".kml":
        gpd.io.file.fiona.drvsupport.supported_drivers["KML"] = "rw"
        df = gpd.read_file(filename, driver="KML")
    elif suffix == ".wkt":
        with open(filename) as wkt_file:
            wkt = wkt_file.read()
            df = pd.DataFrame({"geometry": [wkt]})
            df["geometry"] = df["geometry"].apply(shapely.wkt.loads)
            df = GeoDataFrame(df, geometry="geometry", crs=4326)
    else:
        df = gpd.read_file(filename)

    if df.crs.to_string() != "EPSG:4326":
        df = df.to_crs(epsg=4326)
    if as_dataframe:
        return df
    else:
        return df.__geo_interface__

get_example_aoi(location='Berlin', as_dataframe=False)

Gets predefined, small, rectangular example aoi for the selected location.

Parameters:

Name Type Description Default
location str

Location, one of Berlin, Washington.

'Berlin'
as_dataframe bool

Returns a dataframe instead of dict FeatureColletions (default).

False

Returns:

Type Description
Union[dict, GeoDataFrame]

Feature collection json with the selected aoi.

Source code in up42/tools.py
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
def get_example_aoi(
    self, location: str = "Berlin", as_dataframe: bool = False
) -> Union[dict, GeoDataFrame]:
    """
    Gets predefined, small, rectangular example aoi for the selected location.

    Args:
        location: Location, one of Berlin, Washington.
        as_dataframe: Returns a dataframe instead of dict FeatureColletions
            (default).

    Returns:
        Feature collection json with the selected aoi.
    """
    logger.info(f"Getting small example aoi in location '{location}'.")
    if location == "Berlin":
        example_aoi = self.read_vector_file(
            f"{str(Path(__file__).resolve().parent)}/data/aoi_berlin.geojson"
        )
    elif location == "Washington":
        example_aoi = self.read_vector_file(
            f"{str(Path(__file__).resolve().parent)}/data/aoi_washington.geojson"
        )
    else:
        raise ValueError(
            "Please select one of 'Berlin' or 'Washington' as the location!"
        )

    if as_dataframe:
        df = GeoDataFrame.from_features(example_aoi, crs=4326)
        return df
    else:
        return example_aoi

draw_aoi() staticmethod

Displays an interactive map to draw an aoi by hand, returns the folium object if not run in a Jupyter notebook.

Export the drawn aoi via the export button, then read the geometries via read_aoi_file().

Source code in up42/tools.py
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@staticmethod
def draw_aoi():
    """
    Displays an interactive map to draw an aoi by hand, returns the folium object if
    not run in a Jupyter notebook.

    Export the drawn aoi via the export button, then read the geometries via
    read_aoi_file().
    """
    m = folium_base_map(layer_control=True)
    DrawFoliumOverride(
        export=True,
        filename="aoi.geojson",
        position="topleft",
        draw_options={
            "rectangle": {"repeatMode": False, "showArea": True},
            "polygon": {"showArea": True, "allowIntersection": False},
            "polyline": False,
            "circle": False,
            "marker": False,
            "circlemarker": False,
        },
        edit_options={"polygon": {"allowIntersection": False}},
    ).add_to(m)
    return m

get_blocks(block_type=None, basic=True, as_dataframe=False)

Gets a list of all public blocks on the marketplace. Can not access custom blocks.

Parameters:

Name Type Description Default
block_type Optional[str]

Optionally filters to "data" or "processing" blocks, default None.

None
basic bool

Optionally returns simple version {block_id : block_name}

True
as_dataframe bool

Returns a dataframe instead of json (default).

False

Returns:

Type Description
Union[List[Dict], dict]

A list of the public blocks and their metadata. Optional a simpler version

Union[List[Dict], dict]

dict.

Source code in up42/tools.py
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@check_auth
def get_blocks(
    self,
    block_type: Optional[str] = None,
    basic: bool = True,
    as_dataframe: bool = False,
) -> Union[List[Dict], dict]:
    """
    Gets a list of all public blocks on the marketplace. Can not access custom blocks.

    Args:
        block_type: Optionally filters to "data" or "processing" blocks, default None.
        basic: Optionally returns simple version {block_id : block_name}
        as_dataframe: Returns a dataframe instead of json (default).

    Returns:
        A list of the public blocks and their metadata. Optional a simpler version
        dict.
    """
    try:
        block_type = block_type.lower()  # type: ignore
    except AttributeError:
        pass
    url = f"{self.auth._endpoint()}/blocks"
    response_json = self.auth._request(request_type="GET", url=url)
    public_blocks_json = response_json["data"]

    if block_type == "data":
        logger.info("Getting only data blocks.")
        blocks_json = [
            block for block in public_blocks_json if block["type"] == "DATA"
        ]
    elif block_type == "processing":
        logger.info("Getting only processing blocks.")
        blocks_json = [
            block for block in public_blocks_json if block["type"] == "PROCESSING"
        ]
    else:
        blocks_json = public_blocks_json

    if basic:
        logger.info(
            "Getting blocks name and id, use basic=False for all block details."
        )
        blocks_basic = {block["name"]: block["id"] for block in blocks_json}
        if as_dataframe:
            return pd.DataFrame.from_dict(blocks_basic, orient="index")
        else:
            return blocks_basic

    else:
        if as_dataframe:
            return pd.DataFrame(blocks_json)
        else:
            return blocks_json

get_block_details(block_id, as_dataframe=False)

Gets the detailed information about a specific public block from the server, includes all manifest.json and marketplace.json contents. Can not access custom blocks.

Parameters:

Name Type Description Default
block_id str

The block id.

required
as_dataframe bool

Returns a dataframe instead of json (default).

False

Returns:

Type Description
dict

A dict of the block details metadata for the specific block.

Source code in up42/tools.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@check_auth
def get_block_details(self, block_id: str, as_dataframe: bool = False) -> dict:
    """
    Gets the detailed information about a specific public block from
    the server, includes all manifest.json and marketplace.json contents.
    Can not access custom blocks.

    Args:
        block_id: The block id.
        as_dataframe: Returns a dataframe instead of json (default).

    Returns:
        A dict of the block details metadata for the specific block.
    """
    url = f"{self.auth._endpoint()}/blocks/{block_id}"  # public blocks
    response_json = self.auth._request(request_type="GET", url=url)
    details_json = response_json["data"]

    if as_dataframe:
        return pd.DataFrame.from_dict(details_json, orient="index").transpose()
    else:
        return details_json

get_block_coverage(block_id)

Gets the spatial coverage of a data/processing block as url or GeoJson Feature Collection.

Parameters:

Name Type Description Default
block_id str

The block id.

required

Returns:

Type Description
dict

A dict of the spatial coverage for the specific block.

Source code in up42/tools.py
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@check_auth
def get_block_coverage(self, block_id: str) -> dict:
    # pylint: disable=unused-argument
    """
    Gets the spatial coverage of a data/processing block as
    url or GeoJson Feature Collection.

    Args:
        block_id: The block id.

    Returns:
        A dict of the spatial coverage for the specific block.
    """
    url = f"{self.auth._endpoint()}/blocks/{block_id}/coverage"
    response_json = self.auth._request(request_type="GET", url=url)
    details_json = response_json["data"]
    response_coverage = requests.get(details_json["url"]).json()
    return response_coverage

get_credits_balance()

Display the overall credits available in your account.

Returns:

Type Description
dict

A dict with the balance of credits available in your account.

Source code in up42/tools.py
258
259
260
261
262
263
264
265
266
267
268
269
@check_auth
def get_credits_balance(self) -> dict:
    """
    Display the overall credits available in your account.

    Returns:
        A dict with the balance of credits available in your account.
    """
    endpoint_url = f"{self.auth._endpoint()}/accounts/me/credits/balance"
    response_json = self.auth._request(request_type="GET", url=endpoint_url)
    details_json = response_json["data"]
    return details_json

get_credits_history(start_date=None, end_date=None)

Display the overall credits history consumed in your account. The consumption history will be returned for all workspace_ids on your account.

Parameters:

Name Type Description Default
start_date Optional[Union[str, datetime]]

The start date for the credit consumption search e.g. 2021-12-01. Default start_date None uses 2000-01-01.

None
end_date Optional[Union[str, datetime]]

The end date for the credit consumption search e.g. 2021-12-31. Default end_date None uses current date.

None

Returns:

Type Description
Dict[str, Union[str, int, Dict]]

A dict with the information of the credit consumption records for

Dict[str, Union[str, int, Dict]]

all the users linked by the account_id.

Dict[str, Union[str, int, Dict]]

(see https://docs.up42.com/developers/api#operation/getHistory for

Dict[str, Union[str, int, Dict]]

output description)

Source code in up42/tools.py
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
@check_auth
def get_credits_history(
    self,
    start_date: Optional[Union[str, datetime]] = None,
    end_date: Optional[Union[str, datetime]] = None,
) -> Dict[str, Union[str, int, Dict]]:
    """
    Display the overall credits history consumed in your account.
    The consumption history will be returned for all workspace_ids on your
    account.
    Args:
        start_date: The start date for the credit consumption search e.g.
            2021-12-01. Default start_date None uses 2000-01-01.
        end_date: The end date for the credit consumption search e.g.
            2021-12-31. Default end_date None uses current date.

    Returns:
        A dict with the information of the credit consumption records for
        all the users linked by the account_id.
        (see https://docs.up42.com/developers/api#operation/getHistory for
        output description)
    """
    if start_date is None:
        start_date = "2000-01-01"
    if end_date is None:
        tomorrow_date = date.today() + timedelta(days=1)
        tomorrow_datetime = datetime(
            year=tomorrow_date.year,
            month=tomorrow_date.month,
            day=tomorrow_date.day,
        )
        end_date = tomorrow_datetime.strftime("%Y-%m-%d")

    [start_formatted_date, end_formatted_date] = format_time_period(
        start_date=start_date, end_date=end_date
    ).split("/")
    search_parameters = dict(
        {
            "from": start_formatted_date,
            "to": end_formatted_date,
            "size": 2000,  # 2000 is the maximum page size for this call
            "page": 0,
        }
    )
    endpoint_url = f"{self.auth._endpoint()}/accounts/me/credits/history"
    response_json: dict = self.auth._request(
        request_type="GET", url=endpoint_url, querystring=search_parameters
    )
    isLastPage = response_json["data"]["last"]
    credit_history = response_json["data"]["content"].copy()
    result = dict(response_json["data"])
    del result["content"]
    while not isLastPage:
        search_parameters["page"] += 1
        response_json = self.auth._request(
            request_type="GET", url=endpoint_url, querystring=search_parameters
        )
        isLastPage = response_json["data"]["last"]
        credit_history.extend(response_json["data"]["content"].copy())
    result["content"] = credit_history
    return result

validate_manifest(path_or_json)

Validates a block manifest json.

The block manifest is required to build a custom block on UP42 and contains the metadata about the block as well as block input and output capabilities. Also see the manifest chapter in the UP42 documentation.

Parameters:

Name Type Description Default
path_or_json Union[str, Path, dict]

The input manifest, either a filepath or json string, see example.

required

Returns:

Type Description
dict

A dictionary with the validation results and potential validation errors.

Source code in up42/tools.py
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
@check_auth
def validate_manifest(self, path_or_json: Union[str, Path, dict]) -> dict:
    """
    Validates a block manifest json.

    The block manifest is required to build a custom block on UP42 and contains
    the metadata about the block as well as block input and output capabilities.
    Also see the
    [manifest chapter in the UP42 documentation](https://docs.up42.com/reference/block-manifest.html).

    Args:
        path_or_json: The input manifest, either a filepath or json string, see example.

    Returns:
        A dictionary with the validation results and potential validation errors.
    """
    if isinstance(path_or_json, (str, Path)):
        with open(path_or_json) as src:
            manifest_json = json.load(src)
    else:
        manifest_json = path_or_json
    url = f"{self.auth._endpoint()}/validate-schema/block"
    response_json = self.auth._request(
        request_type="POST", url=url, data=manifest_json
    )
    logger.info("The manifest is valid.")
    return response_json["data"]