163 lines
6.9 KiB
Python
163 lines
6.9 KiB
Python
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, 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.current_order_delivered_items: List[Item] = []
|
|
self.ready_for_execution = False
|
|
|
|
def queue_movement_actions(self, movement_actions: List[Action]):
|
|
self.action_queue.extend(movement_actions)
|
|
|
|
def move(self):
|
|
if len(self.action_queue) > 0:
|
|
action = self.action_queue.pop(0)
|
|
action_type = action.action_type
|
|
|
|
if action_type == ActionType.ROTATE_UP:
|
|
print("rotate {} --> {}".format(self.current_rotation, action_type))
|
|
self.current_rotation = Direction.top
|
|
|
|
elif action_type == ActionType.ROTATE_RIGHT:
|
|
print("rotate {} --> {}".format(self.current_rotation, action_type))
|
|
self.current_rotation = Direction.right
|
|
|
|
elif action_type == ActionType.ROTATE_DOWN:
|
|
print("rotate {} --> {}".format(self.current_rotation, action_type))
|
|
self.current_rotation = Direction.down
|
|
|
|
elif action_type == ActionType.ROTATE_LEFT:
|
|
print("rotate {} --> {}".format(self.current_rotation, action_type))
|
|
self.current_rotation = Direction.left
|
|
|
|
elif action_type == ActionType.MOVE:
|
|
if self.current_rotation == Direction.top:
|
|
print("move {} --> {}".format(self.current_position, action_type))
|
|
self.current_position = (self.current_position[0], self.current_position[1] + 1)
|
|
|
|
elif self.current_rotation == Direction.down:
|
|
print("move {} --> {}".format(self.current_position, action_type))
|
|
self.current_position = (self.current_position[0], self.current_position[1] - 1)
|
|
|
|
elif self.current_rotation == Direction.right:
|
|
print("move {} --> {}".format(self.current_position, action_type))
|
|
self.current_position = (self.current_position[0] + 1, self.current_position[1])
|
|
|
|
elif self.current_rotation == Direction.left:
|
|
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.current_order_delivered_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.SHELF:
|
|
packing_station = stations[PatchType.packingShelf]
|
|
elif i.real_type == ItemType.REFRIGERATOR:
|
|
packing_station = stations[PatchType.packingRefrigerator]
|
|
elif i.real_type == ItemType.DOOR:
|
|
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.current_order_delivered_items.append(self.current_item)
|
|
self.item_station_completed = False
|
|
|
|
def step(self) -> None:
|
|
if len(self.action_queue) > 0:
|
|
self.move()
|
|
elif self.ready_for_execution and 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))
|