Add searching for places

src/rivian/rivian.py

@@ -644,6 +644,67 @@ async def depart_now(
         }
         return await self.__graphql_query(headers, url, graphql_json)
 
+    async def search_places(
+        self, query: str, latitude: float, longitude: float
+    ) -> list[dict[str, Any]]:
+        """Search for places like the Rivian app's map search.
+
+        Results come in the search's own order, each with a Google place `id`,
+        `name`, `location` and the `driveDetails` from `latitude`, `longitude`.
+        """
+        url = GRAPHQL_GATEWAY
+        headers = BASE_HEADERS | {
+            "Csrf-Token": self._csrf_token,
+            "A-Sess": self._app_session_token,
+            "U-Sess": self._user_session_token,
+        }
+        graphql_json = {
+            "operationName": "places",
+            "query": "query places($params: PlacesQueryParams!) { places(params: $params) { results { id name location { formattedAddress coordinate { latitude longitude } navCoordinate { latitude longitude } } driveDetails { driveDistance driveDuration } } } }",
+            "variables": {
+                "params": {
+                    "query": query,
+                    "userLocation": {"latitude": latitude, "longitude": longitude},
+                    "autocomplete": "None",
+                }
+            },
+        }
+
+        response = await self.__graphql_query(headers, url, graphql_json)
+        if response.status == 200:
+            data = await response.json()
+            return (data.get("data", {}).get("places") or {}).get("results") or []
+        return []
+
+    async def suggest_places(
+        self, query: str, latitude: float, longitude: float
+    ) -> list[str]:
+        """Suggest search texts for partial text, for type-ahead in a place search."""
+        url = GRAPHQL_GATEWAY
+        headers = BASE_HEADERS | {
+            "Csrf-Token": self._csrf_token,
+            "A-Sess": self._app_session_token,
+            "U-Sess": self._user_session_token,
+        }
+        graphql_json = {
+            "operationName": "places",
+            "query": "query places($params: PlacesQueryParams!) { places(params: $params) { queryPredictions { query } } }",
+            "variables": {
+                "params": {
+                    "query": query,
+                    "userLocation": {"latitude": latitude, "longitude": longitude},
+                    "autocomplete": "Place",
+                }
+            },
+        }
+
+        response = await self.__graphql_query(headers, url, graphql_json)
+        if response.status == 200:
+            data = await response.json()
+            places = data.get("data", {}).get("places") or {}
+            return [p["query"] for p in places.get("queryPredictions") or []]
+        return []
+
     async def send_location_to_vehicle(
         self, vehicle_id: str, location: str
     ) -> ClientResponse:

src/rivian/utils.py

@@ -8,6 +8,7 @@
 from base64 import b64decode, b64encode
 from datetime import UTC, datetime
 from typing import cast
+from urllib.parse import urlencode
 from warnings import warn
 
 from cryptography.hazmat.primitives import hashes, serialization
@@ -143,3 +144,30 @@ def from_epoch(epoch: float) -> datetime:
     if abs(epoch) >= _MS_THRESHOLD:
         epoch /= 1000
     return datetime.fromtimestamp(epoch, UTC)
+
+
+def place_url(
+    place_id: str,
+    name: str,
+    latitude: float,
+    longitude: float,
+    address: str | None = None,
+) -> str:
+    """Return the Rivian app's link for a place, for `send_location_to_vehicle`.
+
+    The vehicle shows the place's name. `place_id` is a Google place id, such as a
+    `search_places` result's `id`; Rivian refuses these links without one. Use the
+    result's `navCoordinate`, where the drive ends.
+    """
+    coordinates = f"{latitude},{longitude}"
+    query = {
+        "geo_coordinates": coordinates,
+        "entity_id": place_id,
+        "name": name,
+        "nav_coordinates": coordinates,
+        "address": address,
+        "source_type": "telenav",
+    }
+    return "https://telenav.id/entity?" + urlencode(
+        {k: v for k, v in query.items() if v is not None}
+    )

tests/responses.py

@@ -654,6 +654,25 @@
     "data": {"sendParallaxPayload": {"success": True, "sequenceNumber": 0}}
 }
 
+PLACES_RESPONSE = {
+    "data": {
+        "places": {
+            "results": [
+                {
+                    "id": "place_id",
+                    "name": "REI",
+                    "location": {
+                        "formattedAddress": "5901 Mills Civic Pkwy, West Des Moines, IA",
+                        "coordinate": {"latitude": 41.56028, "longitude": -93.78724},
+                        "navCoordinate": {"latitude": 41.56001, "longitude": -93.7873},
+                    },
+                    "driveDetails": {"driveDistance": 239062, "driveDuration": 9960},
+                }
+            ]
+        }
+    }
+}
+
 SEND_VEHICLE_OPERATION_RESPONSE = {
     "data": {
         "sendVehicleOperation": {

tests/test_rivian.py

@@ -42,6 +42,7 @@
     DELETE_DEPARTURE_SCHEDULE_RESPONSE,
     DEPART_NOW_RESPONSE,
     OTP_TOKEN_RESPONSE,
+    PLACES_RESPONSE,
     SEND_LOCATION_TO_VEHICLE_RESPONSE,
     SEND_PARALLAX_COMMAND_RESPONSE,
     SEND_VEHICLE_OPERATION_RESPONSE,
@@ -447,6 +448,49 @@ async def test_send_location_to_vehicle(aresponses: ResponsesMockServer) -> None
         await rivian.close()
 
 
+async def test_search_places(aresponses: ResponsesMockServer) -> None:
+    """Test searching for places."""
+    requests = capture_requests(aresponses, PLACES_RESPONSE, {"data": {"places": None}})
+    async with aiohttp.ClientSession():
+        rivian = Rivian(
+            csrf_token="csrf",
+            app_session_token="token",
+            user_session_token="token",
+        )
+
+        results = await rivian.search_places("REI", 40.175, -92.591)
+        assert results == PLACES_RESPONSE["data"]["places"]["results"]
+        assert requests[0]["variables"]["params"] == {
+            "query": "REI",
+            "userLocation": {"latitude": 40.175, "longitude": -92.591},
+            "autocomplete": "None",
+        }
+        assert await rivian.search_places("REI", 40.175, -92.591) == []
+
+
+async def test_suggest_places(aresponses: ResponsesMockServer) -> None:
+    """Test suggesting search texts for partial text."""
+    predictions = [{"query": "restaurants"}, {"query": "Red Lobster"}]
+    requests = capture_requests(
+        aresponses,
+        {"data": {"places": {"queryPredictions": predictions}}},
+        {"data": {"places": None}},
+    )
+    async with aiohttp.ClientSession():
+        rivian = Rivian(
+            csrf_token="csrf",
+            app_session_token="token",
+            user_session_token="token",
+        )
+
+        assert await rivian.suggest_places("RE", 40.175, -92.591) == [
+            "restaurants",
+            "Red Lobster",
+        ]
+        assert requests[0]["variables"]["params"]["autocomplete"] == "Place"
+        assert await rivian.suggest_places("RE", 40.175, -92.591) == []
+
+
 async def test_subscribe_for_departure_schedules() -> None:
     """Test subscribing to departure schedules."""
     rivian = Rivian(user_session_token="token")

tests/test_utils.py

@@ -48,3 +48,12 @@ def test_generate_vehicle_command_hmac() -> None:
         command, timestamp, VEHICLE_KEY, PRIVATE_KEY
     )
     assert hmac == "2a68bdda69ff8643e37bac595905f6a481435e00bb63bdd415ecbb425a5bb598"
+
+
+def test_place_url() -> None:
+    """A place link carries the place's name, id, address and coordinates."""
+    assert utils.place_url("place_id", "REI", 41.5, -93.7, "1 Main St") == (
+        "https://telenav.id/entity?geo_coordinates=41.5%2C-93.7&entity_id=place_id"
+        "&name=REI&nav_coordinates=41.5%2C-93.7&address=1+Main+St&source_type=telenav"
+    )
+    assert "address" not in utils.place_url("place_id", "REI", 41.5, -93.7)