diff --git a/src/rivian/rivian.py b/src/rivian/rivian.py index 0d0ebc6..eb99ff1 100644 --- a/src/rivian/rivian.py +++ b/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: diff --git a/src/rivian/utils.py b/src/rivian/utils.py index 1bfff46..d94ac55 100644 --- a/src/rivian/utils.py +++ b/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} + ) diff --git a/tests/responses.py b/tests/responses.py index 53e98e9..c885a93 100644 --- a/tests/responses.py +++ b/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": { diff --git a/tests/test_rivian.py b/tests/test_rivian.py index 634571c..580a939 100644 --- a/tests/test_rivian.py +++ b/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") diff --git a/tests/test_utils.py b/tests/test_utils.py index 06fdf4b..0f32d75 100644 --- a/tests/test_utils.py +++ b/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)