zenon/dse-assignment
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Set parameters required by assignment

Browse Source
This commit is contained in:
Tobias Eidelpes 2021-06-13 23:00:58 +02:00
parent 8b0a60d9ab
commit 086aada295
2 changed files with 37 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
components/i_feed/devices/vehicle.py
View File

@ -27,7 +27,7 @@ UPDATE_INTERVAL = 1
# At x km the NCE shall happen # At x km the NCE shall happen
NCE_KM = 2.4 NCE_KM = 2.4
# Time in seconds to recover from NCE (will be scaled) # Time in seconds to recover from NCE (will be scaled)
TIME_TO_RECOVER = 5 TIME_TO_RECOVER = 15
# Resets vehicle at km x # Resets vehicle at km x
RESET_KM = 4 RESET_KM = 4

56
components/orchestration/orchestrator.py
View File

@ -3,6 +3,7 @@ import pickle
import sys import sys
from typing import List, Dict from typing import List, Dict
from datetime import datetime, timedelta from datetime import datetime, timedelta
from math import floor, ceil
import requests import requests
from dse_shared_libs import daf, traffic_light_state, traffic_light_color, target_velocity from dse_shared_libs import daf, traffic_light_state, traffic_light_color, target_velocity
@ -99,34 +100,49 @@ class Orchestrator:
params={'lat': loc.latitude, 'lon': loc.longitude}) params={'lat': loc.latitude, 'lon': loc.longitude})
traffic_lights_geo = response.json() traffic_lights_geo = response.json()
current_vel = received_daf_object.velocity current_vel = received_daf_object.velocity
target_vel = 130 target_vel = 130 * SCALING
print('Nearest traffic lights: {}'.format(traffic_lights_geo['cursor'])) print('Nearest traffic lights: {}'.format(traffic_lights_geo['cursor']))
for traffic_light in traffic_lights_geo['cursor']: for traffic_light in traffic_lights_geo['cursor']:
# Should only ever contain one traffic light # Should only ever contain one traffic light (the next in line)
tl_id = traffic_light['id'] tl_id = traffic_light['id']
distance = traffic_light['calculatedRange'] distance = traffic_light['calculatedRange']
next_switch_time = self.tls[tl_id]['last_switch'] + timedelta(seconds=self.tls[tl_id]['switching_time']) switching_time = self.tls[tl_id]['switching_time'] / float(SCALING)
time_until_switch = next_switch_time - datetime.now() # Time until next switch must be scaled accordingly
next_switch_time = self.tls[tl_id]['last_switch'] + timedelta(seconds=switching_time)
time_until_switch = (next_switch_time - datetime.now()).total_seconds()
print('Distance to TL: {}'.format(distance)) print('Distance to TL: {}'.format(distance))
print('Time until switch in seconds: {}'.format(time_until_switch.total_seconds())) print('Time until switch in seconds: {}'.format(time_until_switch))
if self.tls[tl_id]['color'] is TrafficLightColor.RED: if self.tls[tl_id]['color'] is TrafficLightColor.RED:
# Ceil because we want to get there AFTER the light changed speed_needed_max = (distance / float(time_until_switch)) * 3.6
speed_needed = (distance / float(time_until_switch.total_seconds())) * 3.6 if speed_needed_max < 130 * SCALING:
if current_vel > speed_needed: if current_vel < speed_needed_max:
print('Current velocity too high ({}), adjusting to {}'.format(current_vel, speed_needed)) target_vel = current_vel
target_vel = speed_needed else:
target_vel = floor(speed_needed_max)
else: else:
print('Current velocity lower ({})'.format(current_vel)) # Cannot make it on next green
target_vel = current_vel i = 2
while speed_needed_max > 130 * SCALING:
next_green_phase_start = time_until_switch + switching_time * i
speed_needed_max = (distance / float(next_green_phase_start)) * 3.6
i = i + 2
target_vel = floor(speed_needed_max)
else: else:
# Check if we can cross in time with max speed (130) # Check if we can reach TL in time
speed_needed = (distance / float(time_until_switch.total_seconds())) * 3.6 speed_needed_min = (distance / float(time_until_switch)) * 3.6
if speed_needed >= 130: if speed_needed_min < 130 * SCALING:
# Cannot make it in time, wait for next green if current_vel < speed_needed_min:
print('Wait on green -> red -> green switch') target_vel = 130 * SCALING
time_until_green = time_until_switch + timedelta(seconds=self.tls[tl_id]['switching_time']) else:
target_vel = (distance / float(time_until_green.total_seconds())) * 3.6 target_vel = current_vel
print('Speed lower than 130: {}'.format(target_vel)) else:
i = 1
speed_needed_max = 132 * SCALING
while speed_needed_max > 130 * SCALING:
next_green_phase_start = time_until_switch + switching_time * i
speed_needed_max = (distance / float(next_green_phase_start)) * 3.6
i = i + 2
target_vel = floor(speed_needed_max)
response_channel = self._velocity_mbs[received_daf_object.vehicle_identification_number] response_channel = self._velocity_mbs[received_daf_object.vehicle_identification_number]
print('Target velocity: {}'.format(target_vel)) print('Target velocity: {}'.format(target_vel))