Add remote services for EVs (#520)

* Add remote service to set charging settings (charging target)
* Fix CLI
* Skip UNKOWN in cli image
* Implement only setting target SoC
* Rename response files
* Format responses
* Fix loading of fingerprints for tests
* Add charging settings request to account and tests
* Parse required charging settings to charging profile
* fix py37 & py38
* Allow setting both AC Limit & Target SoC
* Fix VIN for G26
* Refactor remote services
* Add Charging Profile remote service
* Add CLI for charging profile
* Check if remote service is enabled before execution
* Add attribute if start/stop charging is enabled
* Update Charging URLs
* Update tests for remote service capability check
* Add charge start/stop remote service
* Add CLI for charge start/stop
* Add debug to remote services
* Remove unneeded double request
* Fix linting
* Add milliseconds to format for remote service
* Use correct mocked endpoints
* Fix SendPoi & add tests for parsing content
* Fix sending JSON data
* Fix status for SendPoi
* Fix charging profile tests
* Update pre-commit
* Fix mypy
* Fix brand missing on response files
* Fix typos

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.0.241
+    rev: v0.0.251
     hooks:
       - id: ruff
         args:
@@ -24,7 +24,7 @@ repos:
         exclude_types: [csv, json]
         exclude: ^test/responses/
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.991
+    rev: v1.0.1
     hooks: 
       - id: mypy
         name: mypy

bimmer_connected/account.py

@@ -10,7 +10,13 @@
 from bimmer_connected.api.authentication import MyBMWAuthentication
 from bimmer_connected.api.client import RESPONSE_STORE, MyBMWClient, MyBMWClientConfiguration
 from bimmer_connected.api.regions import Regions
-from bimmer_connected.const import VEHICLE_STATE_URL, VEHICLES_URL, CarBrands
+from bimmer_connected.const import (
+    ATTR_CAPABILITIES,
+    VEHICLE_CHARGING_DETAILS_URL,
+    VEHICLE_STATE_URL,
+    VEHICLES_URL,
+    CarBrands,
+)
 from bimmer_connected.models import AnonymizedResponse, GPSPosition
 from bimmer_connected.utils import deprecated
 from bimmer_connected.vehicle import MyBMWVehicle
@@ -76,7 +82,7 @@ async def _init_vehicles(self) -> None:
 
             for response in vehicles_responses:
                 for vehicle_base in response.json():
-                    self.add_vehicle(vehicle_base, {}, fetched_at)
+                    self.add_vehicle(vehicle_base, None, None, fetched_at)
 
     async def get_vehicles(self, force_init: bool = False) -> None:
         """Retrieve vehicle data from BMW servers."""
@@ -99,20 +105,46 @@ async def get_vehicles(self, force_init: bool = False) -> None:
                     },
                 )
                 vehicle_state = state_response.json()
-                self.add_vehicle(vehicle.data, vehicle_state, fetched_at)
+
+                # Get detailed charging settings if supported by vehicle
+                charging_settings = None
+                if vehicle_state[ATTR_CAPABILITIES].get("isChargingPlanSupported", False) or vehicle_state[
+                    ATTR_CAPABILITIES
+                ].get("isChargingSettingsEnabled", False):
+                    charging_settings_response = await client.get(
+                        VEHICLE_CHARGING_DETAILS_URL,
+                        params={
+                            "fields": "charging-profile",
+                            "has_charging_settings_capabilities": vehicle_state[ATTR_CAPABILITIES][
+                                "isChargingSettingsEnabled"
+                            ],
+                        },
+                        headers={
+                            **client.generate_default_header(vehicle.brand),
+                            "bmw-current-date": fetched_at.isoformat(),
+                            "bmw-vin": vehicle.vin,
+                        },
+                    )
+                    charging_settings = charging_settings_response.json()
+
+                self.add_vehicle(vehicle.data, vehicle_state, charging_settings, fetched_at)
 
     def add_vehicle(
-        self, vehicle_base: dict, vehicle_state: dict, fetched_at: Optional[datetime.datetime] = None
+        self,
+        vehicle_base: dict,
+        vehicle_state: Optional[dict],
+        charging_settings: Optional[dict],
+        fetched_at: Optional[datetime.datetime] = None,
     ) -> None:
         """Add or update a vehicle from the API responses."""
 
         existing_vehicle = self.get_vehicle(vehicle_base["vin"])
 
         # If vehicle already exists, just update it's state
         if existing_vehicle:
-            existing_vehicle.update_state(vehicle_base, vehicle_state, fetched_at)
+            existing_vehicle.update_state(vehicle_base, vehicle_state, charging_settings, fetched_at)
         else:
-            self.vehicles.append(MyBMWVehicle(self, vehicle_base, vehicle_state, fetched_at))
+            self.vehicles.append(MyBMWVehicle(self, vehicle_base, vehicle_state, charging_settings, fetched_at))
 
     def get_vehicle(self, vin: str) -> Optional[MyBMWVehicle]:
         """Get vehicle with given VIN.

bimmer_connected/api/utils.py

@@ -107,7 +107,7 @@ def anonymize_response(response: httpx.Response) -> AnonymizedResponse:
     brand = response.request.headers.get("x-user-agent", ";").split(";")[1]
     brand = f"{brand}-" if brand else ""
 
-    url_parts = response.url.path.split("/")[3:]
+    url_parts = response.url.path.split("/")[1:]
     if "bmw-vin" in response.request.headers:
         url_parts.append(response.request.headers["bmw-vin"])
     url_path = RE_VIN.sub(anonymize_vin, "_".join(url_parts))

bimmer_connected/cli.py

@@ -7,17 +7,16 @@
 import logging
 import sys
 import time
-from datetime import datetime
 from pathlib import Path
 
 import httpx
 
 from bimmer_connected.account import MyBMWAccount
-from bimmer_connected.api.client import MyBMWClient
 from bimmer_connected.api.regions import get_region_from_name, valid_regions
 from bimmer_connected.const import DEFAULT_POI_NAME
 from bimmer_connected.utils import MyBMWJSONEncoder, log_response_store_to_file
 from bimmer_connected.vehicle import MyBMWVehicle, VehicleViewDirection
+from bimmer_connected.vehicle.charging_profile import ChargingMode
 
 TEXT_VIN = "Vehicle Identification Number"
 
@@ -58,6 +57,34 @@ def main_parser() -> argparse.ArgumentParser:
     _add_position_arguments(finder_parser)
     finder_parser.set_defaults(func=vehicle_finder)
 
+    chargingsettings_parser = subparsers.add_parser("chargingsettings", description="Set vehicle charging settings.")
+    _add_default_arguments(chargingsettings_parser)
+    chargingsettings_parser.add_argument("vin", help=TEXT_VIN)
+    chargingsettings_parser.add_argument("--target-soc", help="Desired charging target SoC", nargs="?", type=int)
+    chargingsettings_parser.add_argument("--ac-limit", help="Maximum AC limit", nargs="?", type=int)
+    chargingsettings_parser.set_defaults(func=chargingsettings)
+
+    chargingprofile_parser = subparsers.add_parser("chargingprofile", description="Set vehicle charging profile.")
+    _add_default_arguments(chargingprofile_parser)
+    chargingprofile_parser.add_argument("vin", help=TEXT_VIN)
+    chargingprofile_parser.add_argument(
+        "--charging-mode",
+        help="Desired charging mode",
+        nargs="?",
+        type=ChargingMode,
+        choices=[cm.value for cm in ChargingMode if cm != ChargingMode.UNKNOWN],
+    )
+    chargingprofile_parser.add_argument(
+        "--precondition-climate", help="Precondition climate on charging windows", nargs="?", type=bool
+    )
+    chargingprofile_parser.set_defaults(func=chargingprofile)
+
+    charge_parser = subparsers.add_parser("charge", description="Start/stop charging on enabled vehicles.")
+    _add_default_arguments(charge_parser)
+    charge_parser.add_argument("vin", help=TEXT_VIN)
+    charge_parser.add_argument("action", type=str, choices=["start", "stop"])
+    charge_parser.set_defaults(func=charge)
+
     image_parser = subparsers.add_parser("image", description="Download a vehicle image.")
     _add_default_arguments(image_parser)
     image_parser.add_argument("vin", help=TEXT_VIN)
@@ -147,23 +174,6 @@ async def fingerprint(args) -> None:
         account.set_observer_position(args.lat, args.lng)
     await account.get_vehicles()
 
-    # Patching in new My BMW endpoints for fingerprinting
-    async with MyBMWClient(account.config) as client:
-        for vehicle in account.vehicles:
-            try:
-                if vehicle.has_electric_drivetrain:
-                    await client.get(
-                        f"/eadrax-crccs/v1/vehicles/{vehicle.vin}",
-                        params={"fields": "charging-profile", "has_charging_settings_capabilities": True},
-                        headers={
-                            **client.generate_default_header(vehicle.brand),
-                            "bmw-current-date": datetime.utcnow().isoformat(),
-                            "24-hour-format": "true",
-                        },
-                    )
-            except httpx.HTTPStatusError:
-                pass
-
     log_response_store_to_file(account.get_stored_responses(), time_dir)
     print(f"fingerprint of the vehicles written to {time_dir}")
 
@@ -203,6 +213,47 @@ async def vehicle_finder(args) -> None:
     print({"gps_position": vehicle.status.gps_position, "heading": vehicle.status.gps_heading})
 
 
+async def chargingsettings(args) -> None:
+    """Trigger a change to charging settings."""
+    if not args.target_soc and not args.ac_limit:
+        raise ValueError("At least one of 'charging-target' and 'ac-limit' has to be provided.")
+    account = MyBMWAccount(
+        args.username, args.password, get_region_from_name(args.region), use_metric_units=(not args.imperial)
+    )
+    await account.get_vehicles()
+    vehicle = get_vehicle_or_return(account, args.vin)
+    status = await vehicle.remote_services.trigger_charging_settings_update(
+        target_soc=args.target_soc, ac_limit=args.ac_limit
+    )
+    print(status.state)
+
+
+async def chargingprofile(args) -> None:
+    """Trigger a change to charging profile."""
+    if not args.charging_mode and not args.precondition_climate:
+        raise ValueError("At least one of 'charging-mode' and 'precondition-climate' has to be provided.")
+    account = MyBMWAccount(
+        args.username, args.password, get_region_from_name(args.region), use_metric_units=(not args.imperial)
+    )
+    await account.get_vehicles()
+    vehicle = get_vehicle_or_return(account, args.vin)
+    status = await vehicle.remote_services.trigger_charging_profile_update(
+        charging_mode=args.charging_mode, precondition_climate=args.precondition_climate
+    )
+    print(status.state)
+
+
+async def charge(args) -> None:
+    """Trigger a vehicle to start or stop charging."""
+    account = MyBMWAccount(
+        args.username, args.password, get_region_from_name(args.region), use_metric_units=(not args.imperial)
+    )
+    await account.get_vehicles()
+    vehicle = get_vehicle_or_return(account, args.vin)
+    status = await getattr(vehicle.remote_services, f"trigger_charge_{args.action.lower()}")()
+    print(status.state)
+
+
 async def image(args) -> None:
     """Download a rendered image of the vehicle."""
     account = MyBMWAccount(
@@ -212,6 +263,8 @@ async def image(args) -> None:
     vehicle = get_vehicle_or_return(account, args.vin)
 
     for viewdirection in VehicleViewDirection:
+        if viewdirection == VehicleViewDirection.UNKNOWN:
+            continue
         filename = str(viewdirection.name).lower() + ".png"
         with open(filename, "wb") as output_file:
             image_data = await vehicle.get_vehicle_image(viewdirection)

bimmer_connected/const.py

@@ -69,6 +69,12 @@ class Regions(str, Enum):
 REMOTE_SERVICE_STATUS_URL = REMOTE_SERVICE_BASE_URL + "/eventStatus?eventId={event_id}"
 REMOTE_SERVICE_POSITION_URL = REMOTE_SERVICE_BASE_URL + "/eventPosition?eventId={event_id}"
 
+VEHICLE_CHARGING_DETAILS_URL = "/eadrax-crccs/v2/vehicles"
+VEHICLE_CHARGING_BASE_URL = "/eadrax-crccs/v1/vehicles/{vin}"
+VEHICLE_CHARGING_SETTINGS_SET_URL = VEHICLE_CHARGING_BASE_URL + "/charging-settings"
+VEHICLE_CHARGING_PROFILE_SET_URL = VEHICLE_CHARGING_BASE_URL + "/charging-profile"
+VEHICLE_CHARGING_START_STOP_URL = VEHICLE_CHARGING_BASE_URL + "/{service_type}"
+
 VEHICLE_IMAGE_URL = "/eadrax-ics/v3/presentation/vehicles/{vin}/images?carView={view}"
 VEHICLE_POI_URL = "/eadrax-dcs/v1/send-to-car/send-to-car"
 
@@ -83,5 +89,6 @@ class Regions(str, Enum):
 ATTR_STATE = "state"
 ATTR_CAPABILITIES = "capabilities"
 ATTR_ATTRIBUTES = "attributes"
+ATTR_CHARGING_SETTINGS = "charging_settings"
 
 DEFAULT_POI_NAME = "Sent with ♥ by bimmer_connected"

bimmer_connected/models.py

@@ -133,3 +133,13 @@ class AnonymizedResponse:
 
     filename: str
     content: Optional[Union[List, Dict, str]] = None
+
+
+@dataclass
+class ChargingSettings:
+    """Charging settings to control the vehicle."""
+
+    chargingTarget: Optional[int]
+    isUnlockCableActive = None
+    acLimitValue: Optional[int] = None
+    dcLoudness = None

bimmer_connected/vehicle/charging_profile.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
 
-from bimmer_connected.const import ATTR_STATE
+from bimmer_connected.const import ATTR_CHARGING_SETTINGS, ATTR_STATE
 from bimmer_connected.models import StrEnum, VehicleDataBase
 
 _LOGGER = logging.getLogger(__name__)
@@ -19,6 +19,12 @@ class ChargingMode(StrEnum):
     UNKNOWN = "UNKNOWN"
 
 
+MAP_CHARGING_MODE_TO_REMOTE_SERVICE = {
+    ChargingMode.IMMEDIATE_CHARGING: "CHARGING_IMMEDIATELY",
+    ChargingMode.DELAYED_CHARGING: "TIME_SLOT",
+}
+
+
 class ChargingPreferences(StrEnum):
     """Charging preferences of electric vehicle."""
 
@@ -30,13 +36,17 @@ class ChargingPreferences(StrEnum):
 class TimerTypes(StrEnum):
     """Different timer types."""
 
-    TWO_WEEKS = "TWO_WEEKS_TIMER"
-    ONE_WEEK = "WEEKLY_PLANNER"
-    OVERRIDE_TIMER = "OVERRIDE_TIMER"
+    WEEKLY_PLANNER = "WEEKLY_PLANNER"
     TWO_TIMES_TIMER = "TWO_TIMES_TIMER"
     UNKNOWN = "UNKNOWN"
 
 
+MAP_TIMER_TYPES_TO_REMOTE_SERVICE = {
+    TimerTypes.WEEKLY_PLANNER: "WEEKLY_DEPARTURE_TIMER",
+    TimerTypes.TWO_TIMES_TIMER: "TWO_DEPARTURE_TIMER",
+}
+
+
 class ChargingWindow:
     """A charging window."""
 
@@ -46,11 +56,15 @@ def __init__(self, window_dict: dict):
     @property
     def start_time(self) -> datetime.time:
         """Start of the charging window."""
+        if "start" not in self._window_dict:
+            return datetime.time(0, 0)
         return datetime.time(int(self._window_dict["start"]["hour"]), int(self._window_dict["start"]["minute"]))
 
     @property
     def end_time(self) -> datetime.time:
         """End of the charging window."""
+        if "end" not in self._window_dict:
+            return datetime.time(0, 0)
         return datetime.time(int(self._window_dict["end"]["hour"]), int(self._window_dict["end"]["minute"]))
 
 
@@ -108,6 +122,12 @@ class ChargingProfile(VehicleDataBase):
     ac_current_limit: Optional[int] = None
     """Returns the ac current limit."""
 
+    ac_available_limits: Optional[list] = None
+    """Available AC limits to be selected."""
+
+    charging_preferences_service_pack: Optional[str] = None
+    """Service Pack required for remote service format."""
+
     @classmethod
     def _parse_vehicle_data(cls, vehicle_data: Dict) -> Dict:
         """Parse charging data."""
@@ -125,4 +145,47 @@ def _parse_vehicle_data(cls, vehicle_data: Dict) -> Dict:
             if "acCurrentLimit" in charging_profile["chargingSettings"]:
                 retval["ac_current_limit"] = charging_profile["chargingSettings"]["acCurrentLimit"]
 
+        if ATTR_CHARGING_SETTINGS in vehicle_data:
+            charging_settings = vehicle_data[ATTR_CHARGING_SETTINGS]
+            if "servicePack" in charging_settings:
+                retval["charging_preferences_service_pack"] = charging_settings["servicePack"]
+            if (
+                "chargingSettingsDetail" in charging_settings
+                and "acLimit" in charging_settings["chargingSettingsDetail"]
+            ):
+                retval["ac_available_limits"] = charging_settings["chargingSettingsDetail"]["acLimit"]["values"]
+
         return retval
+
+    def format_for_remote_service(self) -> dict:
+        """Format current charging profile as base to be sent to remote service."""
+
+        return {
+            "chargingMode": {
+                "chargingPreference": self.charging_preferences.value,
+                "endTimeSlot": self._format_time(self.preferred_charging_window.end_time),
+                "startTimeSlot": self._format_time(self.preferred_charging_window.start_time),
+                "type": MAP_CHARGING_MODE_TO_REMOTE_SERVICE[self.charging_mode],
+                "timerChange": "NO_CHANGE",
+            },
+            "departureTimer": {
+                "type": MAP_TIMER_TYPES_TO_REMOTE_SERVICE[self.timer_type],
+                "weeklyTimers": [
+                    {
+                        "daysOfTheWeek": t.weekdays,
+                        "id": t.timer_id,
+                        "time": self._format_time(t.start_time),
+                        "timerAction": t.action,
+                    }
+                    for t in self.departure_times
+                ],
+            },
+            "isPreconditionForDepartureActive": self.is_pre_entry_climatization_enabled,
+            "servicePack": self.charging_preferences_service_pack,
+        }
+
+    @staticmethod
+    def _format_time(time: Optional[datetime.time] = None) -> str:
+        if not time:
+            return "0001-01-01T00:00:00.000"
+        return time.strftime("0001-01-01T%H:%M:00.000")