from typing import List, Dict import time, threading, sys import traceback # <--- 新增这一行 from base.controller_base import ControllerBase from configs.global_config import GlobalConfig as cfg from controllers.agv_controller import AGVController from controllers.cnc_controller import CNCController from controllers.store_controller import StoreController from controllers.communication_contorller import CommunicationController from controllers.vms_controller import VMSController from controllers.vms_path_controller import VMSPathController from conditions.task_controller import TaskController from entities.product_entity import ProductEntity class MainLogic: container: List[ControllerBase] comm_ctr: CommunicationController agv_ctr: AGVController vms_ctr: VMSController cnc_ctr: CNCController store_ctr: StoreController path_ctr: VMSPathController task_ctr: TaskController products: List[ProductEntity] """产品列表""" loop: bool invert: float thread: threading.Thread def __init__(self, orders: List[int] | None = None): self.loop = False self.invert = 0.1 self.thread = None self.comm_ctr = CommunicationController() self.agv_ctr = AGVController() self.vms_ctr = VMSController() self.cnc_ctr = CNCController() self.store_ctr = StoreController() self.path_ctr = VMSPathController() self.task_ctr = TaskController() self.container = [self.comm_ctr, self.agv_ctr, self.vms_ctr, self.cnc_ctr, self.store_ctr, self.path_ctr, self.task_ctr] for controller in self.container: controller.init(self.container) for controller in self.container: controller.start() self.task_ctr.set_orders(orders) ### def _build_products(self, raw_products) -> List[ProductEntity]: products: List[ProductEntity] = [] if isinstance(raw_products, dict): raw_products = raw_products.values() for item in raw_products: if isinstance(item, ProductEntity): products.append(item) elif isinstance(item, tuple): if len(item) >= 2 and isinstance(item[1], ProductEntity): products.append(item[1]) else: try: products.append(ProductEntity(*item)) except TypeError as e: print(f"构建 ProductEntity 失败: {item!r}, 错误: {e}") else: print(f"未知产品数据类型: {type(item)}") return products ### def start(self): #1.登录获取Token if self.comm_ctr.login(): self.vms_ctr.start_record(self.callback) else: print("登录失败!!!") def callback(self, sender, args): if self.loop: return #1.同步产线数据 self.cnc_ctr.sync_cncs_base_info() #2.同步料仓数据 self.store_ctr.sync_store_mats() #3.同步排程数据 self.products = self._build_products(self.store_ctr.sync_products()) #4.开始轮循执行任务 self.loop = True self.thread = threading.Thread(target=self.update_thread, args=()) self.thread.daemon = True self.thread.start() print(f"需要生产[{len(self.products)}]个产品!") def update_thread(self): while self.loop: try: self.refresh_tasks() except Exception as e: print(f"刷新任务失败：{str(e)}") traceback.print_exc() # <--- 新增这一行，它会打印出完整的报错层级和行号！ time.sleep(self.invert) def refresh_tasks(self): """循环刷新任务""" #1.先跳过部分不需要执行的任务 for product in self.products: if not hasattr(product, 'comp'): print(f"跳过无效产品数据: {product!r}") continue product.comp.skip_task() #2.选择最优任务 min_order = (sys.maxsize, sys.maxsize) optimal_product = None for product in self.products: order = self.task_ctr.get_order(product) if order is not None and order[0] is not None and order[1] is not None: if order[0] < min_order[0]: min_order = order optimal_product = product elif order[0] == min_order[0] and order[1] < min_order[1]: min_order = order optimal_product = product #3.执行最优任务 if optimal_product != None and self.task_ctr.run_task(optimal_product): pass else: #4.执行AGV去过度点位置的逻辑 self.task_ctr.run_agv_idle_task() def stop(self): self.loop = False if self.thread is not None: self.thread.join() for controller in self.container: controller.destroy() if __name__ == "__main__": #1.设置自定义配置 custom_config: Dict = { "lineCode": "LN260176", # MES产线编码 "stockId": 217, # MES料仓ID "userName": "15939765667", # 用户名 "password": "Qh123456" # 密码 } cfg.set_custom_config(custom_config) #2.运行调度系统 logic = MainLogic() logic.start() try: # 主线程持续运行，等待中断信号（如 Ctrl+C） while True: time.sleep(1) # 避免空转占用 CPU except KeyboardInterrupt: print("\nReceived interrupt signal. Shutting down...") finally: logic.stop()

from base.task_condition_base import TaskConditionBase from tasks.agv_idle_execution import AGVIdleCNCExecution from entities.product_entity import ProductEntity class AGVIdleCondition(TaskConditionBase): def __init__(self): super().__init__() self.name = "AGV等待" ### def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, AGVIdleCNCExecution | None] | None: #1.一台AGV不需要特殊处理（） agvs = self.agv_ctr.agvs if len(agvs) <= 1: return None #2.多台AGV处理逻辑（） for agv in agvs: if agv.comp.is_idle() is False: continue curr_pos = agv.comp.curr_pos() waiting_pos = agv.comp.waiting_pos() #2.1当前位置在等待位置（） if curr_pos == waiting_pos: continue need_move = False #2.2当前位置在料仓位置，去等待位置（） if self.store_ctr.is_in_place(curr_pos): need_move = True #2.3当前位置在机床位置且机床上没有物料，去等待位置（） if not need_move: cnc = self.cnc_ctr.get_cnc_by_pos(curr_pos) if cnc is not None and cnc.comp.is_idle(): need_move = True if need_move: distance = self.path_ctr.distance(curr_pos, waiting_pos) if is_run is False: return (distance, None) exe = self.new_task(AGVIdleCNCExecution) if exe.execute(product, agv, waiting_pos): return (distance, exe) else: del exe return None ### return None

from base.task_condition_base import TaskConditionBase from tasks.processing_execution import ProcessingExecution from entities.product_entity import ProductEntity class ProcessingCondition(TaskConditionBase): def __init__(self): super().__init__() self.order = 3 self.name = "3.机床加工" ### def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, ProcessingExecution | None] | None: #1.根据产品ID获取机床 （） cnc = self.cnc_ctr.get_cnc_by_product(product.id) if cnc is None: return None ### #2.不执行，直接返回距离 if is_run is False: return (0, None) #3.执行，则生成任务对象，运行执行方法 exe = self.new_task(ProcessingExecution) if exe.execute(product, cnc): return (0, exe) else: del exe return None

from base.task_condition_base import TaskConditionBase from tasks.put_to_cnc_execution import PutToCNCExecution from entities.product_entity import ProductEntity import sys class PutToCNCCondition(TaskConditionBase): def __init__(self): super().__init__() self.order = 2 self.name = "2.机床上料" def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, PutToCNCExecution | None] | None: # 1.获取产品所在AGV （去掉多余的解包符号，直接接收对象） agv = self.agv_ctr.get_agv_by_product(product.id) # 加上双保险判断，确保拿到的是有效的 agv 对象 if agv is None or getattr(agv, 'comp', None) is None or agv.comp.is_idle() is False: return None # 2.根据产品工艺ID获取可加工机床列表 （） process_code = product.comp.get_process_code() cncs = self.cnc_ctr.get_cncs_by_process_code(process_code) if len(cncs) == 0: return None # 3.获取最优的机床 （） cnc = cncs[0] distance = self.path_ctr.distance(agv.comp.curr_pos(), cnc.data.pos_id) # 4.不执行，直接返回距离 if is_run is False: return (distance, None) # 5.执行，则生成任务对象，运行执行方法 exe = self.new_task(PutToCNCExecution) if exe.execute(product, agv, cnc): return (distance, exe) else: del exe return None

from base.task_condition_base import TaskConditionBase from tasks.put_to_store_execution import PutToStoreExecution from entities.product_entity import ProductEntity class PutToStoreCondition(TaskConditionBase): def __init__(self): super().__init__() self.order = 5 self.name = "5.料仓放料" ### def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, PutToStoreExecution | None] | None: #1.获取产品所在AGV（） agv = self.agv_ctr.get_agv_by_product(product.id) # 加上我们引以为傲的双保险，防止空壳对象导致后续报错 if agv is None or getattr(agv, 'comp', None) is None or agv.comp.is_idle() is False: return None #2.获取放料仓位信息（） store = self.store_ctr.store if store is None: return None store_pallet_id = product.data.out_pallet_id #3.计算AGV到目标位置的距离（） goal_pos_id = store.comp.get_real_pos(store_pallet_id) distance = self.path_ctr.distance(agv.comp.curr_pos(), goal_pos_id) ### #4.不执行，直接返回距离 if is_run is False: return (distance, None) #5.执行，则生成任务对象，运行执行方法 exe = self.new_task(PutToStoreExecution) if exe.execute(product, store, agv, product.data.out_pallet_id): return (distance, exe) else: del exe return None

from base.task_condition_base import TaskConditionBase from tasks.take_from_cnc_execution import TakeFromCNCExecution from entities.product_entity import ProductEntity class TakeFromCNCCondition(TaskConditionBase): def __init__(self): super().__init__() self.order = 4 self.name = "4.机床下料" ### def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, TakeFromCNCExecution | None] | None: #1.获取产品所在机床（） cnc = self.cnc_ctr.get_cnc_by_product(product.id) if cnc is None: return None #2.获取产品所在AGV（） agv = self.agv_ctr.get_agv_by_product(product.id) if agv is None or agv.comp.is_idle() is False: return None #3.计算AGV到机床的距离（） distance = self.path_ctr.distance(agv.comp.curr_pos(), cnc.data.pos_id) ### #4.不执行，直接返回距离 if is_run is False: return (distance, None) #5.执行，则生成任务对象，运行执行方法 exe = self.new_task(TakeFromCNCExecution) if exe.execute(product, agv, cnc): return (distance, exe) else: del exe return None

from base.task_condition_base import TaskConditionBase from tasks.take_from_store_execution import TakeFromStoreExecution from entities.product_entity import ProductEntity import sys class TakeFromStoreCondition(TaskConditionBase): def __init__(self): super().__init__() self.order = 1 self.name = "1.料仓取料" ### def check_run(self, product: ProductEntity, is_run: bool) -> tuple[float, TakeFromStoreExecution | None] | None: # 1.获取仓位列表（） mat_id = product.data.mat_id pallets = self.store_ctr.get_pallets_by_mat(mat_id) if len(pallets) == 0: return None # 2.获取最优的仓位号以及AGV（） best_distance = sys.maxsize best_store = None best_agv = None best_store_pallet_id = None best_agv_pallet_id = None for store_pallet_id in pallets: store, _ = self.store_ctr.get_store_by_pallet(store_pallet_id) if store is None: continue pos_id = store.comp.get_real_pos(store_pallet_id) # --- 终极解包方案开始 --- agv_raw = self.agv_ctr.get_optimal_agv(pos_id, mat_id) agv = agv_raw # 如果是元组，自动遍历寻找包含 comp 属性的对象 if isinstance(agv_raw, tuple): for item in agv_raw: if hasattr(item, 'comp'): agv = item break # 如果有更深一层的 (key, 对象) 嵌套 elif isinstance(item, tuple) and len(item) > 1 and hasattr(item[1], 'comp'): agv = item[1] break # 加上 getattr 双保险，彻底杜绝 AttributeError if agv is None or getattr(agv, 'comp', None) is None or agv.comp.is_idle() is False: continue # --- 终极解包方案结束 --- agv_pallet_id = agv.comp.get_null_pallet(mat_id) if agv_pallet_id is None: continue distance = self.path_ctr.distance(agv.comp.curr_pos(), pos_id) if distance < best_distance: best_distance = distance best_store = store best_agv = agv best_store_pallet_id = store_pallet_id best_agv_pallet_id = agv_pallet_id if best_store is None or best_agv is None: return None ### # 3.不执行，直接返回距离（） if is_run is False: return (best_distance, None) # 4.执行，则生成任务对象，运行执行方法 exe = self.new_task(TakeFromStoreExecution) if exe.execute(product, best_store, best_agv, best_store_pallet_id, best_agv_pallet_id): return (best_distance, exe) else: del exe return None