shuziluansheng1

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(targetself.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: # 主线程持续运行等待中断信号如 CtrlC 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 Nonefrom 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 Nonefrom 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 Nonefrom 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 Nonefrom 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 Nonefrom 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