lpe234 · June 27, 2024 06:58
diff --git a/README.md b/README.md
diff --git a/main.py b/main.py
 import abc
 import datetime
 import enum
 import random
 import time
 from typing import Union

 from loguru import logger
 from simpy import Resource, Environment

 # 模拟任务 数量
 TASK_SIZE = 10
 # 巷道 数量
 LANE_SIZE = 5
 # 货架大小
 MAX_LAYER = 30
 MAX_COLUMN = 20
 # 耗时 每层、每列
 TIME_COST_PRE_LAYER = 2
 TIME_COST_PRE_COLUMN = 1


 class TaskStatus(enum.Enum):
    """
    任务状态
    """
    READY = 1
    RUNNING = 2
    DONE = 3


 class Direction(enum.Enum):
    """
    方向
    """
    LEFT = 1
    RIGHT = 2


 class Position(object):
    """
    货物位置
    """
    # 巷道
    lane: int = None
    # 方向
    direction: Direction = None
    # 位置 (层)
    layer: int = None
    # 位置 (列)
    column: int = None

    def __init__(self, lane: int, direction: Direction, layer: int, column: int):
        self.lane = lane
        self.direction = direction
        self.layer = layer
        self.column = column

    def __str__(self) -> str:
        return f'lane: {self.lane:02}({self.direction.value}), layer: {self.layer:02}, column: {self.column:02}'


 class Task(object):
    """
    取货任务
    """

    def __init__(self, tid: str, name: str, position: Position):
        self.tid = tid
        self.name = name
        self.position = position
        self.status = TaskStatus.READY


 class Object(object):
    """
    对象
    """
    name: str = None
    env: Environment = None

    def __init__(self, name: str, env: Environment):
        self.name = name
        self.env = env


 class System(Object, abc.ABC):
    """
    系统
    """

    def __init__(self, name: str, env: Environment):
        super().__init__(name, env)
        self.task_list = []
        self.running = False

    def push_task(self, task: Task):
        logger.debug(f'{self.name} 接收到任务 {task.name}')
        self.task_list.append(task)

    def run(self):
        while self.running:
            if not self.task_list:
                yield self.env.timeout(1)
                continue
            task = self.choose_task()
            # 处理该任务
            yield self.env.process(self.run_task(task))

    def choose_task(self) -> Union[Task, None]:
        """
        当任务列表有任务时，根据一定的策略从任务列表选取一个出来
        :return:
        """
        if self.task_list:
            return self.task_list.pop(0)
        return None

    @abc.abstractmethod
    def run_task(self, task: Task):
        """
        具体的任务执行
        :param task:
        :return:
        """
        pass

    def start(self):
        logger.debug(f'{self.name} 开始启动')
        self.running = True
        self.env.process(self.run())

    def stop(self):
        self.running = False
        logger.debug(f'{self.name} 停止运动')


 class Helpers(object):

    @staticmethod
    def time_cost(task: Task) -> float:
        """
        搬运耗时
        假设
        - 每次机器人都在出口
        - 每次只搬运一个货物
        - 每次搬运完毕后，在出口等待指令
        :param task:
        :return:
        """
        pos = task.position
        cost = pos.layer * TIME_COST_PRE_LAYER + pos.column * TIME_COST_PRE_COLUMN
        return cost * 2

    @staticmethod
    def ts_str(ts: float):
        return datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')


 class Robot(System):
    """
    取货机器人
    """

    def __init__(self, name: str, env: Environment):
        super().__init__(name, env)
        self.resource = Resource(env)

    def run_task(self, task: Task):
        time_cost = Helpers.time_cost(task)
        with self.resource.request() as req:
            logger.debug(
                f'[ROBOT] {self.name} 开始任务执行 task: {task.name}({task.position}), time: {Helpers.ts_str(self.env.now)}')
            yield req
            yield self.env.timeout(time_cost)
            task.status = TaskStatus.DONE
        logger.debug(
            f'[ROBOT] {self.name} 完成任务执行 task: {task.name}({task.position}), time: {Helpers.ts_str(self.env.now)}, '
            f'cost: {time_cost}')


 class Shelf(System):
    """
    货架
    """

    def __init__(self, name: str, robot: Robot, env: Environment):
        super().__init__(name, env)
        self.env = env
        self.robot = robot

    def run_task(self, task: Task):
        yield self.env.timeout(0)
        self.robot.push_task(task)

    def start(self):
        self.robot.start()
        super().start()

    def stop(self):
        super().stop()
        self.robot.stop()


 class Warehouse(System):
    """
    仓库
    """

    def __init__(self, name: str, env: Environment):
        super().__init__(name, env)
        self.shelf_list: list[Shelf] = []
        self.__init()

    def __init(self):
        """
        初始化 仓库
        :return:
        """
        for idx in range(LANE_SIZE):
            index = idx + 1
            shelf = Shelf(f'货架_{index}', Robot(f'机器人_{index:02}', self.env), self.env)
            self.shelf_list.append(shelf)
            setattr(self, self.shelf_name(index), shelf)

    def run_task(self, task: Task):
        task.status = TaskStatus.RUNNING
        # 此处需要判断需要哪个货架执行
        pos = task.position
        shelf: Shelf = getattr(self, self.shelf_name(pos.lane))
        shelf.push_task(task)
        #
        yield self.env.timeout(0)

    @staticmethod
    def shelf_name(idx: int):
        return f'shelf_{idx}'

    def start(self):
        for s in self.shelf_list:
            s.start()
        super().start()

    def stop(self):
        super().stop()
        for s in self.shelf_list:
            s.stop()


 def generate_task(size: int) -> list[Task]:
    task_list = []
    for idx in range(size):
        lane = random.randint(1, LANE_SIZE)
        direction = random.choice([Direction.RIGHT, Direction.LEFT])
        layer = random.randint(1, MAX_LAYER)
        column = random.randint(1, MAX_COLUMN)
        pos = Position(lane, direction, layer, column)
        t = Task(f'task_{idx + 1:02}', f'任务_{idx + 1:02}', pos)
        task_list.append(t)
    return task_list


 def send_task(task_list: list[Task], wh: Warehouse, env: Environment):
    #
    for t in task_list:
        wh.push_task(t)
    yield env.timeout(0)


 def watch_task(task_list: list[Task], wh: Warehouse, env: Environment):
    while True:
        if [_ for _ in task_list if not _.status == TaskStatus.DONE]:
            yield env.timeout(1)
            continue
        wh.stop()
        break


 def main():
    base_time = time.time()
    env = Environment(initial_time=base_time)
    wh = Warehouse('仓库', env)
    #
    task_list = generate_task(TASK_SIZE)
    #
    env.process(wh.run())
    env.process(send_task(task_list, wh, env))
    env.process(watch_task(task_list, wh, env))
    #
    wh.start()
    #
    env.run()


 if __name__ == '__main__':
    main()
	import abc
	import datetime
	import enum
	import random
	import time
	from typing import Union

	from loguru import logger
	from simpy import Resource, Environment

	# 模拟任务数量
	TASK_SIZE = 10
	# 巷道数量
	LANE_SIZE = 5
	# 货架大小
	MAX_LAYER = 30
	MAX_COLUMN = 20
	# 耗时每层、每列
	TIME_COST_PRE_LAYER = 2
	TIME_COST_PRE_COLUMN = 1


	class TaskStatus(enum.Enum):
	"""
	任务状态
	"""
	READY = 1
	RUNNING = 2
	DONE = 3


	class Direction(enum.Enum):
	"""
	方向
	"""
	LEFT = 1
	RIGHT = 2


	class Position(object):
	"""
	货物位置
	"""
	# 巷道
	lane: int = None
	# 方向
	direction: Direction = None
	# 位置 (层)
	layer: int = None
	# 位置 (列)
	column: int = None

	def __init__(self, lane: int, direction: Direction, layer: int, column: int):
	self.lane = lane
	self.direction = direction
	self.layer = layer
	self.column = column

	def __str__(self) -> str:
	return f'lane: {self.lane:02}({self.direction.value}), layer: {self.layer:02}, column: {self.column:02}'


	class Task(object):
	"""
	取货任务
	"""

	def __init__(self, tid: str, name: str, position: Position):
	self.tid = tid
	self.name = name
	self.position = position
	self.status = TaskStatus.READY


	class Object(object):
	"""
	对象
	"""
	name: str = None
	env: Environment = None

	def __init__(self, name: str, env: Environment):
	self.name = name
	self.env = env


	class System(Object, abc.ABC):
	"""
	系统
	"""

	def __init__(self, name: str, env: Environment):
	super().__init__(name, env)
	self.task_list = []
	self.running = False

	def push_task(self, task: Task):
	logger.debug(f'{self.name} 接收到任务 {task.name}')
	self.task_list.append(task)

	def run(self):
	while self.running:
	if not self.task_list:
	yield self.env.timeout(1)
	continue
	task = self.choose_task()
	# 处理该任务
	yield self.env.process(self.run_task(task))

	def choose_task(self) -> Union[Task, None]:
	"""
	当任务列表有任务时，根据一定的策略从任务列表选取一个出来
	:return:
	"""
	if self.task_list:
	return self.task_list.pop(0)
	return None

	@abc.abstractmethod
	def run_task(self, task: Task):
	"""
	具体的任务执行
	:param task:
	:return:
	"""
	pass

	def start(self):
	logger.debug(f'{self.name} 开始启动')
	self.running = True
	self.env.process(self.run())

	def stop(self):
	self.running = False
	logger.debug(f'{self.name} 停止运动')


	class Helpers(object):

	@staticmethod
	def time_cost(task: Task) -> float:
	"""
	搬运耗时
	假设
	- 每次机器人都在出口
	- 每次只搬运一个货物
	- 每次搬运完毕后，在出口等待指令
	:param task:
	:return:
	"""
	pos = task.position
	cost = pos.layer * TIME_COST_PRE_LAYER + pos.column * TIME_COST_PRE_COLUMN
	return cost * 2

	@staticmethod
	def ts_str(ts: float):
	return datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')


	class Robot(System):
	"""
	取货机器人
	"""

	def __init__(self, name: str, env: Environment):
	super().__init__(name, env)
	self.resource = Resource(env)

	def run_task(self, task: Task):
	time_cost = Helpers.time_cost(task)
	with self.resource.request() as req:
	logger.debug(
	f'[ROBOT] {self.name} 开始任务执行 task: {task.name}({task.position}), time: {Helpers.ts_str(self.env.now)}')
	yield req
	yield self.env.timeout(time_cost)
	task.status = TaskStatus.DONE
	logger.debug(
	f'[ROBOT] {self.name} 完成任务执行 task: {task.name}({task.position}), time: {Helpers.ts_str(self.env.now)}, '
	f'cost: {time_cost}')


	class Shelf(System):
	"""
	货架
	"""

	def __init__(self, name: str, robot: Robot, env: Environment):
	super().__init__(name, env)
	self.env = env
	self.robot = robot

	def run_task(self, task: Task):
	yield self.env.timeout(0)
	self.robot.push_task(task)

	def start(self):
	self.robot.start()
	super().start()

	def stop(self):
	super().stop()
	self.robot.stop()


	class Warehouse(System):
	"""
	仓库
	"""

	def __init__(self, name: str, env: Environment):
	super().__init__(name, env)
	self.shelf_list: list[Shelf] = []
	self.__init()

	def __init(self):
	"""
	初始化仓库
	:return:
	"""
	for idx in range(LANE_SIZE):
	index = idx + 1
	shelf = Shelf(f'货架_{index}', Robot(f'机器人_{index:02}', self.env), self.env)
	self.shelf_list.append(shelf)
	setattr(self, self.shelf_name(index), shelf)

	def run_task(self, task: Task):
	task.status = TaskStatus.RUNNING
	# 此处需要判断需要哪个货架执行
	pos = task.position
	shelf: Shelf = getattr(self, self.shelf_name(pos.lane))
	shelf.push_task(task)
	#
	yield self.env.timeout(0)

	@staticmethod
	def shelf_name(idx: int):
	return f'shelf_{idx}'

	def start(self):
	for s in self.shelf_list:
	s.start()
	super().start()

	def stop(self):
	super().stop()
	for s in self.shelf_list:
	s.stop()


	def generate_task(size: int) -> list[Task]:
	task_list = []
	for idx in range(size):
	lane = random.randint(1, LANE_SIZE)
	direction = random.choice([Direction.RIGHT, Direction.LEFT])
	layer = random.randint(1, MAX_LAYER)
	column = random.randint(1, MAX_COLUMN)
	pos = Position(lane, direction, layer, column)
	t = Task(f'task_{idx + 1:02}', f'任务_{idx + 1:02}', pos)
	task_list.append(t)
	return task_list


	def send_task(task_list: list[Task], wh: Warehouse, env: Environment):
	#
	for t in task_list:
	wh.push_task(t)
	yield env.timeout(0)


	def watch_task(task_list: list[Task], wh: Warehouse, env: Environment):
	while True:
	if [_ for _ in task_list if not _.status == TaskStatus.DONE]:
	yield env.timeout(1)
	continue
	wh.stop()
	break


	def main():
	base_time = time.time()
	env = Environment(initial_time=base_time)
	wh = Warehouse('仓库', env)
	#
	task_list = generate_task(TASK_SIZE)
	#
	env.process(wh.run())
	env.process(send_task(task_list, wh, env))
	env.process(watch_task(task_list, wh, env))
	#
	wh.start()
	#
	env.run()


	if __name__ == '__main__':
	main()