Skip to content

up42

up42 object

up42 is the base object imported to Python. It provides the elementary functionality that is not bound to a specific object of the UP42 structure (Project > Workflow > Job etc.).

From it you can initialize these other 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


Functions

initialize_catalog()

Returns a Catalog object for using the catalog search.

Source code in up42/__init__.py
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_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
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

The UP42 jobtask_id

required
job_id

The UP42 job_id

required
Source code in up42/__init__.py
def initialize_jobtask(jobtask_id, job_id) -> "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_project()

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

Source code in up42/__init__.py
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_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
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

settings(log=True)

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

Parameters:

Name Type Description Default
log

Activates/deactivates logging, default True is activated logging.

True
Source code in up42/__init__.py
def settings(log=True):
    """
    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.")

    # pylint: disable=expression-not-assigned
    [
        setattr(logging.getLogger(name), "disabled", not log)
        for name in logging.root.manager.loggerDict
    ]

draw_aoi(self)

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
def draw_aoi(self):
    """
    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().
    """
    self._deprecate_tools("draw_aoi")
    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_block_details(self, 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.

Parameters:

Name Type Description Default
block_id str

The block id.

required
as_dataframe

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
def get_block_details(self, block_id: str, as_dataframe=False) -> Dict:
    """
    Gets the detailed information about a specific public block from
    the server, includes all manifest.json and marketplace.json contents.

    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.
    """
    self._deprecate_tools("get_block_details")

    if not hasattr(self, "auth"):
        raise Exception(
            "Requires authentication with UP42, use up42.authenticate()!"
        )
    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_blocks(self, block_type=None, basic=True, as_dataframe=False)

Gets a list of all public blocks on the marketplace.

Parameters:

Name Type Description Default
block_type

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

None
basic bool

Optionally returns simple version {block_id : block_name}

True
as_dataframe

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 dict.

Source code in up42/tools.py
def get_blocks(
    self,
    block_type=None,
    basic: bool = True,
    as_dataframe=False,
) -> Union[List[Dict], Dict]:
    """
    Gets a list of all public blocks on the marketplace.

    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.
    """
    self._deprecate_tools("get_blocks")

    try:
        block_type = block_type.lower()
    except AttributeError:
        pass
    if not hasattr(self, "auth"):
        raise Exception(
            "Requires authentication with UP42, use up42.authenticate()!"
        )
    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_example_aoi(self, 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, geopandas.geodataframe.GeoDataFrame]

Feature collection json with the selected aoi.

Source code in up42/tools.py
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.
    """
    self._deprecate_tools("get_example_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

read_vector_file(self, 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, geopandas.geodataframe.GeoDataFrame]

Feature Collection

Source code in up42/tools.py
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
    """
    self._deprecate_tools("read_vector_file")

    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)
    # TODO: Explode multipolygons (if neccessary as union in aoi anyway most often).
    # TODO: Have both bboxes for each feature and overall?
    if as_dataframe:
        return df
    else:
        return df.__geo_interface__

validate_manifest(self, 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, pathlib.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
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.
    """
    self._deprecate_tools("validate_manifest")

    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
    if not hasattr(self, "auth"):
        raise Exception(
            "Requires authentication with UP42, use up42.authenticate()!"
        )
    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"]