Skip to content

up42

authenticate(cfg_file=None, project_id=None, project_api_key=None, **kwargs)

Authenticate with UP42, either via project_id & project_api_key, or a config json file containing both. Also see the documentation https://sdk.up42.com/authentication/

Parameters:

Name Type Description Default
cfg_file Union[str, Path]

A json file containing project_id & project_api_key

None
project_id Optional[str]

The UP42 project id.

None
project_api_key Optional[str]

The UP42 project api key.

None

Examples:

up42.authenticate(
    project_id="your-project-ID",
    project_api_key="your-project-API-key"
)
Source code in up42/main.py
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
def authenticate(
    cfg_file: Union[str, Path] = None,
    project_id: Optional[str] = None,
    project_api_key: Optional[str] = None,
    **kwargs,
):
    """
    Authenticate with UP42, either via project_id & project_api_key, or a config json file containing both.
    Also see the documentation https://sdk.up42.com/authentication/

    Args:
        cfg_file: A json file containing project_id & project_api_key
        project_id: The UP42 project id.
        project_api_key: The UP42 project api key.

    Examples:
        ```python
        up42.authenticate(
            project_id="your-project-ID",
            project_api_key="your-project-API-key"
        )
        ```
    """
    global _auth
    _auth = Auth(
        cfg_file=cfg_file,
        project_id=project_id,
        project_api_key=project_api_key,
        **kwargs,
    )

SDK conveniance functionality that is made available on the up42 import object (in the init) and is not directly related to API calls.

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
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
def get_example_aoi(
    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 = read_vector_file(
            f"{str(Path(__file__).resolve().parent)}/data/aoi_berlin.geojson"
        )
    elif location == "Washington":
        example_aoi = 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

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
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
def read_vector_file(
    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":
        # pylint: disable=no-member
        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__

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/tools.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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)

Visualization tools available in various objects

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 up42.read_aoi_file().

Requires installation of up42-py[viz] extra dependencies.

Source code in up42/viztools.py
 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
@requires_viz
def draw_aoi() -> "folium.Map":
    """
    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
    up42.read_aoi_file().

    Requires installation of up42-py[viz] extra dependencies.
    """
    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

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/main.py
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@_check_auth
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_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/main.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@_check_auth
def get_block_coverage(block_id: str) -> dict:
    """
    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"{_auth._endpoint()}/blocks/{block_id}/coverage"
    response_json = _auth._request(request_type="GET", url=url)
    details_json = response_json["data"]
    response_coverage = requests.get(details_json["url"]).json()
    return response_coverage

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/main.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@_check_auth
def get_block_details(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"{_auth._endpoint()}/blocks/{block_id}"  # public blocks
    response_json = _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_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/main.py
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
@_check_auth
def get_blocks(
    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"{_auth._endpoint()}/blocks"
    response_json = _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_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/main.py
237
238
239
240
241
242
243
244
245
246
247
248
@_check_auth
def get_credits_balance() -> dict:
    """
    Display the overall credits available in your account.

    Returns:
        A dict with the balance of credits available in your account.
    """
    endpoint_url = f"{_auth._endpoint()}/accounts/me/credits/balance"
    response_json = _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, datetime or isoformat string 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, datetime or isoformat string 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 all the users linked by the account_id.

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

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

Source code in up42/main.py
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
@_check_auth
def get_credits_history(
    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, datetime or isoformat string e.g.
            2021-12-01. Default start_date None uses 2000-01-01.
        end_date: The end date for the credit consumption search, datetime or isoformat string 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")

    search_parameters = dict(
        {
            "from": format_time(start_date),
            "to": format_time(end_date, set_end_of_day=True),
            "size": 2000,  # 2000 is the maximum page size for this call
            "page": 0,
        }
    )
    endpoint_url = f"{_auth._endpoint()}/accounts/me/credits/history"
    response_json: dict = _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 = _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

get_webhook_events()

Gets all available webhook events.

Returns:

Type Description
dict

A dict of the available webhook events.

Source code in up42/main.py
128
129
130
131
132
133
134
135
136
137
@_check_auth
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

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/main.py
88
89
90
91
92
93
94
95
96
97
98
99
@_check_auth
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

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/main.py
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
@_check_auth
def validate_manifest(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"{_auth._endpoint()}/validate-schema/block"
    response_json = _auth._request(request_type="POST", url=url, data=manifest_json)
    logger.info("The manifest is valid.")
    return response_json["data"]

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/initialization.py
145
146
147
148
149
150
151
152
153
154
@_check_auth
def initialize_asset(asset_id: str) -> "Asset":
    """
    Returns an Asset object (has to exist on UP42).
    Args:
        asset_id: The UP42 asset_id
    """
    asset = Asset(auth=main._auth, asset_id=asset_id)
    logger.info(f"Initialized {asset}")
    return asset

initialize_catalog()

Returns a Catalog object for using the catalog search.

Source code in up42/initialization.py
47
48
49
50
51
52
@_check_auth
def initialize_catalog() -> "Catalog":
    """
    Returns a Catalog object for using the catalog search.
    """
    return Catalog(auth=main._auth)

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/initialization.py
77
78
79
80
81
82
83
84
85
86
@_check_auth
def initialize_job(job_id: str) -> "Job":
    """
    Returns a Job object (has to exist on UP42).
    Args:
        job_id: The UP42 job_id
    """
    job = Job(auth=main._auth, job_id=job_id, project_id=str(main._auth.project_id))
    logger.info(f"Initialized {job}")
    return job

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/initialization.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@_check_auth
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
    """
    jobs = [
        Job(auth=main._auth, job_id=job_id, project_id=str(main._auth.project_id))
        for job_id in job_ids
    ]
    jobcollection = JobCollection(
        auth=main._auth, project_id=str(main._auth.project_id), jobs=jobs
    )
    logger.info(f"Initialized {jobcollection}")
    return jobcollection

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/initialization.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@_check_auth
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
    """
    jobtask = JobTask(
        auth=main._auth,
        jobtask_id=jobtask_id,
        job_id=job_id,
        project_id=str(main._auth.project_id),
    )
    logger.info(f"Initialized {jobtask}")
    return jobtask

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/initialization.py
133
134
135
136
137
138
139
140
141
142
@_check_auth
def initialize_order(order_id: str) -> "Order":
    """
    Returns an Order object (has to exist on UP42).
    Args:
        order_id: The UP42 order_id
    """
    order = Order(auth=main._auth, order_id=order_id)
    logger.info(f"Initialized {order}")
    return order

initialize_project()

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

Source code in up42/initialization.py
37
38
39
40
41
42
43
44
@_check_auth
def initialize_project() -> "Project":
    """
    Returns the correct Project object (has to exist on UP42).
    """
    project = Project(auth=main._auth, project_id=str(main._auth.project_id))
    logger.info(f"Initialized {project}")
    return project

initialize_storage()

Returns a Storage object to list orders and assets.

Source code in up42/initialization.py
125
126
127
128
129
130
@_check_auth
def initialize_storage() -> "Storage":
    """
    Returns a Storage object to list orders and assets.
    """
    return Storage(auth=main._auth)

initialize_tasking()

Returns a Tasking object for creating satellite tasking orders.

Source code in up42/initialization.py
55
56
57
58
59
60
@_check_auth
def initialize_tasking() -> "Tasking":
    """
    Returns a Tasking object for creating satellite tasking orders.
    """
    return Tasking(auth=main._auth)

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/initialization.py
157
158
159
160
161
162
163
164
165
166
@_check_auth
def initialize_webhook(webhook_id: str) -> Webhook:
    """
    Returns a Webhook object (has to exist on UP42).
    Args:
        webhook_id: The UP42 webhook_id
    """
    webhook = Webhook(auth=main._auth, webhook_id=webhook_id)
    logger.info(f"Initialized {webhook}")
    return webhook

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/initialization.py
63
64
65
66
67
68
69
70
71
72
73
74
@_check_auth
def initialize_workflow(workflow_id: str) -> "Workflow":
    """
    Returns a Workflow object (has to exist on UP42).
    Args:
        workflow_id: The UP42 workflow_id
    """
    workflow = Workflow(
        auth=main._auth, workflow_id=workflow_id, project_id=str(main._auth.project_id)
    )
    logger.info(f"Initialized {workflow}")
    return workflow