Copyright (C) 2020-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
This module contains a :class:DataInteractiveApi class to perform API operations.

The HERE API reference documentation used in this module can be found here:
|interactive_api_reference|

.. |interactive_api_reference| raw:: html

Interactive API Reference
""" # noqa E501

import urllib.parse
from typing import Any, Dict, List, Optional, Union

from here.platform.api.base_api import BaseApi
from here.platform.auth import Auth
from here.platform.config.application_config import ApplicationConfig
from here.platform.config.platform_config import PlatformConfig
from here.platform.exceptions import (
PayloadTooLargeException,
PlatformException,
RequestEntityTooLargeException,
)
from requests import Response

[docs]
class DataInteractiveApi(BaseApi):
"""
This class provides access to HERE platform Data Interactive APIs.

Interactive APIs offer a set of unique capabilities, enabling you to store, retrieve,
search for, analyze and modify data at a feature (e.g., a place) and feature property
(e.g., the name of the place) level. With interactive APIs, data is stored in GeoJSON
and can be retrieved dynamically at any zoom level.
"""

def init(
self,
base_url: str,
platform_config: PlatformConfig,
application_config: ApplicationConfig,
auth: Optional[Auth],
proxies: Optional[dict] = None,
):
"""
Instantiate DataInteractiveApi object.

:param base_url: a base url of the Interactive API.
:param platform_config: a mandatory :class:PlatformConfig object to provide
configuration information for the API.
:param application_config: a mandatory :class:ApplicationConfig object to provide
configuration information for the API.
:param auth: an Authentication instance.
:param proxies: an optional proxy configuration. Defaults to the environment proxy
configuration.
"""
super().init(
platform_config=platform_config,
application_config=application_config,
auth=auth,
proxies=proxies,
)
self.base_url = base_url

@staticmethod
def _raise_response_exception(resp: Response) -> None:
"""
Parse HTTP errors status code and raise necessary exceptions.

:param resp: An HTTP response to parse.
:raises RequestEntityTooLargeException: If platform responds with HTTP 413.
:raises PayloadTooLargeException: If platform responds with HTTP 513.
:raises PlatformException: If client responds with any other exception.
"""
if resp.status_code == 413:
raise RequestEntityTooLargeException(resp)
elif resp.status_code == 513:
raise PayloadTooLargeException(resp)
else:
raise PlatformException(resp)

@staticmethod
def _query_params_to_string(params: Dict[str, Union[str, list, tuple]]) -> str:
"""
Convert query params in a dictionary to string.

This method is required as character encoding needs to be skipped for ,
to support OR condition, and if property value has any special character then
character encoding is required. As python :module:requests by default does encoding
of all the characters hence, this needs to be handled separately.
For more details please check: https://saeljira.it.here.com/browse/DH-1369.

:param params: A dict to represent query params.
:return: A string.
"""
qlist = []
qval = ""

def param_to_string(v) -> str:
return (
str(v) if isinstance(v, (int, float)) else ("'" + urllib.parse.quote_plus(v) + "'")
)

for key, val in params.items():
if isinstance(val, (list, tuple)):
qval = ",".join(map(param_to_string, val))
else:
qval = param_to_string(val)
qlist.append(f"{key}={qval}")

params_str = "&".join(qlist)
return params_str

[docs]
def get_feature(
self,
layer_id: str,
feature_id: str,
selection: Optional[List[str]] = None,
force2d: bool = False,
) -> dict:
"""
Return the feature with the provided feature_id.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_id: Feature id which is to fetched.
:param selection: A list, only these properties will be present in returned feature.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features/{feature_id}"
params: Dict[str, Union[List, str]] = {"force2D": str(force2d).lower()}
if selection:
params["selection"] = [f"p.{name}" for name in selection]

url = self.format_url(self.base_url, path)
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_features(
self,
layer_id: str,
feature_ids: List,
selection: Optional[List[str]] = None,
force2d: bool = False,
) -> dict:
"""
Return all of the features found for the provided list of feature ids.

The response is always a FeatureCollection, even if there are no features
with the provided ids.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_ids: A list of feature_ids to fetch.
:param selection: A list, only these properties will be present in returned features.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features"
params = {"id": feature_ids, "force2D": str(force2d).lower()}
if selection:
params["selection"] = [f"p.{name}" for name in selection]
url = self.format_url(self.base_url, path)
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_statistics(self, layer_id: str, skip_cache: bool = False) -> dict:
"""
Return statistical information about this layer.

:param layer_id: Identifier of the Interactive Map Layer.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/statistics"
params = {"skipCache": str(skip_cache).lower()}
url = self.format_url(self.base_url, path)
resp = self.get(url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_features_by_bbox(
self,
layer_id: str,
bbox: tuple,
clip: bool = False,
limit: int = 30000,
params: Optional[dict] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
clustering: Optional[str] = None,
clustering_params: Optional[Dict] = None,
force2d: bool = False,
) -> dict:
"""Return the features which are inside a bounding box stipulated by bbox parameter.

:param layer_id: Identifier of the Interactive Map Layer.
:param bbox: A tuple of four numbers representing the West, South,
East and North margins, respectively, of the bounding box.
:param clip: A Boolean indicating if the result should be clipped
(default: False).
:param limit: A maximum number of features to return in the result. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param clustering: The clustering algorithm to apply to the data within the result.
Clustering algorithms supported: hexbin, quadbin.
:param clustering_params: Parameters for the chosen clustering algorithm.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.

"""
path = f"/layers/{layer_id}/bbox"
url = self.format_url(self.base_url, path)
if params:
params_str = self._query_params_to_string(params)
url = "?".join((url, params_str))

bbox_str = ",".join([str(i) for i in bbox])
q_params = {"bbox": bbox_str, "force2D": str(force2d).lower(), "clip": str(clip).lower(), "limit": limit, "skipCache": str(skip_cache).lower(),}
if selection:
q_params["selection"] = [f"p.{name}" for name in selection]
if clustering is not None:
q_params["clustering"] = clustering
if clustering_params:
d = dict((f"clustering.{k}", v) for (k, v) in clustering_params.items())
q_params.update(d)

resp = self.get(url, params=q_params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_features_in_tile(
self,
layer_id: str,
tile_type: str,
tile_id: str,
clip: bool = False,
params: Optional[dict] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
clustering: Optional[str] = None,
clustering_params: Optional[Dict] = None,
margin: Optional[int] = 0,
limit: int = 30000,
force2d: bool = False,
) -> dict:
"""
Return features in tile.

:param layer_id: Identifier of the Interactive Map Layer.
:param tile_type: A string with the name of a tile type, one of
"quadkeys", "web", "tms" or "here". The type of tile identifier.
"quadkey" - Virtual Earth, "web" - Web Mercator,
"tms" - OSGEO Tile Map Service, "here" - Here Tile Schema.
:param tile_id: The tile identifier can be provided as quadkey (1),
Web Mercator level,x,y coordinates (1_1_0) or
OSGEO Tile Map Service level,x,y (1_1_0).
:param clip: A Boolean indicating if the result should be clipped
(default: False).
:param params: A dict to represent additional filters on features to be searched.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param clustering: The clustering algorithm to apply to the data within the result.
Clustering algorithms supported: hexbin, quadbin.
:param clustering_params: Parameters for the chosen clustering algorithm.
:param margin: Margin in pixels on the respective projected level around the tile.
Default is 0.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.

"""
path = f"/layers/{layer_id}/tile/{tile_type}/{tile_id}"
url = self.format_url(self.base_url, path)
if params:
params_str = self._query_params_to_string(params)
url = "?".join((url, params_str))
q_params = {"force2D": str(force2d).lower(), "clip": str(clip).lower(), "limit": limit, "skipCache": str(skip_cache).lower(), "margin": margin,}
if selection:
q_params["selection"] = [f"p.{name}" for name in selection]
if clustering is not None:
q_params["clustering"] = clustering
if clustering_params:
d = dict((f"clustering.{k}", v) for (k, v) in clustering_params.items())
q_params.update(d)

resp = self.get(url, params=q_params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_features_with_radius_search(
self,
layer_id: str,
lat: Optional[float] = None,
lng: Optional[float] = None,
ref_catalog: Optional[str] = None,
ref_layer_id: Optional[str] = None,
ref_feature_id: Optional[str] = None,
radius: Optional[int] = None,
limit: int = 30000,
params: Optional[dict] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force2d: bool = False,
) -> dict:
"""
Return the features which are inside the specified radius.

The origin radius point is calculated based either on latitude & longitude
or by specifying a feature's geometry.

:param layer_id: Identifier of the Interactive Map Layer.
:param lat: The latitude in WGS'84 decimal degree (-90 to +90) of the center Point.
:param lng: The longitude in WGS'84 decimal degree (-180 to +180) of the center Point.
:param ref_catalog: The catalog HRN where the layer containing the referenced feature
is stored. Always to use in combination with ref_feature_id.
:param ref_layer_id: As alternative for defining center coordinates, it is possible to
reference a geometry in a layer. Therefore it is needed to provide the layer
id where the referenced feature is stored. Always to use in combination with
ref_feature_id.
:param ref_feature_id: The unique identifier of a feature in the referenced layer.
The geometry of that feature gets used for the spatial query.
Always to use in combination with ref_catalog and ref_layer_id.
:param radius: Radius in meter which defines the diameter of the search request.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.

"""
path = f"/layers/{layer_id}/spatial"
url = self.format_url(self.base_url, path)
if params:
params_str = self._query_params_to_string(params)
url = "?".join((url, params_str))
q_params: Dict[str, Any] = {"force2D": str(force2d).lower(), "limit": limit, "skipCache": str(skip_cache).lower(),}
if lat is not None:
q_params["lat"] = str(lat)
if lng is not None:
q_params["lon"] = str(lng)
if ref_catalog is not None:
q_params["refCatalogHrn"] = ref_catalog
if ref_layer_id is not None:
q_params["refLayerId"] = ref_catalog
if ref_feature_id is not None:
q_params["refFeatureId"] = ref_feature_id
if radius is not None:
q_params["radius"] = str(radius)
if selection:
q_params["selection"] = [f"p.{name}" for name in selection]

resp = self.get(url, params=q_params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def get_features_with_geometry_intersection(
self,
layer_id: str,
data: dict,
radius: Optional[int] = None,
limit: int = 30000,
params: Optional[dict] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force2d: bool = False,
) -> dict:
"""
Retrieve the features which are inside the specified radius and geometry.

The origin point is calculated based on the geometry provided as payload.

:param layer_id: Identifier of the Interactive Map Layer.
:param data: Geometry which will be used in intersection.
:param radius: Radius in meter which defines the diameter of the search request.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.

"""
path = f"/layers/{layer_id}/spatial"
url = self.format_url(self.base_url, path)
if params:
params_str = self._query_params_to_string(params)
url = "?".join((url, params_str))
headers = self.headers
headers["Content-Type"] = "application/geo+json"
q_params: Dict[str, Any] = {"force2D": str(force2d).lower(), "limit": limit, "skipCache": str(skip_cache).lower(),}

if radius is not None:
q_params["radius"] = str(radius)
if selection:
q_params["selection"] = [f"p.{name}" for name in selection]
resp = self.post(url=url, params=q_params, data=data, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def search_features(
self,
layer_id: str,
limit: int = 30000,
params: Optional[dict] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force2d: bool = False,
) -> dict:
"""
Search for features in the layer.

The results are unordered and the request does not allow to continue the search,

:param layer_id: Identifier of the Interactive Map Layer.
:param limit: The maximum number of features in the response. Default is 30000.
Hard limit is 100000.
:param params: A dict to represent additional filters on features to be searched.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.

"""
path = f"/layers/{layer_id}/search"
url = self.format_url(self.base_url, path)
if params:
params_str = self._query_params_to_string(params)
url = "?".join((url, params_str))
q_params: Dict[str, Any] = {"force2D": str(force2d).lower(), "limit": limit, "skipCache": str(skip_cache).lower(),}

if selection:
q_params["selection"] = [f"p.{name}" for name in selection]
resp = self.get(url, params=q_params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def iter_features(
self,
layer_id: str,
limit: int = 30000,
page_token: Optional[str] = None,
selection: Optional[List[str]] = None,
skip_cache: bool = False,
force2d: bool = False,
) -> dict:
"""
Iterate over all of the features in the layer.

The features in the response are ordered so that no feature is returned twice.

:param layer_id: Identifier of the Interactive Map Layer.
:param limit: The maximum number of features in the response in single iteration.
Default is 30000. Hard limit is 100000.
:param page_token: The page token where the iteration will continue.
:param selection: A list, only these properties will be present in returned features.
:param skip_cache: If set to True the response is not returned from cache.
Default is False.
:param force2d: If set to True then features in the response will have only X and Y
components, else all x,y,z coordinates will be returned.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/iterate"
url = self.format_url(self.base_url, path)
params: Dict[str, Any] = {"limit": limit, "force2D": str(force2d).lower(), "skipCache": str(skip_cache).lower(),}
if selection:
params["selection"] = [f"p.{name}" for name in selection]
if page_token:
params["pageToken"] = page_token

resp = self.get(url=url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def write_features(self, layer_id: str, data: dict) -> dict:
"""
Write the provided features to Layer.

:param layer_id: Identifier of the Interactive Map Layer.
:param data: Request body representing FeatureCollection to create or replace.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features"
url = self.format_url(self.base_url, path)
headers = self.headers
headers["Content-Type"] = "application/geo+json"
resp = self.put(url=url, data=data, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def update_features(self, layer_id: str, data: dict) -> dict:
"""
Update features.

:param layer_id: Identifier of the Interactive Map Layer.
:param data: Request body representing FeatureCollection to create or update.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features"
url = self.format_url(self.base_url, path)
headers = self.headers
headers["Content-Type"] = "application/geo+json"
resp = self.post(url=url, data=data, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def delete_features(self, layer_id: str, feature_ids: List[str]) -> Union[dict, str]:
"""
Delete multiple features from the layer.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_ids: A list of feature_ids to be deleted.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features"
url = self.format_url(self.base_url, path)
params = {"id": feature_ids}
resp = self.delete(url=url, params=params)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
elif resp.status_code == 204:
return resp.text
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def write_feature(self, layer_id: str, feature_id: str, data: dict) -> dict:
"""
Create or replace a feature in the layer.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_id: Feature id which is to fetched.
:param data: Request body representing feature to create or replace.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features/{feature_id}"
url = self.format_url(self.base_url, path)
headers = self.headers
headers["Content-Type"] = "application/geo+json"
resp = self.put(url=url, data=data, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def update_feature(self, layer_id: str, feature_id: str, data: dict) -> dict:
"""
Update an existing feature.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_id: Feature id which is to fetched.
:param data: Request body representing feature to change.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features/{feature_id}"
url = self.format_url(self.base_url, path)
headers = self.headers
headers["Content-Type"] = "application/geo+json"
resp = self.patch(url=url, data=data, headers=headers)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.

[docs]
def delete_feature(self, layer_id: str, feature_id: str) -> Union[dict, str]:
"""
Delete an existing feature.

:param layer_id: Identifier of the Interactive Map Layer.
:param feature_id: Feature id which is to fetched.
:return: Response from the API.
"""
path = f"/layers/{layer_id}/features/{feature_id}"
url = self.format_url(self.base_url, path)
resp = self.delete(url=url)
if resp.status_code == 200:
resp_json: dict = resp.json()
return resp_json
elif resp.status_code == 204:
return resp.text
else:
self._raise_response_exception(resp)
return # Type checker doesn't see that above function always raises.