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

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

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

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

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

@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

@_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

@_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

@_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

@_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

@_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

@_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

@_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

@_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

@_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"]