Adding time map kml support

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())
 ```
 

pyproject.toml

@@ -16,6 +16,8 @@ dependencies = [
 	"typing-extensions",
 	"geojson-pydantic>=1.0.1",
 	"shapely",
+	"fastkml",
+	"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

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(

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"

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)

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)

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)

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[

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],