Add configurable and inferred base coords for address resolution

README.md

@@ -36,6 +36,19 @@ travel_time = asyncio.run(get_time(start, end))
 print(travel_time)
 ```
 
+### Address resolving base coordinates
+
+When one or both endpoints are addresses, `calc_routes()` resolves them via Waze search.
+You can provide custom base coordinates to make sure Waze tries to resolve the address near those coordinates:
+
+```python
+await client.calc_routes(start, end, base_coords=(48.137154, 11.576124))
+```
+
+If `base_coords` is omitted and exactly one endpoint is already coordinates,
+that coordinate endpoint is used automatically as base coordinates for
+resolving the address.
+
 ---
 
 [<img src="https://raw.githubusercontent.com/eifinger/pywaze/main/docs/images/bmc-button.svg" width=150 height=40 style="margin: 5px"/>](https://www.buymeacoffee.com/eifinger)

src/pywaze/route_calculator.py

@@ -1,22 +1,31 @@
 """Waze route calculator."""
 
+import logging
+import re
 from dataclasses import dataclass
 from typing import Any, Literal, TypedDict
+
 import httpx
-import re
-import logging
 
 logger = logging.getLogger(__name__)
 
 
-class Coords(TypedDict):
-    """Coordinates and bounds."""
+class BaseCoords(TypedDict):
+    """Base coordinates."""
 
     lat: float
     lon: float
+
+
+class Coords(BaseCoords):
+    """Coordinates and bounds."""
+
     bounds: dict[str, float]
 
 
+BaseCoordsInput = BaseCoords | str | tuple[float, float]
+
+
 @dataclass(frozen=True)
 class CalcRoutesResponse:
     """The Response from this lib."""
@@ -43,7 +52,7 @@ class WazeRouteCalculator:
         "User-Agent": "pywaze",
         "referer": WAZE_URL,
     }
-    BASE_COORDS = {
+    BASE_COORDS: dict[str, BaseCoords] = {
         "US": {"lat": 40.713, "lon": -74.006},
         "NA": {"lat": 40.713, "lon": -74.006},
         "EU": {"lat": 47.498, "lon": 19.040},
@@ -84,24 +93,44 @@ def already_coords(self, address: str) -> bool:
         m = re.search(self.COORD_MATCH, address)
         return m is not None
 
-    async def _ensure_coords(self, address: str) -> Coords:
-        coords = None
+    async def _ensure_coords(
+        self,
+        address: str,
+        base_coords: BaseCoords | None = None,
+    ) -> Coords:
         if self.already_coords(address):
-            coords = self.coords_string_parser(address)
-        else:
-            coords = await self.address_to_coords(address)
-        return coords
+            return self.coords_string_parser(address)
+        return await self.address_to_coords(address, base_coords=base_coords)
 
     def coords_string_parser(self, coords: str) -> Coords:
         """Parse the address string into coordinates to match address_to_coords return object."""
 
         lat, lon = coords.split(",")
         return {"lat": float(lat.strip()), "lon": float(lon.strip()), "bounds": {}}
 
-    async def address_to_coords(self, address: str) -> Coords:
+    def _normalize_base_coords(self, base_coords: BaseCoordsInput) -> BaseCoords:
+        """Normalize supported base coordinate input formats."""
+
+        if isinstance(base_coords, str):
+            parsed_coords = self.coords_string_parser(base_coords)
+            return {"lat": parsed_coords["lat"], "lon": parsed_coords["lon"]}
+
+        if isinstance(base_coords, tuple):
+            return {"lat": float(base_coords[0]), "lon": float(base_coords[1])}
+
+        if isinstance(base_coords, dict):
+            return {"lat": float(base_coords["lat"]), "lon": float(base_coords["lon"])}
+
+        raise TypeError("base_coords must be a coords string, tuple, or dict")
+
+    async def address_to_coords(
+        self,
+        address: str,
+        base_coords: BaseCoords | None = None,
+    ) -> Coords:
         """Convert address to coordinates."""
 
-        base_coords = self.BASE_COORDS[self.region]
+        base_coords = base_coords or self.BASE_COORDS[self.region]
         get_cord = self.COORD_SERVERS[self.region]
         url_options: dict[str, str | float] = {
             "q": address,
@@ -268,11 +297,29 @@ async def calc_routes(
         time_delta: int = 0,
         real_time: bool = True,
         stop_at_bounds: bool = False,
+        base_coords: BaseCoordsInput | None = None,
     ) -> list[CalcRoutesResponse]:
         """Get route info with enhanced calculations like total distance."""
 
-        start_coords = await self._ensure_coords(start)
-        end_coords = await self._ensure_coords(end)
+        resolved_base_coords = (
+            self._normalize_base_coords(base_coords)
+            if base_coords is not None
+            else None
+        )
+
+        start_is_coords = self.already_coords(start)
+        end_is_coords = self.already_coords(end)
+
+        if resolved_base_coords is None:
+            if start_is_coords and not end_is_coords:
+                resolved_base_coords = self._normalize_base_coords(start)
+            elif end_is_coords and not start_is_coords:
+                resolved_base_coords = self._normalize_base_coords(end)
+
+        start_coords = await self._ensure_coords(
+            start, base_coords=resolved_base_coords
+        )
+        end_coords = await self._ensure_coords(end, base_coords=resolved_base_coords)
 
         routes = await self.get_routes(
             start_coords,
@@ -312,7 +359,7 @@ async def close(self) -> None:
         """Close the client."""
         await self.client.aclose()
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> "WazeRouteCalculator":
         """Support asynchronous context manager protocol."""
         return self
 

tests/test_route_calculator.py

@@ -1,8 +1,11 @@
 """Tests for route_calculator module."""
 
+from httpx import Response
 import pytest
 from pywaze import route_calculator
+from respx import MockRouter
 from tests.const import (
+    ADDRESS_TO_COORDS_RESPONSE_WIESBADEN,
     EMPTY_ROUTE_NAME_RESPONSE,
     GET_ROUTE_RESPONSE_ADDRESSES,
     GET_ROUTE_RESPONSE_COORDS,
@@ -246,6 +249,87 @@ async def test_calc_route_info(
             )
 
 
+async def test_calc_routes_uses_custom_base_coords_for_address_lookup(
+    respx_mock: MockRouter,
+):
+    """Use explicitly provided base coordinates for address resolving."""
+
+    route_response = {
+        "response": {
+            "results": [{"length": 1000, "crossTime": 60}],
+            "streetNames": [],
+        }
+    }
+    respx_mock.get(
+        "https://routing-livemap-row.waze.com/RoutingManager/routingRequest"
+    ).mock(return_value=Response(200, json=route_response))
+
+    coords_lookup_route = respx_mock.route(
+        path="/row-SearchServer/mozi",
+        params={"q": "Luisenstraße 30 65185 Wiesbaden, Germany"},
+    ).mock(return_value=Response(200, json=ADDRESS_TO_COORDS_RESPONSE_WIESBADEN))
+
+    async with route_calculator.WazeRouteCalculator() as client:
+        await client.calc_routes(
+            "50.00332659227126,8.262322651915843",
+            "Luisenstraße 30 65185 Wiesbaden, Germany",
+            base_coords=(48.137154, 11.576124),
+        )
+
+    request_params = coords_lookup_route.calls.last.request.url.params
+    assert float(request_params["lat"]) == pytest.approx(48.137154)
+    assert float(request_params["lon"]) == pytest.approx(11.576124)
+
+
+@pytest.mark.parametrize(
+    ("start", "end", "expected_lat", "expected_lon"),
+    (
+        (
+            "50.00332659227126,8.262322651915843",
+            "Luisenstraße 30 65185 Wiesbaden, Germany",
+            50.00332659227126,
+            8.262322651915843,
+        ),
+        (
+            "Luisenstraße 30 65185 Wiesbaden, Germany",
+            "50.00332659227126,8.262322651915843",
+            50.00332659227126,
+            8.262322651915843,
+        ),
+    ),
+)
+async def test_calc_routes_uses_other_endpoint_coords_as_base_when_missing(
+    start: str,
+    end: str,
+    expected_lat: float,
+    expected_lon: float,
+    respx_mock: MockRouter,
+):
+    """Use coordinate endpoint as base coords when only one side is an address."""
+
+    route_response = {
+        "response": {
+            "results": [{"length": 1000, "crossTime": 60}],
+            "streetNames": [],
+        }
+    }
+    respx_mock.get(
+        "https://routing-livemap-row.waze.com/RoutingManager/routingRequest"
+    ).mock(return_value=Response(200, json=route_response))
+
+    coords_lookup_route = respx_mock.route(
+        path="/row-SearchServer/mozi",
+        params={"q": "Luisenstraße 30 65185 Wiesbaden, Germany"},
+    ).mock(return_value=Response(200, json=ADDRESS_TO_COORDS_RESPONSE_WIESBADEN))
+
+    async with route_calculator.WazeRouteCalculator() as client:
+        await client.calc_routes(start, end)
+
+    request_params = coords_lookup_route.calls.last.request.url.params
+    assert float(request_params["lat"]) == pytest.approx(expected_lat)
+    assert float(request_params["lon"]) == pytest.approx(expected_lon)
+
+
 @pytest.mark.usefixtures("timeout_mock")
 async def test_calc_route_info_timeout():
     """Test calc_route_info with timeout."""