Skip to content

Commit

Permalink
Migrate to vehicles v2 API (#462)
Browse files Browse the repository at this point in the history
* Migrate to vehicles v2 API

* Make pytest run at all on Py<3.9

* Don't fail on UNKOWN enum value

* Adjustments for pytest

* Updates for charging start/end time

* Fix zoneinfo test issue

* Update tests for charging start/end time

* Update fingerprint to only use new endpoints

* Update fingerprint to use vehicle brand

* Remove special ZoneInfo handling due to time_machine bugfix

* Apply time_machine fix only for Py>=3.7

* Fix time_machine requirements on Python 3.6
  • Loading branch information
rikroe authored Jun 25, 2022
1 parent d751fea commit a1e6c36
Show file tree
Hide file tree
Showing 53 changed files with 2,255 additions and 6,542 deletions.
3 changes: 2 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,5 @@ AUTHORS
ChangeLog
/.eggs
build/
htmlcov/
htmlcov/
.DS_Store
28 changes: 20 additions & 8 deletions bimmer_connected/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from bimmer_connected.api.authentication import MyBMWAuthentication
from bimmer_connected.api.client import MyBMWClient, MyBMWClientConfiguration
from bimmer_connected.api.regions import Regions
from bimmer_connected.const import VEHICLES_URL, CarBrands
from bimmer_connected.const import VEHICLE_STATE_URL, VEHICLES_URL, CarBrands
from bimmer_connected.models import GPSPosition
from bimmer_connected.utils import deprecated
from bimmer_connected.vehicle import MyBMWVehicle
Expand Down Expand Up @@ -61,23 +61,35 @@ async def get_vehicles(self) -> None:
"""Retrieve vehicle data from BMW servers."""
_LOGGER.debug("Getting vehicle list")

fetched_at = datetime.datetime.now(datetime.timezone.utc)

async with MyBMWClient(self.config) as client:
vehicles_request_params = {
"apptimezone": self.utcdiff,
"appDateTime": int(datetime.datetime.now().timestamp() * 1000),
"tireGuardMode": "ENABLED",
}
vehicles_responses: List[httpx.Response] = [
await client.get(
VEHICLES_URL,
params=vehicles_request_params,
headers=client.generate_default_header(brand),
headers={
**client.generate_default_header(brand),
"bmw-current-date": fetched_at.isoformat(),
},
)
for brand in CarBrands
]

for response in vehicles_responses:
for vehicle_dict in response.json():
# Get the detailed vehicle state
state_response = await client.get(
VEHICLE_STATE_URL.format(vin=vehicle_dict["vin"]),
headers=response.request.headers, # Reuse the same headers as used to get vehicle list
)

# Add state information to vehicle_dict
vehicle_dict.update(state_response.json())

# Add unit information
vehicle_dict["is_metric"] = self.config.use_metric_units
vehicle_dict["fetched_at"] = fetched_at

# If vehicle already exists, just update it's state
existing_vehicle = self.get_vehicle(vehicle_dict["vin"])
if existing_vehicle:
Expand Down
1 change: 1 addition & 0 deletions bimmer_connected/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,4 +82,5 @@ def generate_default_header(self, brand: CarBrands = None) -> Dict[str, str]:
),
**get_correlation_id(),
"bmw-units-preferences": "d=KM;v=L" if self.config.use_metric_units else "d=MI;v=G",
"24-hour-format": "true",
}
22 changes: 9 additions & 13 deletions bimmer_connected/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,8 @@
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 CarBrands
from bimmer_connected.utils import MyBMWJSONEncoder
from bimmer_connected.vehicle import MyBMWVehicle, VehicleViewDirection
from bimmer_connected.vehicle.vehicle import HV_BATTERY_DRIVE_TRAINS, DriveTrainType

TEXT_VIN = "Vehicle Identification Number"

Expand Down Expand Up @@ -136,20 +134,18 @@ async def fingerprint(args) -> None:
await account.get_vehicles()

# Patching in new My BMW endpoints for fingerprinting
async with MyBMWClient(account.config, brand=CarBrands.BMW) as client:
vehicles_v2 = await client.get("/eadrax-vcs/v2/vehicles")

for vehicle in vehicles_v2.json():
await client.get(
f"/eadrax-vcs/v2/vehicles/{vehicle['vin']}/state",
headers={"bmw-current-date": datetime.utcnow().isoformat(), "24-hour-format": "true"},
)
async with MyBMWClient(account.config) as client:
for vehicle in account.vehicles:
try:
if DriveTrainType(vehicle["attributes"]["driveTrain"]) in HV_BATTERY_DRIVE_TRAINS:
if vehicle.has_electric_drivetrain:
await client.get(
f"/eadrax-crccs/v1/vehicles/{vehicle['vin']}",
f"/eadrax-crccs/v1/vehicles/{vehicle.vin}",
params={"fields": "charging-profile", "has_charging_settings_capabilities": True},
headers={"bmw-current-date": datetime.utcnow().isoformat(), "24-hour-format": "true"},
headers={
**client.generate_default_header(vehicle.brand),
"bmw-current-date": datetime.utcnow().isoformat(),
"24-hour-format": "true",
},
)
except httpx.HTTPStatusError:
pass
Expand Down
11 changes: 8 additions & 3 deletions bimmer_connected/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ class CarBrands(str, Enum):

@classmethod
def _missing_(cls, value):
value = next(iter(value.split("_")))
for member in cls:
if member.value == value.lower():
return member
Expand Down Expand Up @@ -60,7 +61,8 @@ class Regions(str, Enum):

OAUTH_CONFIG_URL = "/eadrax-ucs/v1/presentation/oauth/config"

VEHICLES_URL = "/eadrax-vcs/v1/vehicles"
VEHICLES_URL = "/eadrax-vcs/v2/vehicles"
VEHICLE_STATE_URL = VEHICLES_URL + "/{vin}/state"

REMOTE_SERVICE_BASE_URL = "/eadrax-vrccs/v2/presentation/remote-commands"
REMOTE_SERVICE_URL = REMOTE_SERVICE_BASE_URL + "/{vin}/{service_type}"
Expand All @@ -73,8 +75,11 @@ class Regions(str, Enum):
VEHICLE_CHARGING_STATISTICS_URL = "/eadrax-chs/v1/charging-statistics"
VEHICLE_CHARGING_SESSIONS_URL = "/eadrax-chs/v1/charging-sessions"

SERVICE_PROPERTIES = "properties"
SERVICE_STATUS = "status"
SERVICE_CHARGING_STATISTICS_URL = "CHARGING_STATISTICS"
SERVICE_CHARGING_SESSIONS_URL = "CHARGING_SESSIONS"
SERVICE_CHARGING_PROFILE = "CHARGING_PROFILE"


ATTR_STATE = "state"
ATTR_CAPABILITIES = "capabilities"
ATTR_ATTRIBUTES = "attributes"
13 changes: 12 additions & 1 deletion bimmer_connected/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,19 @@


class StrEnum(str, Enum):
"""A string enumeration of type `(str, Enum)`. All members are compared via `upper()`."""
"""A string enumeration of type `(str, Enum)`. All members are compared via `upper()`. Defaults to UNKNOWN."""

@classmethod
def _missing_(cls, value):
has_unknown = False
for member in cls:
if member.value.upper() == "UNKNOWN":
has_unknown = True
if member.value.upper() == value.upper():
return member
if has_unknown:
_LOGGER.warning("'%s' is not a valid '%s'", value, cls.__name__)
return getattr(cls, "UNKNOWN")
raise ValueError(f"'{value}' is not a valid {cls.__name__}")


Expand Down Expand Up @@ -119,3 +125,8 @@ class ValueWithUnit(NamedTuple):

value: Optional[Union[int, float]]
unit: Optional[str]

# def __add__(self, other: "ValueWithUnit"):
# if self.unit != other.unit or not other:
# raise ValueError("Both values must have the same unit!")
# return ValueWithUnit(self.value + other.value, self.unit)
15 changes: 15 additions & 0 deletions bimmer_connected/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,3 +87,18 @@ def _func_wrapper(*args: "_P.args", **kwargs: "_P.kwargs") -> "_R | None":
return _func_wrapper

return decorator


def to_camel_case(input_str: str) -> str:
"""Converts SNAKE_CASE or snake_case to camelCase."""

retval = ""
flag_upper = False
for curr in input_str.lower():
if not curr.isalnum():
if curr == "_":
flag_upper = True
continue
retval = retval + (curr.upper() if flag_upper else curr)
flag_upper = False
return retval
60 changes: 22 additions & 38 deletions bimmer_connected/vehicle/charging_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

from bimmer_connected.const import ATTR_STATE
from bimmer_connected.models import StrEnum, VehicleDataBase

_LOGGER = logging.getLogger(__name__)
Expand All @@ -13,23 +14,26 @@
class ChargingMode(StrEnum):
"""Charging mode of electric vehicle."""

IMMEDIATE_CHARGING = "immediateCharging"
DELAYED_CHARGING = "delayedCharging"
IMMEDIATE_CHARGING = "IMMEDIATE_CHARGING"
DELAYED_CHARGING = "DELAYED_CHARGING"
UNKNOWN = "UNKNOWN"


class ChargingPreferences(StrEnum):
"""Charging preferences of electric vehicle."""

NO_PRESELECTION = "noPreSelection"
CHARGING_WINDOW = "chargingWindow"
NO_PRESELECTION = "NO_PRESELECTION"
CHARGING_WINDOW = "CHARGING_WINDOW"
UNKNOWN = "UNKNOWN"


class TimerTypes(StrEnum):
"""Different timer types."""

TWO_WEEKS = "twoWeeksTimer"
ONE_WEEK = "weeklyPlanner"
OVERRIDE_TIMER = "overrideTimer"
TWO_WEEKS = "TWO_WEEKS_TIMER"
ONE_WEEK = "WEEKLY_PLANNER"
OVERRIDE_TIMER = "OVERRIDE_TIMER"
UNKNOWN = "UNKNOWN"


class ChargingWindow:
Expand All @@ -41,14 +45,12 @@ def __init__(self, window_dict: dict):
@property
def start_time(self) -> datetime.time:
"""Start of the charging window."""
# end of reductionOfChargeCurrent == start of charging window
return datetime.time(int(self._window_dict["end"]["hour"]), int(self._window_dict["end"]["minute"]))
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."""
# start of reductionOfChargeCurrent == end of charging window
return datetime.time(int(self._window_dict["start"]["hour"]), int(self._window_dict["start"]["minute"]))
return datetime.time(int(self._window_dict["end"]["hour"]), int(self._window_dict["end"]["minute"]))


class DepartureTimer:
Expand Down Expand Up @@ -107,32 +109,14 @@ def _parse_vehicle_data(cls, vehicle_data: Dict) -> Dict:
"""Parse doors and windows."""
retval: Dict[str, Any] = {}

if "status" not in vehicle_data or "chargingProfile" not in vehicle_data["status"]:
if vehicle_data["capabilities"]["isChargingPlanSupported"]:
_LOGGER.error("Unable to read data from `status.chargingProfile`.")
return retval

charging_profile = vehicle_data["status"]["chargingProfile"]

retval["is_pre_entry_climatization_enabled"] = (
bool(charging_profile["climatisationOn"]) if "" in charging_profile else None
)
retval["departure_times"] = [DepartureTimer(t) for t in charging_profile.get("departureTimes", [])]
retval["preferred_charging_window"] = (
ChargingWindow(charging_profile["reductionOfChargeCurrent"])
if "reductionOfChargeCurrent" in charging_profile
else None
)
retval["timer_type"] = (
TimerTypes(charging_profile["chargingControlType"]) if "chargingControlType" in charging_profile else None
)
retval["charging_preferences"] = (
ChargingPreferences(charging_profile["chargingPreference"])
if "chargingPreference" in charging_profile
else None
)
retval["charging_mode"] = (
ChargingMode(charging_profile["chargingMode"]) if "chargingMode" in charging_profile else None
)
if ATTR_STATE in vehicle_data and "chargingProfile" in vehicle_data[ATTR_STATE]:
charging_profile = vehicle_data[ATTR_STATE]["chargingProfile"]

retval["is_pre_entry_climatization_enabled"] = bool(charging_profile.get("climatisationOn", False))
retval["departure_times"] = [DepartureTimer(t) for t in charging_profile.get("departureTimes", [])]
retval["preferred_charging_window"] = ChargingWindow(charging_profile.get("reductionOfChargeCurrent", {}))
retval["timer_type"] = TimerTypes(charging_profile.get("chargingControlType", "UNKNOWN"))
retval["charging_preferences"] = ChargingPreferences(charging_profile.get("chargingPreference", "UNKNOWN"))
retval["charging_mode"] = ChargingMode(charging_profile.get("chargingMode", "UNKNOWN"))

return retval
48 changes: 27 additions & 21 deletions bimmer_connected/vehicle/doors_windows.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@
from dataclasses import dataclass, field
from typing import Any, Dict, List

from bimmer_connected.const import ATTR_STATE
from bimmer_connected.models import StrEnum, VehicleDataBase
from bimmer_connected.utils import to_camel_case

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -61,9 +63,6 @@ class DoorsAndWindows(VehicleDataBase): # pylint:disable=too-many-instance-attr
door_lock_state: LockState = LockState.UNKNOWN
"""Get state of the door locks."""

convertible_top: LidState = LidState.UNKNOWN
"""Get state of the convertible roof."""

lids: List[Lid] = field(default_factory=list)
"""All lids (doors+hood+trunk) of the car."""

Expand All @@ -75,24 +74,31 @@ def _parse_vehicle_data(cls, vehicle_data: Dict) -> Dict:
"""Parse doors and windows."""
retval: Dict[str, Any] = {}

if "properties" not in vehicle_data or "doorsAndWindows" not in vehicle_data["status"]:
_LOGGER.error("Unable to read data from `properties.doorsAndWindows`.")
return retval

doors_and_windows = vehicle_data["properties"]["doorsAndWindows"]

retval["lids"] = [
Lid(k, v) for k, v in doors_and_windows.items() if k in ["hood", "trunk"] and v != LidState.INVALID
] + [Lid(k, v) for k, v in doors_and_windows["doors"].items() if v != LidState.INVALID]

retval["windows"] = [Window(k, v) for k, v in doors_and_windows["windows"].items() if v != LidState.INVALID]
if "moonroof" in doors_and_windows:
retval["windows"].append(Window("moonroof", doors_and_windows["moonroof"]))

if "convertibleTop" in doors_and_windows:
retval["convertible_top"] = LidState(doors_and_windows["convertibleTop"])

retval["door_lock_state"] = LockState(vehicle_data["status"]["doorsGeneralState"].upper())
if ATTR_STATE in vehicle_data:
if "doorsState" in vehicle_data[ATTR_STATE]:
retval["lids"] = [
Lid(k, v)
for k, v in vehicle_data[ATTR_STATE]["doorsState"].items()
if k not in ["combinedState", "combinedSecurityState"] and v != LidState.INVALID
]
retval["door_lock_state"] = LockState(
vehicle_data[ATTR_STATE]["doorsState"].get("combinedSecurityState", "UNKNOWN")
)

if "windowsState" in vehicle_data[ATTR_STATE]:
retval["windows"] = [
Window(k, v)
for k, v in vehicle_data[ATTR_STATE]["windowsState"].items()
if k not in ["combinedState"] and v != LidState.INVALID
]

if "roofState" in vehicle_data[ATTR_STATE]:
retval["lids"].append(
Lid(
to_camel_case(vehicle_data[ATTR_STATE]["roofState"]["roofStateType"]),
vehicle_data[ATTR_STATE]["roofState"]["roofState"],
)
)

return retval

Expand Down
Loading

0 comments on commit a1e6c36

Please sign in to comment.