From 1b903b192f153beb4d57e887218c5c0948aca1fe Mon Sep 17 00:00:00 2001 From: Jacob Berelman <630000+stickpin@users.noreply.github.com> Date: Wed, 10 Jan 2024 15:17:32 +0100 Subject: [PATCH] Code clean-up --- volkswagencarnet/vw_connection.py | 374 ------------------------------ volkswagencarnet/vw_vehicle.py | 221 +----------------- 2 files changed, 5 insertions(+), 590 deletions(-) diff --git a/volkswagencarnet/vw_connection.py b/volkswagencarnet/vw_connection.py index 32fc1178..3743d954 100644 --- a/volkswagencarnet/vw_connection.py +++ b/volkswagencarnet/vw_connection.py @@ -558,29 +558,6 @@ async def update(self): _LOGGER.warning(f"Could not update information: {error}") return False - # Data collect functions # - async def getHomeRegion(self, vin): - """Get API requests base url for VIN.""" - if not await self.validate_tokens: - return False - try: - # TODO: handle multiple home regions! (no examples available currently) - return True - response = await self.get( - "https://mal-1a.prd.ece.vwg-connect.com/api/cs/vds/v1/vehicles/$vin/homeRegion", vin - ) - self._session_auth_ref_urls[vin] = ( - response["homeRegion"]["baseUri"]["content"].split("/api")[0].replace("mal-", "fal-") - if response["homeRegion"]["baseUri"]["content"] != "https://mal-1a.prd.ece.vwg-connect.com/api" - else "https://msg.volkswagen.de" - ) - self._session_spin_ref_urls[vin] = response["homeRegion"]["baseUri"]["content"].split("/api")[0] - return response["homeRegion"]["baseUri"]["content"] - except Exception as error: - _LOGGER.debug(f"Could not get homeregion, error {error}") - self._session_logged_in = False - return False - async def getOperationList(self, vin): """Collect operationlist for VIN, supported/licensed functions.""" if not await self.validate_tokens: @@ -679,180 +656,6 @@ async def getTripLast(self, vin): _LOGGER.warning(f"Could not fetch last trip data, error: {error}") return False - async def getRealCarData(self, vin): - """Get car information from customer profile, VIN, nickname, etc.""" - if not await self.validate_tokens: - return False - try: - _LOGGER.debug("Attempting extraction of subject from identity token.") - atoken = self._session_tokens["identity"]["access_token"] - subject = jwt.decode(atoken, options={"verify_signature": False}, algorithms=JWT_ALGORITHMS).get( - "sub", None - ) - self._session_headers["Accept"] = "application/json" - response = await self.get(f"https://customer-profile.vwgroup.io/v1/customers/{subject}/realCarData") - if response.get("realCars", {}): - data = { - "carData": next( - item for item in response.get("realCars", []) if item["vehicleIdentificationNumber"] == vin - ) - } - return data - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch realCarData, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch realcar data") - except Exception as error: - _LOGGER.warning(f"Could not fetch realCarData, error: {error}") - return False - - async def getVehicleStatusData(self, vin): - """Get stored vehicle data response.""" - try: - response = await self.get(f"fs-car/bs/vsr/v1/{BRAND}/{self._session_country}/vehicles/$vin/status", vin=vin) - if ( - response.get("StoredVehicleDataResponse", {}) - .get("vehicleData", {}) - .get("data", {})[0] - .get("field", {})[0] - ): - data = { - "StoredVehicleDataResponse": response.get("StoredVehicleDataResponse", {}), - "StoredVehicleDataResponseParsed": { - e["id"]: e if "value" in e else "" - for f in [s["field"] for s in response["StoredVehicleDataResponse"]["vehicleData"]["data"]] - for e in f - }, - } - return data - elif response.get("status_code", {}): - _LOGGER.warning( - f'Could not fetch vehicle status report, HTTP status code: {response.get("status_code")}' - ) - else: - _LOGGER.info("Unhandled error while trying to fetch status data") - except Exception as error: - _LOGGER.warning(f"Could not fetch StoredVehicleDataResponse, error: {error}") - return False - - async def getTripStatistics(self, vin): - """Get short term trip statistics.""" - if not await self.validate_tokens: - return False - try: - response = await self.get( - f"fs-car/bs/tripstatistics/v1/{BRAND}/{self._session_country}/vehicles/$vin/tripdata/shortTerm?newest", - vin=vin, - ) - if response.get("tripData", {}): - data = {"tripstatistics": response.get("tripData", {})} - return data - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch trip statistics, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch trip statistics") - except Exception as error: - _LOGGER.warning(f"Could not fetch trip statistics, error: {error}") - return False - - async def getPosition(self, vin): - """Get position data.""" - if not await self.validate_tokens: - return False - try: - response = await self.get( - f"fs-car/bs/cf/v1/{BRAND}/{self._session_country}/vehicles/$vin/position", vin=vin - ) - if response.get("findCarResponse", {}): - data = {"findCarResponse": response.get("findCarResponse", {}), "isMoving": False} - return data - elif response.get("status_code", {}): - if response.get("status_code", 0) == 204: - _LOGGER.debug("Seems car is moving, HTTP 204 received from position") - data = {"isMoving": True, "rate_limit_remaining": 15} - return data - else: - _LOGGER.warning(f'Could not fetch position, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch positional data") - except Exception as error: - _LOGGER.warning(f"Could not fetch position, error: {error}") - return False - - async def getTimers(self, vin) -> TimerData | None: - """Get departure timers.""" - if not await self.validate_tokens: - return None - try: - response = await self.get( - f"fs-car/bs/departuretimer/v1/{BRAND}/{self._session_country}/vehicles/$vin/timer", vin=vin - ) - timer = TimerData(**(response.get("timer", {}))) - if timer.valid: - return timer - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch timers, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unknown error while trying to fetch data for departure timers") - except Exception as error: - _LOGGER.warning(f"Could not fetch timers, error: {error}") - return None - - async def getClimater(self, vin): - """Get climatisation data.""" - if not await self.validate_tokens: - return False - try: - response = await self.get( - f"fs-car/bs/climatisation/v1/{BRAND}/{self._session_country}/vehicles/$vin/climater", vin=vin - ) - if response.get("climater", {}): - data = {"climater": response.get("climater", {})} - return data - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch climatisation, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch climatisation data") - except Exception as error: - _LOGGER.warning(f"Could not fetch climatisation, error: {error}") - return False - - async def getCharger(self, vin): - """Get charger data.""" - if not await self.validate_tokens: - return False - try: - response = await self.get( - f"fs-car/bs/batterycharge/v1/{BRAND}/{self._session_country}/vehicles/$vin/charger", vin=vin - ) - if response.get("charger", {}): - data = {"charger": response.get("charger", {})} - return data - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch pre-heating, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch charger data") - except Exception as error: - _LOGGER.warning(f"Could not fetch charger, error: {error}") - return False - - async def getPreHeater(self, vin): - """Get parking heater data.""" - if not await self.validate_tokens: - return False - try: - response = await self.get(f"fs-car/bs/rs/v1/{BRAND}/{self._session_country}/vehicles/$vin/status", vin=vin) - if response.get("statusResponse", {}): - data = {"heating": response.get("statusResponse", {})} - return data - elif response.get("status_code", {}): - _LOGGER.warning(f'Could not fetch pre-heating, HTTP status code: {response.get("status_code")}') - else: - _LOGGER.info("Unhandled error while trying to fetch pre-heating data") - except Exception as error: - _LOGGER.warning(f"Could not fetch pre-heating, error: {error}") - return False - async def get_request_status(self, vin, sectionId, requestId, actionId=""): """Return status of a request ID for a given section ID.""" if self.logged_in is False: @@ -950,91 +753,6 @@ async def get_sec_token(self, vin, spin, action): _LOGGER.error(f"Could not generate security token (maybe wrong SPIN?), error: {error}") raise - # Data set functions # - async def dataCall(self, query, vin="", **data): - """Execute actions through VW-Group API.""" - if self.logged_in is False: - if not await self.doLogin(): - _LOGGER.warning(f"Login for {BRAND} account failed!") - raise Exception(f"Login for {BRAND} account failed") - try: - if not await self.validate_tokens: - _LOGGER.info(f"Session expired. Initiating new login for {BRAND} account.") - if not await self.doLogin(): - _LOGGER.warning(f"Login for {BRAND} account failed!") - raise Exception(f"Login for {BRAND} account failed") - response = await self.post(query, vin=vin, **data) - _LOGGER.debug(f"Data call returned: {response}") - return response - except client_exceptions.ClientResponseError as error: - if error.status == 401: - _LOGGER.error("Unauthorized") - self._session_logged_in = False - elif error.status == 400: - _LOGGER.error("Bad request") - elif error.status == 429: - _LOGGER.warning( - "Too many requests. Further requests can only be made after the end of next trip in order to" - " protect your vehicles battery." - ) - return 429 - elif error.status == 500: - _LOGGER.error("Internal server error, server might be temporarily unavailable") - elif error.status == 502: - _LOGGER.error("Bad gateway, this function may not be implemented for this vehicle") - else: - _LOGGER.error(f"Unhandled HTTP exception: {error}") - # return False - except Exception as error: - _LOGGER.error(f"Failure to execute: {error}") - return False - - async def setRefresh(self, vin): - """Force vehicle data update.""" - try: - response = await self.dataCall( - f"fs-car/bs/vsr/v1/{BRAND}/{self._session_country}/vehicles/$vin/requests", vin, data=None - ) - if not response: - raise Exception("Invalid or no response") - elif response == 429: - return dict({"id": None, "state": "Throttled", "rate_limit_remaining": 0}) - else: - request_id = response.get("CurrentVehicleDataResponse", {}).get("requestId", 0) - request_state = response.get("CurrentVehicleDataResponse", {}).get("requestState", "queued") - remaining = response.get("rate_limit_remaining", -1) - _LOGGER.debug( - f'Request to refresh data returned with state "{request_state}", request id: {request_id},' - f" remaining requests: {remaining}" - ) - return dict({"id": str(request_id), "state": request_state, "rate_limit_remaining": remaining}) - except: - raise - - async def setCharger(self, vin, data) -> dict[str, str | int | None]: - """Start/Stop charger.""" - try: - response = await self.dataCall( - f"fs-car/bs/batterycharge/v1/{BRAND}/{self._session_country}/vehicles/$vin/charger/actions", - vin, - json=data, - ) - if not response: - raise Exception("Invalid or no response") - elif response == 429: - return dict({"id": None, "state": "Throttled", "rate_limit_remaining": 0}) - else: - request_id = response.get("action", {}).get("actionId", 0) - request_state = response.get("action", {}).get("actionState", "unknown") - remaining = response.get("rate_limit_remaining", -1) - _LOGGER.debug( - f'Request for charger action returned with state "{request_state}", request id: {request_id},' - f" remaining requests: {remaining}" - ) - return dict({"id": str(request_id), "state": request_state, "rate_limit_remaining": remaining}) - except: - raise - async def setClimater(self, vin, data, action): """Execute climatisation actions.""" @@ -1063,98 +781,6 @@ async def setWindowHeater(self, vin, action): except Exception as e: raise Exception("Unknown error during setWindowHeater") from e - async def setPreHeater(self, vin, data, spin): - """Petrol/diesel parking heater actions.""" - content_type = None - try: - if "Content-Type" in self._session_headers: - content_type = self._session_headers["Content-Type"] - else: - content_type = "" - self._session_headers["Content-Type"] = "application/vnd.vwg.mbb.RemoteStandheizung_v2_0_2+json" - if "quickstop" not in data: - self._session_headers["x-mbbSecToken"] = await self.get_sec_token(vin=vin, spin=spin, action="heating") - response = await self.dataCall( - f"fs-car/bs/rs/v1/{BRAND}/{self._session_country}/vehicles/$vin/action", vin=vin, json=data - ) - # Clean up headers - self._session_headers.pop("x-mbbSecToken", None) - self._session_headers.pop("Content-Type", None) - if content_type: - self._session_headers["Content-Type"] = content_type - - if not response: - raise Exception("Invalid or no response") - elif response == 429: - return dict({"id": None, "state": "Throttled", "rate_limit_remaining": 0}) - else: - request_id = response.get("performActionResponse", {}).get("requestId", 0) - remaining = response.get("rate_limit_remaining", -1) - _LOGGER.debug( - f"Request for parking heater is queued with request id: {request_id}, remaining requests:" - f" {remaining}" - ) - return dict({"id": str(request_id), "state": None, "rate_limit_remaining": remaining}) - except Exception: - self._session_headers.pop("x-mbbSecToken", None) - self._session_headers.pop("Content-Type", None) - if content_type: - self._session_headers["Content-Type"] = content_type - raise - - async def setTimersAndProfiles(self, vin, data: TimersAndProfiles): - """Set schedules.""" - return await self._setDepartureTimer(vin, data, "setTimersAndProfiles") - - async def setChargeMinLevel(self, vin: str, limit: int): - """Set schedules.""" - data: TimerData | None = await self.getTimers(vin) - if data is None or data.timersAndProfiles is None or data.timersAndProfiles.timerBasicSetting is None: - raise Exception("No existing timer data?") - data.timersAndProfiles.timerBasicSetting.set_charge_min_limit(limit) - return await self._setDepartureTimer(vin, data.timersAndProfiles, "setChargeMinLimit") - - # Not working :/ - # async def setHeaterSource(self, vin: str, source: str): - # """Set heater source for departure timers.""" - # data: Optional[TimerData] = await self.getTimers(vin) - # if data is None: - # raise Exception("No existing timer data?") - # data.timersAndProfiles.timerBasicSetting.set_heater_source(source) - # return await self._setDepartureTimer(vin, data.timersAndProfiles, "setHeaterSource") - - async def _setDepartureTimer(self, vin, data: TimersAndProfiles, action: str): - """Set schedules.""" - try: - response = await self.dataCall( - f"fs-car/bs/departuretimer/v1/{BRAND}/{self._session_country}/vehicles/$vin/timer/actions", - vin=vin, - json={ - "action": { - "timersAndProfiles": data.json_updated["timer"], - "type": action, - } - }, - ) - - self._session_headers.pop("X-securityToken", None) - if not response: - raise Exception("Invalid or no response") - elif response == 429: - return dict({"id": None, "state": "Throttled", "rate_limit_remaining": 0}) - else: - request_id = response.get("action", {}).get("actionId", 0) - request_state = response.get("action", {}).get("actionState", "unknown") - remaining = response.get("rate_limit_remaining", -1) - _LOGGER.debug( - f'Request for timer action returned with state "{request_state}", request id: {request_id},' - f" remaining requests: {remaining}" - ) - return dict({"id": str(request_id), "state": request_state, "rate_limit_remaining": remaining}) - except: - self._session_headers.pop("X-securityToken", None) - raise - async def setLock(self, vin, lock, spin): """Remote lock and unlock actions.""" await self.check_spin_state() diff --git a/volkswagencarnet/vw_vehicle.py b/volkswagencarnet/vw_vehicle.py index e31c4338..bf4cc407 100644 --- a/volkswagencarnet/vw_vehicle.py +++ b/volkswagencarnet/vw_vehicle.py @@ -183,13 +183,6 @@ async def update(self): self.get_vehicle(), self.get_parkingposition(), self.get_trip_last() - # self.get_preheater(), - # self.get_climater(), - # self.get_trip_statistic(), - # self.get_position(), - # self.get_statusreport(), - # self.get_charger(), - # self.get_timerprogramming(), # return_exceptions=True, ) await asyncio.gather(self.get_service_status()) @@ -229,101 +222,6 @@ async def get_service_status(self): if data: self._states.update({Services.SERVICE_STATUS: data}) - async def get_realcardata(self): - """Fetch realcardata.""" - data = await self._connection.getRealCarData(self.vin) - if data: - self._states.update(data) - - async def get_preheater(self): - """Fetch pre-heater data if function is enabled.""" - if self._services.get("rheating_v1", {}).get("active", False): - if not await self.expired("rheating_v1"): - data = await self._connection.getPreHeater(self.vin) - if data: - self._states.update(data) - else: - _LOGGER.debug("Could not fetch preheater data") - else: - self._requests.pop("preheater", None) - - async def get_climater(self): - """Fetch climater data if function is enabled.""" - if self._services.get("rclima_v1", {}).get("active", False): - if not await self.expired("rclima_v1"): - data = await self._connection.getClimater(self.vin) - if data: - self._states.update(data) - else: - _LOGGER.debug("Could not fetch climater data") - else: - self._requests.pop("climatisation", None) - - async def get_trip_statistic(self): - """Fetch trip data if function is enabled.""" - if self._services.get("trip_statistic_v1", {}).get("active", False): - if not await self.expired("trip_statistic_v1"): - data = await self._connection.getTripStatistics(self.vin) - if data: - self._states.update(data) - else: - _LOGGER.debug("Could not fetch trip statistics") - - async def get_position(self): - """Fetch position data if function is enabled.""" - if self._services.get("carfinder_v1", {}).get("active", False): - if not await self.expired("carfinder_v1"): - data = await self._connection.getPosition(self.vin) - if data: - # Reset requests remaining to 15 if parking time has been updated - if data.get("findCarResponse", {}).get("parkingTimeUTC", None) is not None: - try: - new_time = data.get("findCarResponse").get("parkingTimeUTC") - old_time = self.attrs.get("findCarResponse", {}).get("parkingTimeUTC", None) - if old_time is None or new_time > old_time: - _LOGGER.debug(f"Detected new parking time: {new_time}") - self.requests_remaining = -1 if old_time is None else 15 - self.requests_remaining_last_updated = datetime.utcnow() - except Exception as e: - _LOGGER.warning(f"Failed to parse parking time: {e}") - self._states.update(data) - else: - _LOGGER.debug("Could not fetch any positional data") - - async def get_statusreport(self): - """Fetch status data if function is enabled.""" - if self._services.get("statusreport_v1", {}).get("active", False): - if not await self.expired("statusreport_v1"): - data = await self._connection.getVehicleStatusData(self.vin) - if data: - self._states.update(data) - else: - _LOGGER.debug("Could not fetch status report") - - async def get_charger(self): - """Fetch charger data if function is enabled.""" - if self._services.get("rbatterycharge_v1", {}).get("active", False): - if not await self.expired("rbatterycharge_v1"): - data = await self._connection.getCharger(self.vin) - if data: - self._states.update(data) - else: - _LOGGER.debug("Could not fetch charger data") - else: - self._requests.pop("charger", None) - - async def get_timerprogramming(self) -> None: - """Fetch timer data if function is enabled.""" - if self._services.get("timerprogramming_v1", {}).get("active", False): - if not await self.expired("timerprogramming_v1"): - data = await self._connection.getTimers(self.vin) - if data: - self._states.update({"timer": data}) - else: - _LOGGER.debug("Could not fetch timers") - else: - self._requests.pop("departuretimer", None) - async def wait_for_request(self, section, request, retry_count=36, action=""): """Update status of outstanding requests.""" retry_count -= 1 @@ -360,67 +258,11 @@ async def set_charger_current(self, value) -> bool: async def set_charge_min_level(self, level: int): """Set the desired minimum charge level for departure schedules.""" - if self.is_schedule_min_charge_level_supported: - if (0 <= level <= 100) and level % 10 == 0: - if self._in_progress("departuretimer"): - return False - try: - self._requests["latest"] = "Departuretimer" - response = await self._connection.setChargeMinLevel(self.vin, level) - return await self._handle_response( - response=response, topic="departuretimer", error_msg="Failed to set minimum charge level" - ) - except Exception as error: - _LOGGER.warning(f"Failed to set minimum charge level - {error}") - self._requests["departuretimer"] = {"status": "Exception", "timestamp": datetime.now()} - raise Exception(f"Failed to set minimum charge level - {error}") - else: - raise Exception("Level must be 0, 10, ..., 100") - else: - _LOGGER.error("Cannot set minimum level") - raise Exception("Cannot set minimum level") - - # Need to figure out a way to implement this... - # async def set_departure_timer_heater_source(self, source: str): - # """Set the heater source to use.""" - # if self.is_timer_basic_settings_supported: - # if source not in ["auxiliary", "electric"]: - # raise ValueError(f"Source '{source}' not supported.") - # try: - # self._requests["latest"] = "Departuretimer" - # response = await self._connection.setHeaterSource(self.vin, source) - # return await self._handle_response( - # response=response, topic="departuretimer", error_msg="Failed to set heater source" - # ) - # except Exception as error: - # _LOGGER.warning(f"Failed to set heater source - {error}") - # self._requests["departuretimer"] = {"status": "Exception"} - # raise Exception(f"Failed to set heater source - {error}") + raise Exception("Should have to be re-implemented") async def set_charger(self, action) -> bool: """Charging actions.""" - if not self._services.get("charging", False): - _LOGGER.info("Remote start/stop of charger is not supported.") - raise Exception("Remote start/stop of charger is not supported.") - if self._in_progress("batterycharge"): - return False - if action in ["start", "stop"]: - data = {"action": {"type": action}} - elif action.get("action", {}).get("type", "") == "setSettings": - data = action - else: - _LOGGER.error(f"Invalid charger action: {action}. Must be either start or stop") - raise Exception(f"Invalid charger action: {action}. Must be either start or stop") - try: - self._requests["latest"] = "Charger" - response = await self._connection.setCharger(self.vin, data) - return await self._handle_response( - response=response, topic="batterycharge", error_msg=f"Failed to {action} charging" - ) - except Exception as error: - _LOGGER.warning(f"Failed to {action} charging - {error}") - self._requests["batterycharge"] = {"status": "Exception", "timestamp": datetime.now()} - raise Exception(f"Failed to {action} charging - {error}") + raise Exception("Should have to be re-implemented") # Climatisation electric/auxiliary/windows (CLIMATISATION) async def set_climatisation_temp(self, temperature=20): @@ -511,32 +353,7 @@ async def set_climater(self, data, spin=False): # Parking heater heating/ventilation (RS) async def set_pheater(self, mode, spin): """Set the mode for the parking heater.""" - if not self.is_pheater_heating_supported: - _LOGGER.error("No parking heater support.") - raise Exception("No parking heater support.") - if self._in_progress("preheater"): - return False - if mode not in ["heating", "ventilation", "off"]: - _LOGGER.error(f"{mode} is an invalid action for parking heater") - raise Exception(f"{mode} is an invalid action for parking heater") - if mode == "off": - data = {"performAction": {"quickstop": {"active": False}}} - else: - data = { - "performAction": { - "quickstart": {"climatisationDuration": self.pheater_duration, "startMode": mode, "active": True} - } - } - try: - self._requests["latest"] = "Preheater" - response = await self._connection.setPreHeater(self.vin, data, spin) - return await self._handle_response( - response=response, topic="preheater", error_msg=f"Failed to set parking heater to {mode}" - ) - except Exception as error: - _LOGGER.warning(f"Failed to set parking heater mode to {mode} - {error}") - self._requests["preheater"] = {"status": "Exception", "timestamp": datetime.now()} - raise Exception("Pre-heater action failed") + raise Exception("Should have to be re-implemented") # Lock (RLU) async def set_lock(self, action, spin): @@ -564,39 +381,11 @@ async def set_lock(self, action, spin): # Refresh vehicle data (VSR) async def set_refresh(self): """Wake up vehicle and update status data.""" - if not self._services.get("statusreport_v1", {}).get("active", False): - _LOGGER.info("Data refresh is not supported.") - raise Exception("Data refresh is not supported.") - if self._in_progress("refresh", unknown_offset=-5): - return False - try: - self._requests["latest"] = "Refresh" - response = await self._connection.setRefresh(self.vin) - return await self._handle_response( - response=response, topic="refresh", error_msg="Failed to request vehicle update" - ) - except Exception as error: - _LOGGER.warning(f"Failed to execute data refresh - {error}") - self._requests["refresh"] = {"status": "Exception", "timestamp": datetime.now()} - raise Exception("Data refresh failed") + raise Exception("Should have to be re-implemented") async def set_schedule(self, data: TimerData) -> bool: """Store schedule.""" - if not self._services.get("timerprogramming_v1", False): - _LOGGER.info("Remote control of timer functions is not supported.") - raise Exception("Remote control of timer functions is not supported.") - if self._in_progress("departuretimer"): - return False - try: - self._requests["latest"] = "Departuretimer" - response = await self._connection.setTimersAndProfiles(self.vin, data.timersAndProfiles) - return await self._handle_response( - response=response, topic="departuretimer", error_msg="Failed to execute timer request" - ) - except Exception as error: - _LOGGER.warning(f"Failed to execute timer request - {error}") - self._requests["departuretimer"] = {"status": "Exception", "timestamp": datetime.now()} - raise Exception("Timer action failed") + raise Exception("Should have to be re-implemented") # Vehicle class helpers # # Vehicle info