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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Tobias Eidelpes 2021-06-17 22:26:42 +02:00
commit e66f1fcab8
1 changed files with 27 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
components/orchestration/orchestrator.py
View File

@ -88,7 +88,7 @@ class Orchestrator:
"""
Gets the daf object's pickle binary dump.
Unpickle, calculate new target velocity based on daf and current tl data and
respond new target velicity for this vehicle.
respond new target velocity for this vehicle.
:param pickle_binary: daf object pickle binary dump
"""
@ -100,6 +100,31 @@ class Orchestrator:
params={'lat': loc.latitude, 'lon': loc.longitude})
traffic_lights_geo = response.json()
current_vel = received_daf_object.velocity
target_vel = self._compute_velocity(traffic_lights_geo, current_vel)
response_channel = self._velocity_mbs[received_daf_object.vehicle_identification_number]
print('Target velocity: {}'.format(target_vel))
response_channel.send(pickle.dumps(
TargetVelocity(vin=received_daf_object.vehicle_identification_number, target_velocity=target_vel,
timestamp=datetime.now())))
def handle_tl_state_receive(self, msg):
"""
Gets a dict full of traffic light state information:
{
'tlid': str,
'color': TrafficLightColor,
'last_switch': datetime
}
Updates internal information about available TLs
"""
tl_state: TrafficLightState = pickle.loads(msg)
self.tls[tl_state.tlid]['color'] = tl_state.color
self.tls[tl_state.tlid]['last_switch'] = tl_state.last_switch
print(tl_state)
def _compute_velocity(self, traffic_lights_geo, current_vel):
target_vel = 130 * SCALING
print('Nearest traffic lights: {}'.format(traffic_lights_geo['cursor']))
@ -148,23 +173,4 @@ class Orchestrator:
i = i + 2
target_vel = floor(speed_needed_max)
response_channel = self._velocity_mbs[received_daf_object.vehicle_identification_number]
print('Target velocity: {}'.format(target_vel))
response_channel.send(pickle.dumps(
TargetVelocity(vin=received_daf_object.vehicle_identification_number, target_velocity=target_vel,
timestamp=datetime.now())))
def handle_tl_state_receive(self, msg):
"""
Gets a dict full of traffic light state information:
{
'tlid': str,
'color': TrafficLightColor,
'last_switch': datetime
}
Updates internal information about available TLs
"""
tl_state: TrafficLightState = pickle.loads(msg)
self.tls[tl_state.tlid]['color'] = tl_state.color
self.tls[tl_state.tlid]['last_switch'] = tl_state.last_switch
print(tl_state)
return target_vel