Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
rotation = action.get('rotation')
if rotation is not None and type(rotation) != dict:
action['rotation'] = {}
action['rotation']['y'] = rotation
if should_fail:
new_event = copy.deepcopy(self.last_event)
new_event.metadata['lastActionSuccess'] = False
self.last_event = new_event
return new_event
assert self.request_queue.empty()
self.response_queue.put_nowait(action)
self.last_event = queue_get(self.request_queue)
if not self.last_event.metadata['lastActionSuccess'] and self.last_event.metadata['errorCode'] == 'InvalidAction':
raise ValueError(self.last_event.metadata['errorMessage'])
if raise_for_failure:
assert self.last_event.metadata['lastActionSuccess']
return self.last_event
def reset(self, scene='FloorPlan_Train1_1'):
if re.match(r'^FloorPlan[0-9]+$', scene):
scene = scene + "_physics"
self.response_queue.put_nowait(dict(action='Reset', sceneName=scene, sequenceId=0))
self.last_event = queue_get(self.request_queue) # can this be deleted?
self.last_event = self.step(action='Initialize', **self.initialization_parameters)
return self.last_event
self.check_x_display(env['DISPLAY'])
if not self.local_executable_path:
self.download_binary()
self.lock_release()
self.prune_releases()
unity_thread = threading.Thread(
target=self._start_unity_thread,
args=(env, player_screen_width, player_screen_height, host, port, image_name))
unity_thread.daemon = True
unity_thread.start()
# receive the first request
self.last_event = queue_get(self.request_queue)
return self.last_event