WIP: Lexus Work

mytoyota/api.py

@@ -49,17 +49,20 @@ def __init__(self, controller: Controller) -> None:
         """
         self.controller = controller
 
-    async def _request_and_parse(self, model, method: str, endpoint: str, **kwargs):
+    async def _request_and_parse(self, model, method: str, endpoint: str, brand: str, **kwargs):
         """Parse requests and responses."""
-        response = await self.controller.request_json(method=method, endpoint=endpoint, **kwargs)
+        response = await self.controller.request_json(
+            method=method, endpoint=endpoint, brand=brand, **kwargs
+        )
         return model(**response)
 
-    async def set_vehicle_alias_endpoint(self, alias: str, guid: str, vin: str):
+    async def set_vehicle_alias_endpoint(self, alias: str, guid: str, vin: str, brand: str):
         """Set the alias for a vehicle."""
         return await self.controller.request_raw(
             method="PUT",
             endpoint=VEHICLE_ASSOCIATION_ENDPOINT,
             vin=vin,
+            brand=brand,
             headers={
                 "datetime": str(int(datetime.now(timezone.utc).timestamp() * 1000)),
                 "x-correlationid": str(uuid4()),
@@ -76,73 +79,81 @@ async def set_vehicle_alias_endpoint(self, alias: str, guid: str, vin: str):
     #            method="POST", endpoint="/v2/global/remote/wake"
     #        )
 
-    async def get_vehicles_endpoint(self) -> VehiclesResponseModel:
+    async def get_vehicles_endpoint(self, brand: str = "T") -> VehiclesResponseModel:
         """Return list of vehicles registered with provider."""
         parsed_response = await self._request_and_parse(
-            VehiclesResponseModel, "GET", VEHICLE_GUID_ENDPOINT
+            VehiclesResponseModel, "GET", VEHICLE_GUID_ENDPOINT, brand
         )
         _LOGGER.debug(msg=f"Parsed 'VehiclesResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_location_endpoint(self, vin: str) -> LocationResponseModel:
+    async def get_location_endpoint(self, vin: str, brand: str) -> LocationResponseModel:
         """Get the last known location of your car. Only updates when car is parked.
 
         Response includes Lat, Lon position. * If supported.
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
             LocationResponseModel: A pydantic model for the location response
         """
         parsed_response = await self._request_and_parse(
-            LocationResponseModel, "GET", VEHICLE_LOCATION_ENDPOINT, vin=vin
+            LocationResponseModel, "GET", VEHICLE_LOCATION_ENDPOINT, vin=vin, brand=brand
         )
         _LOGGER.debug(msg=f"Parsed 'LocationResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_vehicle_health_status_endpoint(self, vin: str) -> VehicleHealthResponseModel:
+    async def get_vehicle_health_status_endpoint(
+        self, vin: str, brand: str
+    ) -> VehicleHealthResponseModel:
         r"""Get the latest health status.
 
         Response includes the quantity of engine oil and any dashboard warning lights. \n
         * If supported.
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
             VehicleHealthResponseModel: A pydantic model for the vehicle health response
         """
         parsed_response = await self._request_and_parse(
-            VehicleHealthResponseModel, "GET", VEHICLE_HEALTH_STATUS_ENDPOINT, vin=vin
+            VehicleHealthResponseModel, "GET", VEHICLE_HEALTH_STATUS_ENDPOINT, vin=vin, brand=brand
         )
         _LOGGER.debug(msg=f"Parsed 'VehicleHealthResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_remote_status_endpoint(self, vin: str) -> RemoteStatusResponseModel:
+    async def get_remote_status_endpoint(self, vin: str, brand: str) -> RemoteStatusResponseModel:
         """Get information about the vehicle."""
         parsed_response = await self._request_and_parse(
             RemoteStatusResponseModel,
             "GET",
             VEHICLE_GLOBAL_REMOTE_STATUS_ENDPOINT,
             vin=vin,
+            brand=brand,
         )
         _LOGGER.debug(msg=f"Parsed 'RemoteStatusResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_vehicle_electric_status_endpoint(self, vin: str) -> ElectricResponseModel:
+    async def get_vehicle_electric_status_endpoint(
+        self, vin: str, brand: str
+    ) -> ElectricResponseModel:
         r"""Get the latest electric status.
 
         Response includes current battery level, EV Range, EV Range with AC, \n
         fuel level, fuel range and current charging status
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
@@ -153,30 +164,32 @@ async def get_vehicle_electric_status_endpoint(self, vin: str) -> ElectricRespon
             "GET",
             VEHICLE_GLOBAL_REMOTE_ELECTRIC_STATUS_ENDPOINT,
             vin=vin,
+            brand=brand,
         )
         _LOGGER.debug(msg=f"Parsed 'ElectricResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_telemetry_endpoint(self, vin: str) -> TelemetryResponseModel:
+    async def get_telemetry_endpoint(self, vin: str, brand: str) -> TelemetryResponseModel:
         """Get the latest telemetry status.
 
         Response includes current fuel level, distance to empty and odometer
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
             TelemetryResponseModel: A pydantic model for the telemetry response
         """
         parsed_response = await self._request_and_parse(
-            TelemetryResponseModel, "GET", VEHICLE_TELEMETRY_ENDPOINT, vin=vin
+            TelemetryResponseModel, "GET", VEHICLE_TELEMETRY_ENDPOINT, vin=vin, brand=brand
         )
         _LOGGER.debug(msg=f"Parsed 'TelemetryResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_notification_endpoint(self, vin: str) -> NotificationResponseModel:
+    async def get_notification_endpoint(self, vin: str, brand: str) -> NotificationResponseModel:
         """Get all available notifications for the vehicle.
 
         A notification includes a message, notification date, read flag, date read.
@@ -185,7 +198,8 @@ async def get_notification_endpoint(self, vin: str) -> NotificationResponseModel
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
@@ -196,13 +210,15 @@ async def get_notification_endpoint(self, vin: str) -> NotificationResponseModel
             "GET",
             VEHICLE_NOTIFICATION_HISTORY_ENDPOINT,
             vin=vin,
+            brand=brand,
         )
         _LOGGER.debug(msg=f"Parsed 'NotificationResponseModel': {parsed_response}")
         return parsed_response
 
     async def get_trips_endpoint(  # noqa: PLR0913
         self,
         vin: str,
+        brand: str,
         from_date: date,
         to_date: date,
         route: bool = False,
@@ -219,6 +235,7 @@ async def get_trips_endpoint(  # noqa: PLR0913
         Args:
         ----
             vin: str:        The vehicles VIN
+            brand: str:      The car brand used for the request.
             from_date: date: From date to include trips, inclusive. Cant be in the future.
             to_date: date:   To date to include trips, inclusive. Cant be in the future.
             route: bool:     If true returns the route of each trip as a list of coordinates.
@@ -240,26 +257,33 @@ async def get_trips_endpoint(  # noqa: PLR0913
             offset=offset,
         )
         parsed_response = await self._request_and_parse(
-            TripsResponseModel, "GET", endpoint, vin=vin
+            TripsResponseModel, "GET", endpoint, vin=vin, brand=brand
         )
         _LOGGER.debug(msg=f"Parsed 'TripsResponseModel': {parsed_response}")
         return parsed_response
 
-    async def get_service_history_endpoint(self, vin: str) -> ServiceHistoryResponseModel:
+    async def get_service_history_endpoint(
+        self, vin: str, brand: str
+    ) -> ServiceHistoryResponseModel:
         """Get the current servic history.
 
         Response includes service category, date and dealer.
 
         Args:
         ----
-            vin: str:   The vehicles VIN
+            vin: (str): The vehicles VIN
+            brand (str): The car brand used for the request.
 
         Returns:
         -------
             ServicHistoryResponseModel: A pydantic model for the service history response
         """
         parsed_response = await self._request_and_parse(
-            ServiceHistoryResponseModel, "GET", VEHICLE_SERVICE_HISTORY_ENDPONT, vin=vin
+            ServiceHistoryResponseModel,
+            "GET",
+            VEHICLE_SERVICE_HISTORY_ENDPONT,
+            vin=vin,
+            brand=brand,
         )
         _LOGGER.debug(msg=f"Parsed 'ServiceHistoryResponseModel': {parsed_response}")
         return parsed_response

mytoyota/client.py

@@ -58,10 +58,10 @@
         _LOGGER.debug("Performing first login")
         await self._api.controller.login()
 
-    async def get_vehicles(self, metric: bool = True) -> Optional[List[Vehicle]]:
+    async def get_vehicles(self, metric: bool = True, brand: str = "T") -> Optional[List[Vehicle]]:
         """Return a list of vehicles."""
         _LOGGER.debug("Getting list of vehicles associated with the account")
-        vehicles = await self._api.get_vehicles_endpoint()
+        vehicles = await self._api.get_vehicles_endpoint(brand)
         if vehicles.payload is not None:
             return [Vehicle(self._api, v, metric) for v in vehicles.payload]
 

mytoyota/controller.py

@@ -35,6 +35,19 @@
 #      This seems to work sometimes but no others. Needs investigation.
 
 
+def get_brand_headers(headers: Dict[str, Any], brand: str) -> Dict[str, Any]:
+    """Update the headers, depending on the given car brand."""
+    if brand == "L":
+        headers |= {
+            "x-appbrand": "L",
+            "brand": "L",
+            "x-brand": "L",
+        }
+    elif brand == "T":
+        headers["x-brand"] = "T"
+    return headers
+
+
 class Controller:
     """Controller class."""
 
@@ -215,6 +228,7 @@
         self,
         method: str,
         endpoint: str,
+        brand: str = "T",
         vin: Optional[str] = None,
         body: Optional[Dict[str, Any]] = None,
         params: Optional[Dict[str, Any]] = None,
@@ -236,10 +250,10 @@
                 "guid": self._uuid,
                 "authorization": f"Bearer {self._token}",
                 "x-channel": "ONEAPP",
-                "x-brand": "T",
                 "user-agent": "okhttp/4.10.0",
             },
         )
+        headers = get_brand_headers(headers, brand)
         # Add vin if passed
         if vin is not None:
             headers.update({"vin": vin})
@@ -266,6 +280,7 @@
         self,
         method: str,
         endpoint: str,
+        brand: str = "T",
         vin: Optional[str] = None,
         body: Optional[Dict[str, Any]] = None,
         params: Optional[Dict[str, Any]] = None,
@@ -279,6 +294,8 @@
             endpoint (str): The endpoint to send the request to.
             vin (Optional[str], optional): The VIN (Vehicle Identification Number) to include
                 in the request. Defaults to None.
+            brand (str): The car brand used for the request.
+                Defaults to Toyota.
             body (Optional[Dict[str, Any]], optional): The JSON body to include in the request.
                 Defaults to None.
             params (Optional[Dict[str, Any]], optional): The query parameters to
@@ -294,6 +311,6 @@
         --------
             response = await request_json("GET", "/cars", vin="1234567890")
         """
-        response = await self.request_raw(method, endpoint, vin, body, params, headers)
+        response = await self.request_raw(method, endpoint, brand, vin, body, params, headers)
 
         return response.json()

mytoyota/models/endpoints/vehicle_guid.py

@@ -390,7 +390,7 @@ class VehicleGuidModel(BaseModel):
     head_unit: _HeadUnitModel = Field(alias="headUnit")
     hw_type: Optional[Any] = Field(alias="hwType")  # TODO unsure what this returns
     image: str
-    imei: str
+    imei: Optional[str] = None
     katashiki_code: str = Field(alias="katashikiCode")
     manufactured_date: date = Field(alias="manufacturedDate")
     manufactured_code: str = Field(alias="manufacturerCode")

mytoyota/models/vehicle.py

@@ -43,14 +43,17 @@ def __init__(self, api: Api, vehicle_info: VehicleGuidModel, metric: bool = True
                 "name": "location",
                 "capable": vehicle_info.extended_capabilities.last_parked_capable
                 or vehicle_info.features.last_parked,
-                "function": partial(self._api.get_location_endpoint, vin=vehicle_info.vin),
+                "function": partial(
+                    self._api.get_location_endpoint, vin=vehicle_info.vin, brand=vehicle_info.brand
+                ),
             },
             {
                 "name": "health_status",
                 "capable": True,  # TODO Unsure of the required capability
                 "function": partial(
                     self._api.get_vehicle_health_status_endpoint,
                     vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
                 ),
             },
             {
@@ -59,27 +62,44 @@ def __init__(self, api: Api, vehicle_info: VehicleGuidModel, metric: bool = True
                 "function": partial(
                     self._api.get_vehicle_electric_status_endpoint,
                     vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
                 ),
             },
             {
                 "name": "telemetry",
                 "capable": vehicle_info.extended_capabilities.telemetry_capable,
-                "function": partial(self._api.get_telemetry_endpoint, vin=vehicle_info.vin),
+                "function": partial(
+                    self._api.get_telemetry_endpoint,
+                    vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
+                ),
             },
             {
                 "name": "notifications",
                 "capable": True,  # TODO Unsure of the required capability
-                "function": partial(self._api.get_notification_endpoint, vin=vehicle_info.vin),
+                "function": partial(
+                    self._api.get_notification_endpoint,
+                    vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
+                ),
             },
             {
                 "name": "status",
                 "capable": vehicle_info.extended_capabilities.vehicle_status,
-                "function": partial(self._api.get_remote_status_endpoint, vin=vehicle_info.vin),
+                "function": partial(
+                    self._api.get_remote_status_endpoint,
+                    vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
+                ),
             },
             {
                 "name": "service_history",
                 "capable": vehicle_info.features.service_history,
-                "function": partial(self._api.get_service_history_endpoint, vin=vehicle_info.vin),
+                "function": partial(
+                    self._api.get_service_history_endpoint,
+                    vin=vehicle_info.vin,
+                    brand=vehicle_info.brand,
+                ),
             },
         ]
         self._endpoint_collect = [

simple_client_example.py

@@ -47,7 +47,7 @@ async def get_information():
     await client.login()
 
     print("Retrieving cars...")
-    cars = await client.get_vehicles(metric=True)
+    cars = await client.get_vehicles(metric=True, brand="L")
 
     for car in cars:
         await car.update()
@@ -85,6 +85,4 @@ async def get_information():
         # pp.pprint(car._dump_all())
 
 
-loop = asyncio.get_event_loop()
-loop.run_until_complete(get_information())
-loop.close()
+asyncio.run(get_information())