steffen/FIKS_Entwicklungsumgebung
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

No more Monitor,

Browse Source 
env hparams pickeling,
pomdp,
now training and learning
This commit is contained in:
steffen-illium
2021-06-01 16:38:55 +02:00
parent 55b409c72f
commit ff9846eb54
6 changed files with 95 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
environments/factory/simple_factory.py
View File

@ -34,27 +34,27 @@ class SimpleFactory(BaseFactory):
return self._actions[action] == CLEAN_UP_ACTION
def __init__(self, *args, dirt_properties: DirtProperties, verbose=False, **kwargs):
self._dirt_properties = dirt_properties
self.dirt_properties = dirt_properties
self.verbose = verbose
self.max_dirt = 20
super(SimpleFactory, self).__init__(*args, **kwargs)
self.state_slices.register_additional_items('dirt')
self.renderer = None # expensive - don't use it when not required !
self._state_slices.register_additional_items('dirt')
self._renderer = None # expensive - don't use it when not required !
def render(self):
if not self.renderer: # lazy init
height, width = self.state.shape[1:]
self.renderer = Renderer(width, height, view_radius=2)
if not self._renderer: # lazy init
height, width = self._state.shape[1:]
self._renderer = Renderer(width, height, view_radius=self.pomdp_radius)
dirt = [Entity('dirt', [x, y], min(0.15+self.state[DIRT_INDEX, x, y], 1.5), 'scale')
for x, y in np.argwhere(self.state[DIRT_INDEX] > h.IS_FREE_CELL)]
walls = [Entity('wall', pos) for pos in np.argwhere(self.state[h.LEVEL_IDX] > h.IS_FREE_CELL)]
dirt = [Entity('dirt', [x, y], min(0.15 + self._state[DIRT_INDEX, x, y], 1.5), 'scale')
for x, y in np.argwhere(self._state[DIRT_INDEX] > h.IS_FREE_CELL)]
walls = [Entity('wall', pos) for pos in np.argwhere(self._state[h.LEVEL_IDX] > h.IS_FREE_CELL)]
def asset_str(agent):
if any([x is None for x in [self.state_slices[j] for j in agent.collisions]]):
if any([x is None for x in [self._state_slices[j] for j in agent.collisions]]):
print('error')
cols = ' '.join([self.state_slices[j] for j in agent.collisions])
cols = ' '.join([self._state_slices[j] for j in agent.collisions])
if 'agent' in cols:
return 'agent_collision'
elif not agent.action_valid or 'level' in cols or 'agent' in cols:
@ -65,38 +65,38 @@ class SimpleFactory(BaseFactory):
return f'agent{agent.i + 1}'
agents = {f'agent{i+1}': [Entity(asset_str(agent), agent.pos)]
for i, agent in enumerate(self.agent_states)}
self.renderer.render(OrderedDict(dirt=dirt, wall=walls, **agents))
for i, agent in enumerate(self._agent_states)}
self._renderer.render(OrderedDict(dirt=dirt, wall=walls, **agents))
def spawn_dirt(self) -> None:
if not np.argwhere(self.state[DIRT_INDEX] != h.IS_FREE_CELL).shape[0] > self._dirt_properties.max_global_amount:
if not np.argwhere(self._state[DIRT_INDEX] != h.IS_FREE_CELL).shape[0] > self.dirt_properties.max_global_amount:
free_for_dirt = self.free_cells(excluded_slices=DIRT_INDEX)
# randomly distribute dirt across the grid
n_dirt_tiles = int(random.uniform(0, self._dirt_properties.max_spawn_ratio) * len(free_for_dirt))
n_dirt_tiles = int(random.uniform(0, self.dirt_properties.max_spawn_ratio) * len(free_for_dirt))
for x, y in free_for_dirt[:n_dirt_tiles]:
new_value = self.state[DIRT_INDEX, x, y] + self._dirt_properties.gain_amount
self.state[DIRT_INDEX, x, y] = max(new_value, self._dirt_properties.max_local_amount)
new_value = self._state[DIRT_INDEX, x, y] + self.dirt_properties.gain_amount
self._state[DIRT_INDEX, x, y] = max(new_value, self.dirt_properties.max_local_amount)
else:
pass
def clean_up(self, pos: (int, int)) -> ((int, int), bool):
new_dirt_amount = self.state[DIRT_INDEX][pos] - self._dirt_properties.clean_amount
new_dirt_amount = self._state[DIRT_INDEX][pos] - self.dirt_properties.clean_amount
cleanup_was_sucessfull: bool
if self.state[DIRT_INDEX][pos] == h.IS_FREE_CELL:
if self._state[DIRT_INDEX][pos] == h.IS_FREE_CELL:
cleanup_was_sucessfull = False
return pos, cleanup_was_sucessfull
else:
cleanup_was_sucessfull = True
self.state[DIRT_INDEX][pos] = max(new_dirt_amount, h.IS_FREE_CELL)
self._state[DIRT_INDEX][pos] = max(new_dirt_amount, h.IS_FREE_CELL)
return pos, cleanup_was_sucessfull
def step(self, actions):
_, r, done, info = super(SimpleFactory, self).step(actions)
if not self.next_dirt_spawn:
self.spawn_dirt()
self.next_dirt_spawn = self._dirt_properties.spawn_frequency
self.next_dirt_spawn = self.dirt_properties.spawn_frequency
else:
self.next_dirt_spawn -= 1
obs = self._return_state()
@ -115,17 +115,17 @@ class SimpleFactory(BaseFactory):
def reset(self) -> (np.ndarray, int, bool, dict):
_ = super().reset() # state, reward, done, info ... =
dirt_slice = np.zeros((1, *self.state.shape[1:]))
self.state = np.concatenate((self.state, dirt_slice)) # dirt is now the last slice
dirt_slice = np.zeros((1, *self._state.shape[1:]))
self._state = np.concatenate((self._state, dirt_slice)) # dirt is now the last slice
self.spawn_dirt()
self.next_dirt_spawn = self._dirt_properties.spawn_frequency
self.next_dirt_spawn = self.dirt_properties.spawn_frequency
obs = self._return_state()
return obs
def calculate_reward(self, agent_states: List[AgentState]) -> (int, dict):
# TODO: What reward to use?
current_dirt_amount = self.state[DIRT_INDEX].sum()
dirty_tiles = np.argwhere(self.state[DIRT_INDEX] != h.IS_FREE_CELL).shape[0]
current_dirt_amount = self._state[DIRT_INDEX].sum()
dirty_tiles = np.argwhere(self._state[DIRT_INDEX] != h.IS_FREE_CELL).shape[0]
info_dict = dict()
try:
@ -137,11 +137,13 @@ class SimpleFactory(BaseFactory):
for agent_state in agent_states:
cols = agent_state.collisions
list_of_collisions = [self.state_slices[entity] for entity in cols
if entity != self.state_slices.by_name("dirt")]
list_of_collisions = [self._state_slices[entity] for entity in cols
if entity != self._state_slices.by_name("dirt")]
if list_of_collisions:
self.print(f't = {self.steps}\tAgent {agent_state.i} has collisions with '
f'{list_of_collisions}')
self.print(f't = {self.steps}\tAgent {agent_state.i} has collisions with '
f'{list_of_collisions}')
if self._is_clean_up_action(agent_state.action):
if agent_state.action_valid:
reward += 1
@ -155,19 +157,19 @@ class SimpleFactory(BaseFactory):
elif self._is_moving_action(agent_state.action):
if agent_state.action_valid:
info_dict.update(movement=1)
# info_dict.update(movement=1)
reward -= 0.00
else:
info_dict.update(collision=1)
# info_dict.update(collision=1)
# self.print('collision')
reward -= 0.00
else:
info_dict.update(collision=1)
info_dict.update(no_op=1)
reward -= 0.00
for entity in cols:
if entity != self.state_slices.by_name("dirt"):
info_dict.update({f'agent_{agent_state.i}_vs_{self.state_slices[entity]}': 1})
for entity in list_of_collisions:
info_dict.update({f'agent_{agent_state.i}_vs_{self._state_slices.by_name(entity)}': 1})
info_dict.update(dirt_amount=current_dirt_amount)
info_dict.update(dirty_tile_count=dirty_tiles)