| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- import time, threading
- from utils.s7data import S7data
- from utils.logger import Logger
- from models.billet_trace_pusher import Trace_pusher
- from models.parking import Parking
- from models.data_sender import Sender
- class Crane:
- def __init__(self, data_s7: S7data, tracer_5: Trace_pusher, tracer_6: Trace_pusher, parking: Parking, sender: Sender, logger: Logger):
- self.data = data_s7
- self.tracer_5 = tracer_5
- self.tracer_6 = tracer_6
- self.parking = parking
- self.sender = sender
- self.logger = logger
- self.cranes = {
- "A1": self.data.make_point("天车A1位置"),
- "A2": self.data.make_point("天车A2位置"),
- "A3": self.data.make_point("天车A3位置")
- }
- self.billetsNo = {
- "A1": "",
- "A2": "",
- "A3": ""
- }
- self.areas = {
- "6#coolbed1": [35000, 38500],
- "6#coolbed2": [22000, 24000],
- "car3": [2000, 4000],
- "highline": [60300, 62300]
- }
- self.threads_poor = [
- threading.Thread(target=self.area_producer, args=("6#coolbed1",)),
- threading.Thread(target=self.area_producer, args=("6#coolbed2",)),
- threading.Thread(target=self.area_consumer, args=("car3",)),
- threading.Thread(target=self.area_consumer, args=("highline",))
- ]
- for i in self.threads_poor:
- i.start()
- def scan_crane(self, areaNo):
- start = self.areas[areaNo][0]
- end = self.areas[areaNo][1]
- while True:
- for key, value in self.cranes.items():
- if start <= value.data <= end:
- return key
- time.sleep(0.5)
- def crane_leave(self, craneNo, areaNo):
- start = self.areas[areaNo][0]
- end = self.areas[areaNo][1]
- while True:
- if self.cranes[craneNo].data < start or self.cranes[craneNo].data > end:
- return True
- time.sleep(0.5)
-
- def area_producer(self, areaNo):
- while True:
- craneNo = self.scan_crane(areaNo)
- self.logger.debug(f"天车{craneNo}进入区域{areaNo}")
- if self.billetsNo[craneNo] == "":
- self.logger.debug(f"天车{craneNo}正在等待或尝试夹起钢坯")
- self.get_billet(craneNo, areaNo)
- self.logger.debug(f"天车{craneNo}获取到钢坯")
- else:
- self.logger.debug(f"天车{craneNo}有未卸下的钢坯")
- self.crane_leave(craneNo, areaNo)
- self.logger.debug(f"天车{craneNo}离开区域{areaNo}")
- time.sleep(1)
- def area_consumer(self, areaNo):
- while True:
- craneNo = self.scan_crane(areaNo)
- self.logger.debug(f"天车{craneNo}进入区域{areaNo}")
- if self.billetsNo[craneNo] == "":
- self.logger.debug(f"天车{craneNo}未携带钢坯")
- else:
- self.logger.debug(f"天车{craneNo}正在等待或尝试卸下钢坯")
- self.unload_billet(craneNo, areaNo)
- self.crane_leave(craneNo, areaNo)
- self.logger.debug(f"天车{craneNo}离开区域{areaNo}")
- time.sleep(1)
- def get_billet(self, craneNo, areaNo):
- # 阻塞获取
- tmp = []
- while not bool(tmp):
- if areaNo == "6#coolbed1":
- tmp = self.tracer_6.get_billet("left")
- elif areaNo == "6#coolbed2":
- tmp = self.tracer_6.get_billet("right")
-
- self.billetsNo[craneNo] = ','.join(tmp)
- return tmp
- def unload_billet(self, craneNo, areaNo):
- # 直接放下
- tmp = self.billetsNo[craneNo]
- self.billetsNo[craneNo] = ""
- if areaNo == "car3":
- self.logger.info(f"有一夹子钢坯放置进了车位3")
- self.sender.car_save(self.parking.ccmNo_list[2], tmp, self.parking.current_car[2], craneNo, time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()))
- elif areaNo == "highline":
- self.logger.info(f"有一夹子钢坯放置进了高线辊道")
- self.sender.host_send("6", tmp, "高线", craneNo)
|
