diff --git a/README.md b/README.md index b676fb1..fd4ade8 100644 --- a/README.md +++ b/README.md @@ -189,6 +189,38 @@ async def main(): print(response) # default Python print +asyncio.run(main()) +``` + +### KML response + +#### Returns: + +* results: TimeMapKmlResponse - TimeMapKmlResponse with isochrone shapes. + +#### Example: + +```python +import asyncio +from datetime import datetime + +from traveltimepy import Driving, Coordinates, TravelTimeSdk + + +async def main(): + sdk = TravelTimeSdk("YOUR_APP_ID", "YOUR_APP_KEY") + + response = await sdk.time_map_kml_async( + coordinates=[Coordinates(lat=51.507609, lng=-0.128315), Coordinates(lat=51.517609, lng=-0.138315)], + arrival_time=datetime.now(), + transportation=Driving() + ) + + print(results) # list of KML objects + print(results.results[0].pretty_string()) # human-readable output + print(results.results[0].search_id()) # search_id is the name of the Placemark + + asyncio.run(main()) ``` diff --git a/pyproject.toml b/pyproject.toml index 7181817..b2530f2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,8 @@ dependencies = [ "typing-extensions", "geojson-pydantic>=1.0.1", "shapely", + "fastkml==0.12", + "lxml", "dacite", "certifi>=2021.5.30", "aiohttp", @@ -64,3 +66,7 @@ exclude = "^(traveltimepy/proto/.*|build/.*|venv/.*)$" [[tool.mypy.overrides]] module = "traveltimepy.proto.*" follow_imports = "skip" + +[[tool.mypy.overrides]] +module = "fastkml" +ignore_missing_imports = true diff --git a/tests/time_map_test.py b/tests/time_map_test.py index 479930f..91c5f86 100644 --- a/tests/time_map_test.py +++ b/tests/time_map_test.py @@ -68,6 +68,22 @@ async def test_departures_wkt_no_holes(sdk: TravelTimeSdk): assert len(response.results) == 2 +@pytest.mark.asyncio +async def test_departures_kml(sdk: TravelTimeSdk): + response = await sdk.time_map_kml_async( + coordinates=[ + Coordinates(lat=51.507609, lng=-0.128315), + Coordinates(lat=51.517609, lng=-0.138315), + ], + departure_time=datetime.now(), + travel_time=900, + transportation=Driving(), + search_range=Range(enabled=True, width=1800), + level_of_detail=LevelOfDetail(scale_type="simple", level="lowest"), + ) + assert len(response.results) == 2 + + @pytest.mark.asyncio async def test_arrivals(sdk: TravelTimeSdk): results = await sdk.time_map_async( @@ -132,6 +148,22 @@ async def test_arrivals_wkt_no_holes(sdk: TravelTimeSdk): assert len(response.results) == 2 +@pytest.mark.asyncio +async def test_arrivals_kml(sdk: TravelTimeSdk): + response = await sdk.time_map_kml_async( + coordinates=[ + Coordinates(lat=51.507609, lng=-0.128315), + Coordinates(lat=51.517609, lng=-0.138315), + ], + arrival_time=datetime.now(), + travel_time=900, + transportation=Driving(), + search_range=Range(enabled=True, width=1800), + level_of_detail=LevelOfDetail(scale_type="simple", level="lowest"), + ) + assert len(response.results) == 2 + + @pytest.mark.asyncio async def test_union_departures(sdk: TravelTimeSdk): result = await sdk.union_async( diff --git a/traveltimepy/accept_type.py b/traveltimepy/accept_type.py index 8372398..b9b7f85 100644 --- a/traveltimepy/accept_type.py +++ b/traveltimepy/accept_type.py @@ -8,3 +8,4 @@ class AcceptType(Enum): BOUNDING_BOXES_JSON = "application/vnd.bounding-boxes+json" GEO_JSON = "application/geo+json" OCTET_STREAM = "application/octet-stream" + KML_XML = "application/vnd.google-earth.kml+xml" diff --git a/traveltimepy/dto/requests/time_map_kml.py b/traveltimepy/dto/requests/time_map_kml.py new file mode 100644 index 0000000..cca76b1 --- /dev/null +++ b/traveltimepy/dto/requests/time_map_kml.py @@ -0,0 +1,29 @@ +from typing import List + +from traveltimepy.dto.requests.request import TravelTimeRequest +from traveltimepy.dto.requests.time_map import ( + DepartureSearch, + ArrivalSearch, +) +from traveltimepy.dto.responses.time_map_kml import TimeMapKmlResponse +from traveltimepy.itertools import split, flatten + + +class TimeMapRequestKML(TravelTimeRequest[TimeMapKmlResponse]): + departure_searches: List[DepartureSearch] + arrival_searches: List[ArrivalSearch] + + def split_searches(self, window_size: int) -> List[TravelTimeRequest]: + return [ + TimeMapRequestKML( + departure_searches=departures, + arrival_searches=arrivals, + ) + for departures, arrivals in split( + self.departure_searches, self.arrival_searches, window_size + ) + ] + + def merge(self, responses: List[TimeMapKmlResponse]) -> TimeMapKmlResponse: + merged_features = flatten([response.results for response in responses]) + return TimeMapKmlResponse(results=merged_features) diff --git a/traveltimepy/dto/responses/time_map_kml.py b/traveltimepy/dto/responses/time_map_kml.py new file mode 100644 index 0000000..5b56f05 --- /dev/null +++ b/traveltimepy/dto/responses/time_map_kml.py @@ -0,0 +1,38 @@ +from typing import List +from fastkml import Placemark, kml +from pydantic import BaseModel + +# The stable version of fastkml is from 2021 and does not have types defined, nor +# a `py.typed` file. There are newer pre-release versions as new as 2024, but they +# seem to be undoccumented. +# TODO: Maybe port this into newer versions of fastkml for proper type checking support + + +class TimeMapKmlResult(BaseModel): + placemark: Placemark + + def search_id(self) -> str: + return self.placemark.name # type: ignore + + def pretty_string(self) -> str: # type: ignore + return self.placemark.to_string(prettyprint=True) + + class Config: + arbitrary_types_allowed = True + + +class TimeMapKmlResponse(BaseModel): + results: List[TimeMapKmlResult] + + +def parse_kml_as(kml_string: str) -> TimeMapKmlResponse: + k = kml.KML() + k.from_string(kml_string.encode("utf-8")) + + results = [ + TimeMapKmlResult(placemark=feature) + for feature in k.features() + if isinstance(feature, kml.Placemark) + ] + + return TimeMapKmlResponse(results=results) diff --git a/traveltimepy/http.py b/traveltimepy/http.py index 19dca1b..2b83658 100644 --- a/traveltimepy/http.py +++ b/traveltimepy/http.py @@ -1,12 +1,16 @@ import asyncio import json from dataclasses import dataclass -from typing import TypeVar, Type, Dict, Optional +from typing import TypeVar, Type, Dict, Optional, Union from aiohttp import ClientSession, ClientResponse, TCPConnector, ClientTimeout from pydantic import BaseModel from traveltimepy.dto.requests.request import TravelTimeRequest +from traveltimepy.dto.responses.time_map_kml import ( + TimeMapKmlResponse, + parse_kml_as, +) from traveltimepy.dto.responses.error import ResponseError from traveltimepy.errors import ApiError from aiohttp_retry import RetryClient, ExponentialRetry @@ -34,12 +38,15 @@ async def send_post_request_async( headers: Dict[str, str], request: TravelTimeRequest, rate_limit: AsyncLimiter, -) -> T: +) -> Union[T, TimeMapKmlResponse]: async with rate_limit: async with client.post( url=url, headers=headers, data=request.model_dump_json() ) as resp: - return await _process_response(response_class, resp) + if response_class == TimeMapKmlResponse: + return await _process_kml_response(resp) + else: + return await _process_json_response(response_class, resp) async def send_post_async( @@ -103,20 +110,35 @@ async def send_get_async( headers=headers, params=params, ) as resp: - return await _process_response(response_class, resp) + return await _process_json_response(response_class, resp) + + +def _handle_non_ok_response(json_data): + parsed = ResponseError.model_validate_json(json.dumps(json_data)) + msg = ( + f"Travel Time API request failed: {parsed.description}\n" + f"Error code: {parsed.error_code}\n" + f"Additional info: {parsed.additional_info}\n" + f"<{parsed.documentation_link}>\n" + ) + raise ApiError(msg) -async def _process_response(response_class: Type[T], response: ClientResponse) -> T: +async def _process_json_response( + response_class: Type[T], response: ClientResponse +) -> T: text = await response.text() json_data = json.loads(text) if response.status != 200: - parsed = ResponseError.model_validate_json(json.dumps(json_data)) - msg = ( - f"Travel Time API request failed: {parsed.description}\n" - f"Error code: {parsed.error_code}\n" - f"Additional info: {parsed.additional_info}\n" - f"<{parsed.documentation_link}>\n" - ) - raise ApiError(msg) + return _handle_non_ok_response(json_data) else: return response_class.model_validate(json_data) + + +async def _process_kml_response(response: ClientResponse) -> TimeMapKmlResponse: + text = await response.text() + if response.status != 200: + json_data = json.loads(text) + return _handle_non_ok_response(json_data) + else: + return parse_kml_as(text) diff --git a/traveltimepy/mapper.py b/traveltimepy/mapper.py index bb5a818..5669cbe 100644 --- a/traveltimepy/mapper.py +++ b/traveltimepy/mapper.py @@ -3,6 +3,7 @@ from traveltimepy.dto.requests.distance_map import DistanceMapRequest from traveltimepy.dto.requests.time_map_geojson import TimeMapRequestGeojson +from traveltimepy.dto.requests.time_map_kml import TimeMapRequestKML from traveltimepy.dto.requests.time_map_wkt import TimeMapWKTRequest from traveltimepy.errors import ApiError from traveltimepy.proto import TimeFilterFastRequest_pb2 @@ -510,6 +511,60 @@ def create_time_map_wkt( raise ApiError("arrival_time or departure_time should be specified") +def create_time_map_kml( + coordinates: List[Coordinates], + transportation: Union[ + PublicTransport, + Driving, + Ferry, + Walking, + Cycling, + DrivingTrain, + CyclingPublicTransport, + ], + travel_time: int, + time_info: TimeInfo, + search_range: Optional[Range], + level_of_detail: Optional[LevelOfDetail], + snapping: Optional[Snapping] = None, +) -> TimeMapRequestKML: + if isinstance(time_info, ArrivalTime): + return TimeMapRequestKML( + arrival_searches=[ + time_map.ArrivalSearch( + id=f"Search {ind}", + coords=cur_coordinates, + travel_time=travel_time, + arrival_time=time_info.value, + transportation=transportation, + range=search_range, + level_of_detail=level_of_detail, + snapping=snapping, + ) + for ind, cur_coordinates in enumerate(coordinates) + ], + departure_searches=[], + ) + elif isinstance(time_info, DepartureTime): + return TimeMapRequestKML( + departure_searches=[ + time_map.DepartureSearch( + id=f"Search {ind}", + coords=cur_coordinates, + travel_time=travel_time, + departure_time=time_info.value, + transportation=transportation, + range=search_range, + snapping=snapping, + ) + for ind, cur_coordinates in enumerate(coordinates) + ], + arrival_searches=[], + ) + else: + raise ApiError("arrival_time or departure_time should be specified") + + def create_distance_map( coordinates: List[Coordinates], transportation: Union[ diff --git a/traveltimepy/sdk.py b/traveltimepy/sdk.py index e6d8c9e..557a299 100644 --- a/traveltimepy/sdk.py +++ b/traveltimepy/sdk.py @@ -17,6 +17,7 @@ from traveltimepy.dto.responses.time_map_wkt import ( TimeMapWKTResponse, ) +from traveltimepy.dto.responses.time_map_kml import TimeMapKmlResponse from traveltimepy.dto.responses.time_filter_proto import TimeFilterProtoResponse from traveltimepy.dto.transportation import ( PublicTransport, @@ -71,6 +72,7 @@ create_union, create_time_map_geojson, create_time_map_wkt, + create_time_map_kml, ) from traveltimepy.proto_http import send_proto_async @@ -645,6 +647,43 @@ async def time_map_wkt_no_holes_async( ) return resp + async def time_map_kml_async( + self, + coordinates: List[Coordinates], + transportation: Union[ + PublicTransport, + Driving, + Ferry, + Walking, + Cycling, + DrivingTrain, + CyclingPublicTransport, + ], + departure_time: Optional[datetime] = None, + arrival_time: Optional[datetime] = None, + travel_time: int = 3600, + search_range: Optional[Range] = None, + level_of_detail: Optional[LevelOfDetail] = None, + snapping: Optional[Snapping] = None, + ) -> TimeMapKmlResponse: + time_info = get_time_info(departure_time, arrival_time) + resp = await send_post_async( + TimeMapKmlResponse, + "time-map", + self._headers(AcceptType.KML_XML), + create_time_map_kml( + coordinates, + transportation, + travel_time, + time_info, + search_range, + level_of_detail, + snapping, + ), + self._sdk_params, + ) + return resp + async def distance_map_async( self, coordinates: List[Coordinates],