Skip to content

Add Send Destination function #207

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 14 commits into
base: main
Choose a base branch
from
Open
85 changes: 85 additions & 0 deletions examples/destinations.py
Original file line number Diff line number Diff line change
@@ -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()
203 changes: 203 additions & 0 deletions tests/test_route.py
Original file line number Diff line number Diff line change
@@ -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)
56 changes: 56 additions & 0 deletions weconnect/elements/controls.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
import logging
import json
from typing import Optional, Union
import requests

from weconnect.addressable import AddressableObject, ChangeableAttribute
from weconnect.elements.control_operation import ControlOperation, AccessControlOperation, HonkAndFlashControlOperation
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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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})')
Loading