# Copyright (C) 2019-2021 HERE Europe B.V.
# SPDX-License-Identifier: Apache-2.0
"""
This module contains classes for accessing the responses from Location Services RESTful APIs.
"""
import json
import flexpolyline as fp
from geojson import Feature, FeatureCollection, LineString, Point, Polygon
from pandas import DataFrame
[docs]class ApiResponse:
"""Base class for all the responses from Location Services RESTful APIs."""
[docs] def __init__(self, **kwargs):
self._filters = {}
self.response = None
[docs] def __str__(self):
return self.as_json_string()
[docs] def as_json_string(self, encoding: str = "utf8"):
"""Return API response as json string."""
json_string = json.dumps(self.response, sort_keys=True, ensure_ascii=False).encode(
encoding
)
return json_string.decode()
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
feature_collection = FeatureCollection([])
for item in self.response["items"]:
f = Feature(
geometry=Point((item["position"]["lng"], item["position"]["lat"])),
properties=item,
)
feature_collection.features.append(f)
return feature_collection
[docs] @classmethod
def new(cls, resp):
"""Instantiate a response object from raw response returned by API."""
json_data = resp.copy()
obj = cls(**json_data)
obj.response = resp
return obj
[docs]class GeocoderResponse(ApiResponse):
"""A class representing the Geocoder API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"items": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs]class ReverseGeocoderResponse(ApiResponse):
"""A class representing the Reverse Geocoder API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"items": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs]class IsolineResponse(ApiResponse):
"""A class representing the Reverse Isoline routing API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {
"departure": None,
"arrival": None,
"isolines": None,
"notices": None,
}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
feature_collection = FeatureCollection([])
for isoline in self.response["isolines"]:
for polygon in isoline["polygons"]:
polyline = polygon["outer"]
lstring = fp.decode(polyline)
lstring = [(coord[1], coord[0]) for coord in lstring]
f = Feature(geometry=Polygon([lstring]), properties={"range": isoline["range"]})
feature_collection.features.append(f)
return feature_collection
[docs]class DiscoverResponse(ApiResponse):
"""A class representing the search discover API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"items": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs]class BrowseResponse(ApiResponse):
"""A class representing the search browse API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"items": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs]class LookupResponse(ApiResponse):
"""A class representing the search lookup API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"items": None}
for param, default in self._filters.items():
setattr(self, param, kwargs)
[docs]class RoutingResponse(ApiResponse):
"""A class representing the search routing API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"routes": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
feature_collection = FeatureCollection([])
for route in self.response["routes"]:
for section in route["sections"]:
polyline = section["polyline"]
lstring = fp.decode(polyline)
lstring = [(coord[1], coord[0], coord[2]) for coord in lstring]
f = Feature(geometry=LineString(lstring), properties=section)
feature_collection.features.append(f)
return feature_collection
[docs]class MatrixRoutingResponse(ApiResponse):
"""A class representing Matrix routing response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"matrix": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
raise NotImplementedError("This method is not valid for MatrixRoutingResponse.")
[docs] def to_distnaces_matrix(self):
"""Return distnaces matrix in a dataframe."""
if self.matrix and self.matrix.get("distances"):
distances = self.matrix.get("distances")
dest_count = self.matrix.get("numDestinations")
nested_distances = [
distances[i : i + dest_count] for i in range(0, len(distances), dest_count)
]
return DataFrame(nested_distances, columns=range(dest_count))
[docs] def to_travel_times_matrix(self):
"""Return travel times matrix in a dataframe."""
if self.matrix and self.matrix.get("travelTimes"):
distances = self.matrix.get("travelTimes")
dest_count = self.matrix.get("numDestinations")
nested_distances = [
distances[i : i + dest_count] for i in range(0, len(distances), dest_count)
]
return DataFrame(nested_distances, columns=range(dest_count))
[docs]class AutosuggestResponse(ApiResponse):
"""A class representing the Autosuggest API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {
"items": None,
"queryTerms": None,
}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs]class DestinationWeatherResponse(ApiResponse):
"""A class representing the Destination Weather API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"places": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
raise NotImplementedError("This method is not valid for DestinationWeatherResponse.")
[docs]class WeatherAlertsResponse(ApiResponse):
"""A class representing the Destination Weather API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"features": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
feature_collection = FeatureCollection([])
for feature in self.response["features"]:
f = Feature(geometry=feature["geometry"], properties=feature["properties"])
feature_collection.features.append(f)
return feature_collection
[docs]class TourPlanningResponse(ApiResponse):
"""A class representing the Tour Planning API response data."""
[docs] def __init__(self, **kwargs):
super().__init__()
self._filters = {"problemId": None, "statistic": None, "tours": None, "unassigned": None}
for param, default in self._filters.items():
setattr(self, param, kwargs.get(param, default))
[docs] def to_geojson(self):
"""Return API response as GeoJSON."""
raise NotImplementedError("This method is not valid for TourPlanningResponse.")