diff --git a/examples/destinations.py b/examples/destinations.py new file mode 100644 index 0000000..24e8990 --- /dev/null +++ b/examples/destinations.py @@ -0,0 +1,85 @@ +import argparse + +from weconnect import weconnect +from weconnect.elements.route import Address, Destination, GeoCoordinate, Route + + +def main(): + """ + Simple example showing how to send destinations in a vehicle by providing the VIN as a parameter + Giving an address for the destination(s) is optional, and is only used for display purposes. + """ + parser = argparse.ArgumentParser( + prog="destinations", description="Example sending destinations" + ) + parser.add_argument( + "-u", "--username", help="Username of Volkswagen id", required=True + ) + parser.add_argument( + "-p", "--password", help="Password of Volkswagen id", required=True + ) + parser.add_argument( + "--vin", help="VIN of the vehicle to start destinations", required=True + ) + + args = parser.parse_args() + + route = Route( + [ + Destination( + name="VW Museum", + geoCoordinate=GeoCoordinate( + latitude=52.4278793, + longitude=10.8077433, + ), + ), + Destination( + name="Autostadt", + geoCoordinate=GeoCoordinate( + latitude=52.429380, + longitude=10.791520, + ), + address=Address( + country="Germany", + street="Stadtbrücke", + zipCode="38440", + city="Wolfsburg", + ), + ), + ] + ) + + print("# Initialize WeConnect") + weConnect = weconnect.WeConnect( + username=args.username, + password=args.password, + updateAfterLogin=False, + loginOnInit=False, + ) + print("# Login") + weConnect.login() + print("# update") + weConnect.update() + + for vin, vehicle in weConnect.vehicles.items(): + if vin == args.vin: + print("# send destinations") + + if ( + "destinations" not in vehicle.capabilities + or not vehicle.capabilities["destinations"] + ): + print("# destinations is not supported") + continue + + if not vehicle.controls.sendDestinations: + print("# sendDestinations is not available") + continue + + vehicle.controls.sendDestinations.value = route + + print("# destinations sent") + + +if __name__ == "__main__": + main() diff --git a/tests/test_route.py b/tests/test_route.py new file mode 100644 index 0000000..5e71ceb --- /dev/null +++ b/tests/test_route.py @@ -0,0 +1,203 @@ +import json +import pytest +from weconnect.elements.route import Address, GeoCoordinate, Destination, Route + + +def test_valid_coordinates(): + geo = GeoCoordinate(latitude=52.52, longitude=13.405) + assert geo.to_dict() == {"latitude": 52.52, "longitude": 13.405} + + +def test_invalid_coordinate_types(): + with pytest.raises(TypeError): + GeoCoordinate(latitude="invalid", longitude="invalid") + + +def test_invalid_coordinates(): + with pytest.raises(ValueError): + GeoCoordinate(latitude=255.0, longitude=512.0) + + +def test_edge_case_coordinates(): + # Test with edge values like 0.0 + GeoCoordinate(latitude=0.0, longitude=0.0) + + +def test_address_to_dict(): + address = Address( + country="Germany", street="Unter den Linden", zipCode="10117", city="Berlin" + ) + expected_dict = { + "country": "Germany", + "street": "Unter den Linden", + "zipCode": "10117", + "city": "Berlin", + } + assert address.to_dict() == expected_dict + + +def test_valid_destination(): + geo = GeoCoordinate(latitude=52.52, longitude=13.405) + dest = Destination(geoCoordinate=geo, name="Brandenburg Gate") + expected_dict = { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "destinationName": "Brandenburg Gate", + "poiProvider": "unknown", + "destinationSource": "MobileApp", + } + assert dest.to_dict() == expected_dict + + +def test_destination_missing_geo(): + with pytest.raises(ValueError): + Destination(geoCoordinate=None) + + +def test_route_with_single_destination(): + geo = GeoCoordinate(latitude=52.52, longitude=13.405) + dest = Destination(geoCoordinate=geo) + route = Route(destinations=dest) + expected_list = [ + { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "destinationName": "Destination", + "poiProvider": "unknown", + "destinationSource": "MobileApp", + } + ] + assert route.to_list() == expected_list + + +def test_route_with_multiple_destinations(): + geo1 = GeoCoordinate(latitude=52.52, longitude=13.405) + dest1 = Destination(geoCoordinate=geo1) + geo2 = GeoCoordinate(latitude=48.8566, longitude=2.3522) + dest2 = Destination(geoCoordinate=geo2, name="Eiffel Tower") + route = Route(destinations=[dest1, dest2]) + expected_list = [ + { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "destinationName": "Destination", + "poiProvider": "unknown", + "destinationSource": "MobileApp", + "destinationType": "stopover", + }, + { + "geoCoordinate": {"latitude": 48.8566, "longitude": 2.3522}, + "destinationName": "Eiffel Tower", + "poiProvider": "unknown", + "destinationSource": "MobileApp", + }, + ] + assert route.to_list() == expected_list + + +def test_route_from_collection(): + data = [ + { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "name": "Brandenburg Gate", + }, + { + "geoCoordinate": {"latitude": 48.8566, "longitude": 2.3522}, + "name": "Eiffel Tower", + }, + ] + route = Route.from_collection(data) + assert len(route.destinations) == 2 + + +def test_invalid_destination_geo(): + with pytest.raises(ValueError): + Destination(geoCoordinate=None) + + +def test_route_with_invalid_destinations(): + with pytest.raises(TypeError): + Route(destinations="not a list") + + +def test_route_from_invalid_collection(): + with pytest.raises(ValueError): + Route.from_collection("invalid data") + + +def test_route_from_value_with_destination(): + geo = GeoCoordinate(latitude=52.52, longitude=13.405) + dest = Destination(geoCoordinate=geo, name="Brandenburg Gate") + route = Route.from_value(dest) + assert isinstance(route, Route) + assert len(route.destinations) == 1 + assert route.destinations[0].name == "Brandenburg Gate" + + +def test_route_from_value_with_list_of_destinations(): + geo1 = GeoCoordinate(latitude=52.52, longitude=13.405) + dest1 = Destination(geoCoordinate=geo1) + geo2 = GeoCoordinate(latitude=48.8566, longitude=2.3522) + dest2 = Destination(geoCoordinate=geo2, name="Eiffel Tower") + route = Route.from_value([dest1, dest2]) + assert isinstance(route, Route) + assert len(route.destinations) == 2 + + +def test_route_from_value_with_dict(): + data = { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "name": "Brandenburg Gate", + } + route = Route.from_value(data) + assert isinstance(route, Route) + assert len(route.destinations) == 1 + assert route.destinations[0].name == "Brandenburg Gate" + + +def test_route_from_value_with_list_of_dicts(): + data = [ + { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "name": "Brandenburg Gate", + }, + { + "geoCoordinate": {"latitude": 48.8566, "longitude": 2.3522}, + "name": "Eiffel Tower", + }, + ] + route = Route.from_value(data) + assert isinstance(route, Route) + assert len(route.destinations) == 2 + assert route.destinations[1].name == "Eiffel Tower" + + +def test_route_from_value_with_json_string(): + data = [ + { + "geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, + "name": "Brandenburg Gate", + }, + { + "geoCoordinate": {"latitude": 48.8566, "longitude": 2.3522}, + "name": "Eiffel Tower", + }, + ] + json_data = json.dumps(data) + route = Route.from_value(json_data) + assert isinstance(route, Route) + assert len(route.destinations) == 2 + assert route.destinations[0].name == "Brandenburg Gate" + + +def test_route_from_value_with_invalid_json_string(): + invalid_json = '{"geoCoordinate": {"latitude": 52.52, "longitude": 13.405}, "name": "Brandenburg Gate"' # Missing closing brace + with pytest.raises(json.JSONDecodeError): + Route.from_value(invalid_json) + + +def test_route_from_value_with_invalid_type(): + with pytest.raises(TypeError): + Route.from_value(12345) + + +def test_route_from_value_with_none(): + with pytest.raises(TypeError): + Route.from_value(None) diff --git a/weconnect/elements/controls.py b/weconnect/elements/controls.py index a7fccdb..9afd541 100644 --- a/weconnect/elements/controls.py +++ b/weconnect/elements/controls.py @@ -1,5 +1,6 @@ import logging import json +from typing import Optional, Union import requests from weconnect.addressable import AddressableObject, ChangeableAttribute @@ -7,6 +8,7 @@ from weconnect.elements.charging_settings import ChargingSettings from weconnect.elements.climatization_settings import ClimatizationSettings from weconnect.elements.error import Error +from weconnect.elements.route import Route, Destination from weconnect.elements.window_heating_status import WindowHeatingStatus from weconnect.elements.access_status import AccessStatus from weconnect.elements.parking_position import ParkingPosition @@ -36,6 +38,7 @@ def __init__( self.honkAndFlashControl = None self.auxiliaryHeating = None self.activeVentilation = None + self.sendDestinations = None self.update() def update(self): # noqa: C901 @@ -79,6 +82,14 @@ def update(self): # noqa: C901 self.honkAndFlashControl = ChangeableAttribute( localAddress='honkAndFlash', parent=self, value=HonkAndFlashControlOperation.NONE, valueType=(HonkAndFlashControlOperation, int), valueSetter=self.__setHonkAndFlashControlChange) + if self.sendDestinations is None and 'destinations' in capabilities and not capabilities['destinations'].status.value: + self.sendDestinations = ChangeableAttribute( + localAddress="destinations", + parent=self, + value='[]', + valueType=(str, list, dict, Route, Destination), + valueSetter=self.__setDestinationsControlChange, + ) if self.wakeupControl is None and 'vehicleWakeUpTrigger' in capabilities and not capabilities['vehicleWakeUpTrigger'].status.value: self.wakeupControl = ChangeableAttribute(localAddress='wakeup', parent=self, value=ControlOperation.NONE, valueType=ControlOperation, valueSetter=self.__setWakeupControlChange) @@ -388,3 +399,48 @@ def __setHonkAndFlashControlChange(self, value): # noqa: C901 else: raise ControlError(f'Could not control honkandflash ({controlResponse.status_code})') raise ControlError(f'Could not control honkandflash ({controlResponse.status_code})') + + def __setDestinationsControlChange(self, value: Optional[Union[str, list, dict, Route, Destination]]): # noqa: C901 + route = None + if value is None: + raise ControlError("Could not control destination, value must not be None.") + if isinstance(value, Route): + # Value is already a Route, no further action needed + route = value + elif isinstance(value, (str, list, dict, Destination)): + try: + route = Route.from_value(value) + except json.JSONDecodeError as err: + raise ControlError(f'Could not control destination, invalid JSON string: {str(err)}') + except (TypeError, ValueError) as err: + raise ControlError(f'Could not control destination, invalid data: {str(err)}') + else: + raise ControlError( + "Could not control destination, value must be a JSON string, list, dict, Route, or Destination." + ) + + url = f'https://emea.bff.cariad.digital/vehicle/v1/vehicles/{self.vehicle.vin.value}/destinations' + data = { + 'destinations': route.to_list() + } + + controlResponse = self.vehicle.weConnect.session.put(url, json=data, allow_redirects=True) + if controlResponse.status_code != requests.codes['accepted']: + errorDict = controlResponse.json() + if errorDict is not None and 'error' in errorDict: + error = Error(localAddress='error', parent=self, fromDict=errorDict['error']) + if error is not None: + message = '' + if error.message.enabled and error.message.value is not None: + message += error.message.value + if error.info.enabled and error.info.value is not None: + message += ' - ' + error.info.value + if error.retry.enabled and error.retry.value is not None: + if error.retry.value: + message += ' - Please retry in a moment' + else: + message += ' - No retry possible' + raise ControlError(f'Could not control destination ({message})') + else: + raise ControlError(f'Could not control destination ({controlResponse.status_code})') + raise ControlError(f'Could not control destination ({controlResponse.status_code})') diff --git a/weconnect/elements/route.py b/weconnect/elements/route.py new file mode 100644 index 0000000..e8d6b15 --- /dev/null +++ b/weconnect/elements/route.py @@ -0,0 +1,185 @@ +from typing import Any, Optional, Union, Dict, List +from dataclasses import dataclass +import json + + +@dataclass +class Address: + country: str + street: str + zipCode: str + city: str + + def to_dict(self) -> Dict[str, str]: + return { + "country": self.country, + "street": self.street, + "zipCode": self.zipCode, + "city": self.city, + } + + +@dataclass +class GeoCoordinate: + latitude: float + longitude: float + + def __post_init__(self) -> None: + if not isinstance(self.latitude, float) or not isinstance( + self.longitude, float + ): + raise TypeError("Latitude and longitude must be floats") + if not (-90.0 <= self.latitude <= 90.0 and -180.0 <= self.longitude <= 180.0): + raise ValueError( + "Latitude must be between -90 and 90 degrees, and longitude between -180 and 180 degrees." + ) + + def to_dict(self) -> Dict[str, float]: + return { + "latitude": self.latitude, + "longitude": self.longitude, + } + + +class Destination: + def __init__( + self, + geoCoordinate: GeoCoordinate, + name: Optional[str] = None, + address: Optional[Address] = None, + poiProvider: Optional[str] = None, + ) -> None: + """ + A single destination on a route. + + Args: + geoCoordinate (GeoCoordinate): A GeoCoordinate instance containing the coordinates of the destination (Required). + name (str): A name for the destination to be displayed in the car (Optional, defaults to "Destination"). + address (Address): The address of the destination, for display purposes only, not used for navigation (Optional). + poiProvider (str): The source of the location (Optional, defaults to "unknown"). + """ + if not isinstance(geoCoordinate, GeoCoordinate): + raise ValueError("geoCoordinate is required") + + self.geoCoordinate = geoCoordinate + self.name = name or "Destination" + self.address = address + self.poiProvider = poiProvider or "unknown" + + def to_dict(self) -> Dict[str, Any]: + data: Dict[str, Any] = { + "geoCoordinate": self.geoCoordinate.to_dict(), + "destinationName": self.name, + "poiProvider": self.poiProvider, + "destinationSource": "MobileApp", + } + + if self.address is not None: + data["address"] = self.address.to_dict() + + return data + + @classmethod + def from_dict(cls, dest_dict: Dict[str, Any]) -> "Destination": + geoCoordinate: Optional[GeoCoordinate] = None + address: Optional[Address] = None + + if "geoCoordinate" in dest_dict: + geoCoordinate = GeoCoordinate(**dest_dict["geoCoordinate"]) + else: + raise ValueError("geoCoordinate is required in destination data") + + if "address" in dest_dict: + address = Address(**dest_dict["address"]) + + return cls( + geoCoordinate=geoCoordinate, + name=dest_dict.get("name", "Destination"), + address=address, + poiProvider=dest_dict.get("poiProvider", "unknown"), + ) + + +class Route: + def __init__( + self, destinations: Union[List[Destination], Destination] = [] + ) -> None: + if isinstance(destinations, Destination): + destinations = [destinations] + elif ( + destinations is None + or not isinstance(destinations, list) + or not all(isinstance(dest, Destination) for dest in destinations) + ): + raise TypeError( + "destinations must be a single Destination or a list of Destination objects." + ) + + self.destinations = destinations + + def to_list(self) -> List[Dict[str, Any]]: + route = [] + for i, destination in enumerate(self.destinations): + data = destination.to_dict() + if i < len(self.destinations) - 1: + data["destinationType"] = "stopover" + route.append(data) + + return route + + @classmethod + def from_collection(cls, collection: Union[list, dict]) -> "Route": + """ + Create a route from a dict or list of dicts containing destinations. + + Args: + route_list (Union[list, dict]): A single destination dict or a list of destinations. + + Example: + Route.from_collection([ + { + "name": "VW Museum", + "geoCoordinate": { + "latitude": 52.4278793, + "longitude": 10.8077433, + }, + }, + { + "name": "Autostadt", + "geoCoordinate": { + "latitude": 52.429380, + "longitude": 10.791520, + }, + "address": { + "country": "Germany", + "street": "Stadtbrücke", + "zipCode": "38440", + "city": "Wolfsburg", + }, + }, + ]) + """ + destinations: List[Destination] = [] + + if isinstance(collection, dict): + destinations.append(Destination.from_dict(collection)) + else: + for dest in collection: + if isinstance(dest, Destination): + destinations.append(dest) + else: + destinations.append(Destination.from_dict(dest)) + + return cls(destinations) + + @classmethod + def from_value(cls, value: Union[str, list, dict, Destination]) -> "Route": + if isinstance(value, Destination): + return cls([value]) + elif isinstance(value, (list, dict)): + return cls.from_collection(value) + elif isinstance(value, str): + data = json.loads(value) + return cls.from_collection(data) + else: + raise TypeError("Unsupported type for Route.from_value")