order execution

This commit is contained in:
Aleksander Szamałek 2022-05-25 14:33:37 +02:00 committed by Aleksander Szamałek
parent c568bc02df
commit ed1262930c
3 changed files with 122 additions and 94 deletions

View File

@ -1,18 +1,37 @@
from typing import Tuple, List
from AgentBase import AgentBase
from PatchAgent import PatchAgent
from PatchType import PatchType
from data.Item import Item
from data.Order import Order
from data.enum.Direction import Direction
from data.enum.ItemType import ItemType
from decision.Action import Action
from decision.ActionType import ActionType
from pathfinding.PathFinderState import PathFinderState
from pathfinding.PathfinderOnStates import PathFinderOnStates
from util.PathDefinitions import GridLocation, GridWithWeights
class ForkliftAgent(AgentBase):
def __init__(self, model):
def __init__(self, model, game_constants, client_delivery: PatchAgent, drop_off: PatchAgent,
graph: GridWithWeights):
super().__init__(model)
self.action_queue: List[Action] = []
self.current_position = Tuple[int, int]
self.current_rotation = Direction.right
self.orderList = []
self.fulfilled_orders: List[Order] = []
self.client_delivery: PatchAgent = client_delivery
self.drop_off: PatchAgent = drop_off
self.graph = graph
self.game_constants = game_constants
self.current_order: Order = None
self.current_item = None
self.item_station_completed = False
self.provided_items: List[Item] = []
def queue_movement_actions(self, movement_actions: List[Action]):
self.action_queue.extend(movement_actions)
@ -55,8 +74,88 @@ class ForkliftAgent(AgentBase):
print("move {} --> {}".format(self.current_position, action_type))
self.current_position = (self.current_position[0] - 1, self.current_position[1])
def plan_actions(self):
if len(self.current_order.items) > 0:
i = self.current_order.items.pop(0)
if self.current_item is None:
self.provided_items.clear()
self.current_item = i
print("PLAN MOVEMENT")
# get item
pathFinder = PathFinderOnStates(
self.game_constants,
self.drop_off.location,
PathFinderState(self.current_position,
self.current_rotation,
0,
Action(ActionType.NONE),
[])
)
actions = pathFinder.get_action_list()
self.queue_movement_actions(actions)
elif not self.item_station_completed:
# go through station
packing_station: GridLocation = None
stations = dict(self.graph.packingStations)
if i.real_type == ItemType.PASTA:
packing_station = stations[PatchType.packingA]
elif i.real_type == ItemType.EGG:
packing_station = stations[PatchType.packingB]
elif i.real_type == ItemType.PIZZA:
packing_station = stations[PatchType.packingC]
pathFinder = PathFinderOnStates(
self.game_constants,
packing_station,
PathFinderState(self.current_position,
self.current_rotation,
0,
Action(
desired_item=i,
action_type=ActionType.PICK_ITEM
),
[])
)
actions = pathFinder.get_action_list()
self.queue_movement_actions(actions)
self.item_station_completed = True
else:
# go to client delivery area
pathFinder = PathFinderOnStates(
self.game_constants,
self.client_delivery.location,
PathFinderState(self.current_position,
self.current_rotation,
0,
Action(ActionType.NONE),
[])
)
actions = pathFinder.get_action_list()
self.queue_movement_actions(actions)
self.queue_movement_actions(
[Action(ActionType.DROP_ITEM)]
)
self.current_item = None
self.provided_items.append(self.current_item)
self.item_station_completed = False
def step(self) -> None:
self.move()
if len(self.action_queue) > 0:
self.move()
elif len(self.orderList) > 0:
if (self.current_order is not None and len(self.current_order.items)) == 0:
self.fulfilled_orders.append(self.current_order)
self.current_order = None
if self.current_order is None:
self.current_order = self.orderList.pop(0)
self.plan_actions()
def creation_log(self):
print("Created Forklift Agent [id: {}]".format(self.unique_id))

View File

@ -13,8 +13,6 @@ from PatchType import PatchType
from data.GameConstants import GameConstants
from data.Item import Item
from data.Order import Order
from data.enum.Direction import Direction
from data.enum.ItemType import ItemType
from decision.Action import Action
from decision.ActionType import ActionType
from pathfinding.PathfinderOnStates import PathFinderOnStates, PathFinderState
@ -41,11 +39,6 @@ class GameModel(Model):
self.client_delivery: PatchAgent = None
self.drop_off: PatchAgent = None
self.agents = [AgentBase]
self.forklift_agent = ForkliftAgent(self)
self.schedule.add(self.forklift_agent)
self.agents.append(self.forklift_agent)
self.graph = graph
self.game_constants = GameConstants(
@ -55,11 +48,27 @@ class GameModel(Model):
graph.puddles
)
self.agents = [AgentBase]
self.forklift_agent = ForkliftAgent(
self,
self.game_constants,
self.client_delivery,
self.drop_off,
self.graph
)
self.schedule.add(self.forklift_agent)
self.agents.append(self.forklift_agent)
# INITIALIZATION #
print("############## INITIALIZATION ##############")
self.phase = Phase.INIT
self.initialize_grid(graph)
self.orderList: List[Order] = InitialStateFactory.generate_order_list(3)
self.fulfilled_orders: List[Order] = []
self.forklift_agent.orderList = self.orderList
self.forklift_agent.fulfilled_orders = self.fulfilled_orders
print("############## RECOGNISE ITEMS ##############")
self.phase = Phase.ITEM_RECOGNITION
@ -99,11 +108,13 @@ class GameModel(Model):
self.grid.place_agent(agent, agent.location)
self.agents.append(agent)
self.client_delivery = agent
self.forklift_agent.client_delivery = self.client_delivery
agent = PatchAgent(self, (0, int(self.grid.height / 2)), PatchType.dropOff)
self.grid.place_agent(agent, agent.location)
self.agents.append(agent)
self.drop_off = agent
self.forklift_agent.drop_off = self.drop_off
def place_walls_agents(self, walls: List[GridLocation]):
for w in walls:
@ -145,93 +156,10 @@ class GameModel(Model):
# TODO GENERICS SORTING
sorted(self.orderList, key=lambda x: len(x.items))
print("FINISHED CLIENT ORDER SORTING")
self.phase = Phase.PLAN_MOVEMENT
if self.phase == Phase.PLAN_MOVEMENT:
if len(self.orderList) > 0:
self.plan_actions()
self.phase = Phase.EXECUTION
else:
print("NO MORE ORDERS")
self.phase = Phase.EXECUTION
if self.phase == Phase.EXECUTION:
print("Execution")
if (len(self.forklift_agent.action_queue) == 0):
self.phase = Phase.PLAN_MOVEMENT
def plan_actions(self):
order = self.orderList.pop(0)
print("PLAN MOVEMENT")
for i in order.items:
if self.first:
# get item
pathFinder = PathFinderOnStates(
self.game_constants,
self.drop_off.location,
PathFinderState(self.forklift_agent.current_position,
self.forklift_agent.current_rotation,
0,
Action(ActionType.NONE),
[])
)
actions = pathFinder.get_action_list()
self.forklift_agent.queue_movement_actions(actions)
self.first = False
else:
# get item
pathFinder = PathFinderOnStates(
self.game_constants,
self.drop_off.location,
PathFinderState(self.client_delivery.location,
Direction.right,
0,
Action(ActionType.NONE),
[])
)
actions = pathFinder.get_action_list()
self.forklift_agent.queue_movement_actions(actions)
self.first = False
# go through station
packing_station: GridLocation = None
stations = dict(self.graph.packingStations)
if i.real_type == ItemType.PASTA:
packing_station = stations[PatchType.packingA]
elif i.real_type == ItemType.EGG:
packing_station = stations[PatchType.packingB]
elif i.real_type == ItemType.PIZZA:
packing_station = stations[PatchType.packingC]
pathFinder = PathFinderOnStates(
self.game_constants,
packing_station,
PathFinderState(self.drop_off.location,
Direction.left,
0,
Action(
desired_item=i,
action_type=ActionType.PICK_ITEM
),
[])
)
actions = pathFinder.get_action_list()
self.forklift_agent.queue_movement_actions(actions)
# go to client delivery area
pathFinder = PathFinderOnStates(
self.game_constants,
self.client_delivery.location,
PathFinderState(packing_station,
Direction.right,
0,
Action(ActionType.NONE),
[])
)
actions = pathFinder.get_action_list()
self.forklift_agent.queue_movement_actions(actions)
self.forklift_agent.queue_movement_actions(
[Action(ActionType.DROP_ITEM)]
)
def recognise_item(self, item: Item):
# TODO IMAGE PROCESSING

View File

@ -69,9 +69,10 @@ if __name__ == '__main__':
provided_itesm = DisplayAttributeElement("provided_items")
recognised_items = DisplayAttributeElement("recognised_items")
ordersText = DisplayAttributeElement("orderList")
fulfilled_orders = DisplayAttributeElement("fulfilled_orders")
server = ModularServer(GameModel,
[grid, readyText, provided_itesm, recognised_items],
[grid, readyText, provided_itesm, recognised_items, ordersText, fulfilled_orders],
"Automatyczny Wózek Widłowy",
{"width": gridHeight, "height": gridWidth, "graph": diagram}, )