diff --git a/vigobusapi/app.py b/vigobusapi/app.py index 879541d..9c50280 100644 --- a/vigobusapi/app.py +++ b/vigobusapi/app.py @@ -4,6 +4,7 @@ # # Native # # import io +import json from typing import Optional, Set # # Installed # # @@ -108,7 +109,7 @@ async def endpoint_get_stop_map( ): """Get a picture of a map with the stop location marked on it.""" stop = await get_stop(stop_id) - if (stop.lat, stop.lon) == (None, None): + if not stop.has_location: raise HTTPException(status_code=409, detail="The stop does not have information about the location") map_request = GoogleMapRequest( @@ -126,14 +127,30 @@ async def endpoint_get_stop_map( @app.get("/stops/map") async def endpoint_get_stops_map( - stops_ids: Set[int] = Query(None, alias="stop_id", min_items=1, max_items=35), + stops_ids: Set[int] = Query(None, alias="stop_id", + min_items=1, max_items=len(GoogleMapRequest.Tag.get_allowed_labels())), map_params: MapQueryParams = Depends(), ): - """Get a picture of a map with the locations of the given stops marked on it.""" - stops = await get_stops(stops_ids) + """Get a picture of a map with the locations of the given stops marked on it. + + Non existing stops, or those without location available, are ignored, + but if none of the given stops are valid, returns 404. - # noinspection PyTypeChecker - stops_tags = [GoogleMapRequest.Tag(label=stop.stop_id, location_x=stop.lat, location_y=stop.lon) for stop in stops] + A header "X-Stops-Tags" is returned, being a JSON associating the Stops IDs with the tag label on the map, + with the format: {"" : ""} + """ + stops = await get_stops(stops_ids) + stops = [stop for stop in stops if stop.has_location] + if not stops: + raise HTTPException(status_code=404, detail="None of the stops exist or have location available") + + stops_tags = list() + stops_tags_relation = dict() + for i, stop in enumerate(stops): + tag_label = GoogleMapRequest.Tag.get_allowed_labels()[i] + tag = GoogleMapRequest.Tag(label=tag_label, location_x=stop.lat, location_y=stop.lon) + stops_tags.append(tag) + stops_tags_relation[stop.stop_id] = tag_label map_request = GoogleMapRequest( size_x=map_params.size_x, @@ -146,7 +163,8 @@ async def endpoint_get_stops_map( return StreamingResponse( content=io.BytesIO(map_data), - media_type="image/png" + media_type="image/png", + headers={"X-Stops-Tags": json.dumps(stops_tags_relation)} ) @@ -157,7 +175,7 @@ async def endpoint_get_stop_photo( size_y: int = google_maps_settings.stop_photo_default_size_y, ): stop = await get_stop(stop_id) - if (stop.lat, stop.lon) == (None, None): + if not stop.has_location: raise HTTPException(status_code=409, detail="The stop does not have information about the location") photo_request = GoogleStreetviewRequest( diff --git a/vigobusapi/entities.py b/vigobusapi/entities.py index a8947e7..4379a07 100644 --- a/vigobusapi/entities.py +++ b/vigobusapi/entities.py @@ -77,6 +77,10 @@ def get_mongo_dict(self): d.pop("source") return d + @property + def has_location(self): + return self.lat is not None and self.lon is not None + OptionalStop = Optional[Stop] StopOrNotExist = Union[Stop, StopNotExist] diff --git a/vigobusapi/services/google_maps.py b/vigobusapi/services/google_maps.py deleted file mode 100644 index 545b741..0000000 --- a/vigobusapi/services/google_maps.py +++ /dev/null @@ -1,246 +0,0 @@ -"""GOOGLE MAPS -Functions for getting static Maps and Pictures, using Google Maps and Streetview static APIs -""" - -# # Native # # -import string -from enum import Enum -from typing import * - -# # Installed # # -import pydantic -from pydantic import BaseModel - -# # Project # # -from vigobusapi.settings import google_maps_settings as settings -from vigobusapi.utils import ChecksumableClass, new_hash_values, update_hash_values -from vigobusapi.logger import logger -from .http_requester import http_request, ListOfTuples - -__all__ = ("GoogleMapRequest", "GoogleStreetviewRequest", "get_map", "get_photo") - -# TODO May refactor in package with different modules (at least split classes and logic) - -GOOGLE_MAPS_STATIC_API_URL = "https://maps.googleapis.com/maps/api/staticmap" -GOOGLE_STREETVIEW_STATIC_API_URL = "https://maps.googleapis.com/maps/api/streetview" - - -def get_labelled_icon_url(label: str) -> str: - """Get an URL pointing to a picture of a map marker, with a custom label on top of it""" - # TODO self-hosted generation, and/or cache of generated labels - return f"https://cdn.mapmarker.io/api/v1/font-awesome/v5/pin?text={label}&size=40&background=D94B43&color=000000&hoffset=-1" - - -class _GoogleMapsBaseRequest(BaseModel, ChecksumableClass): - location_x: float - location_y: float - size_x: int - size_y: int - - @property - def location_str(self) -> Optional[str]: - if self.location_x is None or self.location_y is None: - return None - return f"{self.location_x},{self.location_y}" - - @property - def size_str(self): - return f"{self.size_x}x{self.size_y}" - - @property - def checksum_hash(self): - """Returns a SHA256 checksum of all the fields in the request""" - return new_hash_values( - self.location_x, - self.location_y, - self.size_x, - self.size_y, - algorithm="sha256" - ) - - -class GoogleMapRequest(_GoogleMapsBaseRequest): - - # Embed classes # - - class Tag(BaseModel, ChecksumableClass): - __ALLOWED_LABELS = [*[str(i) for i in range(1, 10)], *[c for c in string.ascii_uppercase]] - - label: Optional[str] = None # TODO constrain values accepted (avoid enum?) - icon_url: Optional[str] = None - location_x: float - location_y: float - - @classmethod - def get_allowed_labels(cls): - return cls.__ALLOWED_LABELS - - @pydantic.root_validator(pre=True) - def label_to_icon(cls, kwargs: dict): - """If label is not an "allowed label", generate an icon for it and set it as "icon_url".""" - label = kwargs.get("label") - if label not in cls.__ALLOWED_LABELS: - kwargs["label"] = None - kwargs["icon_url"] = get_labelled_icon_url(label) - return kwargs - - @property - def location_str(self): - return f"{self.location_x},{self.location_y}" - - @property - def checksum_hash(self): - return new_hash_values( - self.label, - self.location_x, - self.location_y, - algorithm="md5" - ) - - class MapTypes(str, Enum): - """Available map types. - - References: - https://developers.google.com/maps/documentation/maps-static/start#MapTypes - """ - roadmap = "roadmap" - """specifies a standard roadmap image, as is normally shown on the Google Maps website""" - satellite = "satellite" - """specifies a satellite image""" - terrain = "terrain" - """specifies a physical relief map image, showing terrain and vegetation""" - hybrid = "hybrid" - """specifies a hybrid of the satellite and roadmap image, - showing a transparent layer of major streets and place names on the satellite image""" - - # Properties # - - @property - def checksum_hash(self): - _hash = super().checksum_hash - - if self.tags: - sorted_tags_checksums = sorted(tag.checksum_value for tag in self.tags) - else: - sorted_tags_checksums = "NoTags" - - return update_hash_values( - self.zoom, - self.map_type.value, - sorted_tags_checksums, - _hash=_hash - ) - - # Class Attributes # - - location_x: Optional[float] = None - location_y: Optional[float] = None - tags: Optional[List[Tag]] = None - zoom: int - """https://developers.google.com/maps/documentation/maps-static/start#Zoomlevels""" - map_type: MapTypes - - -class GoogleStreetviewRequest(_GoogleMapsBaseRequest): - pass - - -async def _request(url: str, params: Union[dict, ListOfTuples], expect_http_error: bool = False): - """HTTP requester for Google Maps API calls, automatically including the configured API key. - Raises exception if the API Key is not configured. - - :param url: URL for the Google API, WITHOUT query parameters - :param params: query parameters - :param expect_http_error: if True, raise_for_status=False and not_retry_400_errors=True - """ - if not settings.enabled: - raise Exception("Google Maps API Key not set in settings") - - if isinstance(params, list): - params.append(("key", settings.api_key)) - else: - params = dict(**params, key=settings.api_key) - - return await http_request( - url=url, - method="GET", - params=params, - retries=1, - raise_for_status=not expect_http_error, - not_retry_400_errors=expect_http_error - ) - - -def _get_map_tags_params(request_tags: List[GoogleMapRequest.Tag]) -> ListOfTuples: - params = list() - for tag in request_tags: - tag_param_values = [tag.location_str] # Location always at the end - - if tag.label: - tag_param_values.insert(0, "label:" + tag.label) - if tag.icon_url: - tag_param_values.insert(0, "icon:" + tag.icon_url) - - tag_param = "|".join(tag_param_values) - params.append(("markers", tag_param)) - - return params - - -def _get_map_params(request: GoogleMapRequest) -> ListOfTuples: - params = [ - ("size", request.size_str), - ("maptype", request.map_type.value), - ("language", settings.language), - ("format", "png8"), - ] - - location_str = request.location_str - if location_str: - params.append(("center", location_str)) - params.append(("zoom", str(request.zoom))) - - if request.tags: - params.extend(_get_map_tags_params(request.tags)) - - return params - - -async def get_map(request: GoogleMapRequest) -> bytes: - """Get a static Map picture from the Google Maps Static API. Return the acquired PNG picture as bytes. - - References: - https://developers.google.com/maps/documentation/maps-static/overview - https://developers.google.com/maps/documentation/maps-static/start - """ - logger.bind(map_request=request.dict()).debug("Requesting Google Static Map picture...") - # TODO cache loaded pictures - - params = _get_map_params(request) - return (await _request(url=GOOGLE_MAPS_STATIC_API_URL, params=params)).content - - -async def get_photo(request: GoogleStreetviewRequest) -> Optional[bytes]: - """Get a static StreetView picture from the Google StreetView Static API. Return the acquired PNG picture as bytes. - If the requested location does not have an available picture, returns None. - - References: - https://developers.google.com/maps/documentation/streetview/overview - """ - logger.bind(streetview_request=request.dict()).debug("Requesting Google Static StreetView picture...") - # TODO cache loaded pictures - # TODO Support specific parameters for tuning camera, if required - params = [ - ("location", request.location_str), - ("size", request.size_str), - ("return_error_code", "true"), - ("source", "outdoor") - ] - - response = await _request(GOOGLE_STREETVIEW_STATIC_API_URL, params=params, expect_http_error=True) - if response.status_code == 404: - logger.debug("No StreetView picture available for the request") - return None - - response.raise_for_status() - return response.content diff --git a/vigobusapi/services/google_maps/__init__.py b/vigobusapi/services/google_maps/__init__.py new file mode 100644 index 0000000..313aa09 --- /dev/null +++ b/vigobusapi/services/google_maps/__init__.py @@ -0,0 +1,7 @@ +"""GOOGLE MAPS +Classes and functions for acquiring static maps and photos using Google Maps & Street View Static APIs +""" + +from ._entities import * +from ._getter_maps import * +from ._getter_streetview import * diff --git a/vigobusapi/services/google_maps/_entities.py b/vigobusapi/services/google_maps/_entities.py new file mode 100644 index 0000000..166bdf8 --- /dev/null +++ b/vigobusapi/services/google_maps/_entities.py @@ -0,0 +1,121 @@ +"""GOOGLE MAPS - ENTITIES +Classes used to define Google Maps/StreetView API requests +""" + +# # Native # # +import string +from enum import Enum +from typing import * + +# # Installed # # +from pydantic import BaseModel + +# # Project # # +from vigobusapi.utils import ChecksumableClass, new_hash_values, update_hash_values + +__all__ = ("GoogleMapRequest", "GoogleStreetviewRequest") + + +class _GoogleMapsBaseRequest(BaseModel, ChecksumableClass): + location_x: float + location_y: float + size_x: int + size_y: int + + @property + def location_str(self) -> Optional[str]: + if self.location_x is None or self.location_y is None: + return None + return f"{self.location_x},{self.location_y}" + + @property + def size_str(self): + return f"{self.size_x}x{self.size_y}" + + @property + def checksum_hash(self): + """Returns a SHA256 checksum of all the fields in the request""" + return new_hash_values( + self.location_x, + self.location_y, + self.size_x, + self.size_y, + algorithm="sha256" + ) + + +class GoogleMapRequest(_GoogleMapsBaseRequest): + + # Embed classes # + + class Tag(BaseModel, ChecksumableClass): + __ALLOWED_LABELS = [*[str(i) for i in range(1, 10)], *[c for c in string.ascii_uppercase]] + + label: Optional[str] = None # TODO constrain values accepted (avoid enum?) + location_x: float + location_y: float + + @classmethod + def get_allowed_labels(cls): + """Get a list with the available labels for Tags, corresponding to numbers 0~9 and characters A~Z.""" + return cls.__ALLOWED_LABELS + + @property + def location_str(self): + return f"{self.location_x},{self.location_y}" + + @property + def checksum_hash(self): + return new_hash_values( + self.label, + self.location_x, + self.location_y, + algorithm="md5" + ) + + class MapTypes(str, Enum): + """Available map types. + + References: + https://developers.google.com/maps/documentation/maps-static/start#MapTypes + """ + roadmap = "roadmap" + """specifies a standard roadmap image, as is normally shown on the Google Maps website""" + satellite = "satellite" + """specifies a satellite image""" + terrain = "terrain" + """specifies a physical relief map image, showing terrain and vegetation""" + hybrid = "hybrid" + """specifies a hybrid of the satellite and roadmap image, + showing a transparent layer of major streets and place names on the satellite image""" + + # Properties # + + @property + def checksum_hash(self): + _hash = super().checksum_hash + + if self.tags: + sorted_tags_checksums = sorted(tag.checksum_value for tag in self.tags) + else: + sorted_tags_checksums = "NoTags" + + return update_hash_values( + self.zoom, + self.map_type.value, + sorted_tags_checksums, + _hash=_hash + ) + + # Class Attributes # + + location_x: Optional[float] = None + location_y: Optional[float] = None + tags: Optional[List[Tag]] = None + zoom: int + """https://developers.google.com/maps/documentation/maps-static/start#Zoomlevels""" + map_type: MapTypes + + +class GoogleStreetviewRequest(_GoogleMapsBaseRequest): + pass diff --git a/vigobusapi/services/google_maps/_getter_maps.py b/vigobusapi/services/google_maps/_getter_maps.py new file mode 100644 index 0000000..74943b2 --- /dev/null +++ b/vigobusapi/services/google_maps/_getter_maps.py @@ -0,0 +1,63 @@ +"""GOOGLE MAPS - GETTER MAPS +Functions for acquiring a picture of a Map +""" + +# # Native # # +from typing import * + +# # Project # # +from vigobusapi.settings import google_maps_settings as settings +from vigobusapi.logger import logger +from ._requester import google_maps_request, ListOfTuples +from ._entities import GoogleMapRequest + +__all__ = ("get_map",) + +GOOGLE_MAPS_STATIC_API_URL = "https://maps.googleapis.com/maps/api/staticmap" + + +def _get_map_tags_params(request_tags: List[GoogleMapRequest.Tag]) -> ListOfTuples: + params = list() + for tag in request_tags: + tag_param_values = [tag.location_str] # Location always at the end + + if tag.label: + tag_param_values.insert(0, "label:" + tag.label) + + tag_param = "|".join(tag_param_values) + params.append(("markers", tag_param)) + + return params + + +def _get_map_params(request: GoogleMapRequest) -> ListOfTuples: + params = [ + ("size", request.size_str), + ("maptype", request.map_type.value), + ("language", settings.language), + ("format", "png8"), + ] + + location_str = request.location_str + if location_str: + params.append(("center", location_str)) + params.append(("zoom", str(request.zoom))) + + if request.tags: + params.extend(_get_map_tags_params(request.tags)) + + return params + + +async def get_map(request: GoogleMapRequest) -> bytes: + """Get a static Map picture from the Google Maps Static API. Return the acquired PNG picture as bytes. + + References: + https://developers.google.com/maps/documentation/maps-static/overview + https://developers.google.com/maps/documentation/maps-static/start + """ + logger.bind(map_request=request.dict()).debug("Requesting Google Static Map picture...") + # TODO cache loaded pictures + + params = _get_map_params(request) + return (await google_maps_request(url=GOOGLE_MAPS_STATIC_API_URL, params=params)).content diff --git a/vigobusapi/services/google_maps/_getter_streetview.py b/vigobusapi/services/google_maps/_getter_streetview.py new file mode 100644 index 0000000..446bb63 --- /dev/null +++ b/vigobusapi/services/google_maps/_getter_streetview.py @@ -0,0 +1,47 @@ +"""GOOGLE MAPS - GETTER STREETVIEW +Functions for acquiring a photo of a location +""" + +# # Native # # +from typing import * + +# # Project # # +from vigobusapi.logger import logger +from ._entities import GoogleStreetviewRequest +from ._requester import google_maps_request, ListOfTuples + +__all__ = ("get_photo",) + +GOOGLE_MAPS_STATIC_API_URL = "https://maps.googleapis.com/maps/api/staticmap" +GOOGLE_STREETVIEW_STATIC_API_URL = "https://maps.googleapis.com/maps/api/streetview" + + +def _get_photo_params(request: GoogleStreetviewRequest) -> ListOfTuples: + params = [ + ("location", request.location_str), + ("size", request.size_str), + ("return_error_code", "true"), + ("source", "outdoor") + ] + return params + + +async def get_photo(request: GoogleStreetviewRequest) -> Optional[bytes]: + """Get a static StreetView picture from the Google StreetView Static API. Return the acquired PNG picture as bytes. + If the requested location does not have an available picture, returns None. + + References: + https://developers.google.com/maps/documentation/streetview/overview + """ + logger.bind(streetview_request=request.dict()).debug("Requesting Google Static StreetView picture...") + # TODO cache loaded pictures + # TODO Support specific parameters for tuning camera, if required + + params = _get_photo_params(request) + response = await google_maps_request(GOOGLE_STREETVIEW_STATIC_API_URL, params=params, expect_http_error=True) + if response.status_code == 404: + logger.debug("No StreetView picture available for the request") + return None + + response.raise_for_status() + return response.content diff --git a/vigobusapi/services/google_maps/_requester.py b/vigobusapi/services/google_maps/_requester.py new file mode 100644 index 0000000..7910d88 --- /dev/null +++ b/vigobusapi/services/google_maps/_requester.py @@ -0,0 +1,38 @@ +"""GOOGLE MAPS - REQUESTER +Functions for requesting the Google Maps/StreetView APIs +""" + +# # Native # # +from typing import * + +# # Project # # +from vigobusapi.settings import google_maps_settings as settings +from vigobusapi.services.http_requester import http_request, ListOfTuples + +__all__ = ("google_maps_request", "ListOfTuples") + + +async def google_maps_request(url: str, params: Union[dict, ListOfTuples], expect_http_error: bool = False): + """HTTP requester for Google Maps API calls, automatically including the configured API key. + Raises exception if the API Key is not configured. + + :param url: URL for the Google API, WITHOUT query parameters + :param params: query parameters + :param expect_http_error: if True, raise_for_status=False and not_retry_400_errors=True + """ + if not settings.enabled: + raise Exception("Google Maps API Key not set in settings") + + if isinstance(params, list): + params.append(("key", settings.api_key)) + else: + params = dict(**params, key=settings.api_key) + + return await http_request( + url=url, + method="GET", + params=params, + retries=1, + raise_for_status=not expect_http_error, + not_retry_400_errors=expect_http_error + )