billetManager/models/batcher.py

from utils.statepoint import *
from utils.mqttdata import *
from utils.s7data import *
from models.data_sender import *
from models.mysql_data import MysqlData, MysqlDataSizing
import logging, datetime, queue, requests

def _get_heat_init_strand(heat_no, retry = 3):
    url = f"http://192.168.0.119:7005/actualControl/billetActual/heatNoStats?heatNo={heat_no}"
    while retry:
        try:
            response = requests.get(url)
            result = response.json()
            if result["success"]:
                return result["result"]["strandTotal"]
        except Exception as e:
            warnings.warn(str(e))
        finally:
            retry -= 1
    return [0] * 8

class Heat:
    def __init__(self, ccm_no, heat_no, start_pour_time: datetime.datetime = None, init_strands_status: list[int] = [0] * 8):
        self.ccm_no = str(ccm_no)
        self.heat_no = str(heat_no)
        self.start_pour_time = start_pour_time

        self.using_event = threading.Event()

        self.strands_req = init_strands_status
        self.total = sum(init_strands_status)
        self.billet_union_count = {}
        self.resource_lock = threading.Lock()

    @classmethod
    def from_dict(cls, src_dict: dict):
        return cls(
            src_dict["ccmNo"],
            src_dict["heatNo"],
            src_dict["sendTime"] if "sendTime" in src_dict and src_dict["sendTime"] else None
        )
    
    def to_dict(self):
        return {
            'ccmNo': self.ccm_no,
            'heatNo': self.heat_no,
            'sendTime': self.start_pour_time.strftime("%Y-%m-%d %H:%M:%S")
        }
    
    def begin_use(self):
        self.using_event.set()

    def is_poured(self) -> bool:
        if self.start_pour_time:
            return True
        return False
    
    def is_using(self) -> bool:
        return self.using_event.is_set()
    
    def init_strands_status(self, init_status: list[int]):
        if self.is_using():
            raise ValueError("无法对已开始出坯的炉进行流初始化")
        with self.resource_lock:
            self.strands_req = init_status
            self.total = sum(init_status)
    
    def claim_strand_seq(self, strand_no) -> tuple[int]:
        if not self.is_using():
            return 0
        index = strand_no - 1
        with self.resource_lock:
            self.total += 1
            self.strands_req[index] += 1
            return self.strands_req[index], self.total
        
    def claim_union_seq(self, length: int) -> int:
        if not self.is_using():
            return 0
        with self.resource_lock:
            if length not in self.billet_union_count:
                self.billet_union_count[length] = 0
            self.billet_union_count[length] += 1
            return self.billet_union_count[length]

class Billet:
    def __init__(self, cutting_time: datetime.datetime = None):
        if cutting_time:
            self.start_cutting_time: datetime.datetime = cutting_time
        else:
            self.start_cutting_time: datetime.datetime = datetime.datetime.now()

        self.heat: Heat = None
        self.stop_cutting_time: datetime.datetime = None
        self.strand_no: int = None
        self._strand_seq: int = None
        self._heat_seq: int = None
        self.sizing = None
        self.drawing_speed = None

    @property
    def billet_no(self):
        if self.heat and self.strand_no and self.strand_seq:
            return self.heat_no + self.ccm_no + str(self.strand_no) + f"{self.strand_seq:02d}"

    @property
    def ccm_no(self):
        if self.heat:
            return self.heat.ccm_no

    @property
    def heat_no(self):
        if self.heat:
            return self.heat.heat_no

    @property
    def strand_seq(self):
        return self._strand_seq
    
    @property
    def heat_seq(self):
        return self._heat_seq
    
    def batch_strand(self, strand_no = None) -> bool:
        if strand_no:
            self.strand_no = strand_no
        if not (self.strand_no and self.heat):
            return False
        res = self.heat.claim_strand_seq(self.strand_no)
        if res:
            self._strand_seq, self._heat_seq = res
            return True
        
        return False

class BatchHeat:
    def __init__(self, ccm_no, manual_change_heat_sig: Statepoint,
                 mysql_heat_point: Statepoint, mqtt_heat_point: Statepoint,
                 ladle_weight_point_1: Statepoint, ladle_weight_point_2: Statepoint,
                 sender: Sender, logger: logging.Logger, queue_max = 5):
        self.ccm_no = ccm_no
        self.manual_change_heat_sig = manual_change_heat_sig
        self.mysql_heat_point = mysql_heat_point
        self.mqtt_heat_point = mqtt_heat_point
        self.ladle_weight_point_1 = ladle_weight_point_1
        self.ladle_weight_point_2 = ladle_weight_point_2
        self.sender = sender
        self.logger = logger

        self.last_enqueue = ()
        self.heats_queue = queue.PriorityQueue(queue_max)
        self.heat_info_shared = {}
        self.heat_info_lock = threading.Lock()

        self.current_heat: Heat = None
        self._thread: threading.Thread = None
        self._thread_event = threading.Event()
        self.change_heat_lock = threading.Lock()

        self.manual_change_heat_sig.set_excite_action(self.manual_change_heat_action)
        self.mysql_heat_point.set_excite_action(self.heat_arrive_in_action)
        self.mqtt_heat_point.set_excite_action(self.heat_start_pour_action)

        self.current_heat_point = Statepoint(0)

    def heat_arrive_in_action(self):
        # 维护内部状态
        self.mysql_heat_point.set_state(False)

        # 获取炉次到站信息
        heat_no, into_time = self.mysql_heat_point.data
        time_for_check = into_time + datetime.timedelta(minutes=12)
        self.heat_info_shared[heat_no] = {}

        self.logger.debug(f"[Counter]检测到炉次进站{heat_no}")

        # 大包重量选择算法
        ladle_weight = max(self.ladle_weight_point_1.data, self.ladle_weight_point_2.data)

        # 等待诺德开浇信号判断
        while self.heat_info_shared[heat_no] == {}:
            # 如果超时，根据入站时间预测开浇时间
            if datetime.datetime.now() > time_for_check:
                with self.heat_info_lock:
                    if "heatNo" not in self.heat_info_shared[heat_no]:
                        self.heat_info_shared[heat_no]["ccmNo"] = self.ccm_no
                        self.heat_info_shared[heat_no]["heatNo"] = heat_no
                        self.heat_info_shared[heat_no]["sendTime"] = into_time + datetime.timedelta(minutes=5)
                break
            time.sleep(2)

        # 标志是否为铸机开机的第一次开浇
        first_heat_flag = self.current_heat == None and self.heats_queue.empty()

        # 写入日志
        if first_heat_flag:
            self.logger.info(f'[Counter]首次开浇:{heat_no}')
        else:
            self.logger.info(f'[Counter]炉次开浇:{heat_no}')

        # 维护浇铸队列
        try:
            self.last_enqueue = (self.heat_info_shared[heat_no]["sendTime"], Heat.from_dict(self.heat_info_shared[heat_no]))
            self.heats_queue.put(self.last_enqueue, True, 3)
        except TimeoutError as e:
            self.logger.error(f"[Counter]炉次队列已满({self.heats_queue.maxsize})，存在异常:{e}")
            return

        # 使用sender向外发送信号
        self.sender.begin_pour(self.heat_info_shared[heat_no], ladle_weight)
        self.heat_info_shared.pop(heat_no)

    def heat_start_pour_action(self):
        heat_info = self.mqtt_heat_point.data
        heat_no = str(heat_info['heatNo'])
        if "sendTime" in heat_info:
            heat_info["sendTime"] = datetime.datetime.strptime(heat_info["sendTime"], "%Y-%m-%d %H:%M:%S")

        if heat_no in self.heat_info_shared:
            with self.heat_info_lock:
                self.heat_info_shared[heat_no] = heat_info
            self.logger.debug(f'[Counter]收到诺德开浇信号:{heat_no}')
        elif self.last_enqueue[-1].heat_no == heat_no:
            self.logger.warning(f"[Counter]诺德开浇信号晚到{heat_no}")
        else:
            self.logger.info(f'[Counter]开浇信号表示倒浇操作:{heat_no}')
            # 维护浇铸队列
            try:
                self.last_enqueue = (heat_info["sendTime"], Heat.from_dict(heat_info))
                self.heats_queue.put(self.last_enqueue, True, 3)
            except TimeoutError as e:
                self.logger.error(f"[Counter]炉次队列已满({self.heats_queue.maxsize})，无法插入{heat_no}，异常:{e}")
                return
            
    def manual_change_heat_action(self):
        with self.change_heat_lock:
            heat_info = self.manual_change_heat_sig.data
            if self.current_heat and heat_info['heatNo'] == self.current_heat.heat_no:
                return
            
            if "startPourTime" in heat_info:
                heat_info["sendTime"] = datetime.datetime.strptime(heat_info["startPourTime"], "%Y-%m-%d %H:%M:%S")

            self.logger.info(f"手动换炉触发:{self.current_heat.heat_no if self.current_heat else "未知"}->{heat_info['heatNo']}")

            self.sender.begin_pour(heat_info, max(self.ladle_weight_point_1.data, self.ladle_weight_point_2.data))

            self.current_heat = Heat.from_dict(heat_info)
            self.current_heat.init_strands_status(_get_heat_init_strand(self.current_heat.heat_no))
            self.current_heat.begin_use()
            self.current_heat_point.inject(int(self.current_heat.heat_no))

    def loop_forever(self):
        try:
            while self._thread_event.is_set():
                if not self.heats_queue.empty() and self.heats_queue.queue[0][0] + datetime.timedelta(minutes=10) <= datetime.datetime.now():
                    with self.change_heat_lock:
                        _, heat = self.heats_queue.get()
                        if self.current_heat:
                            self.logger.debug(f"[Counter]炉次划分{self.current_heat.heat_no}->{heat.heat_no}")
                        else:
                            self.logger.debug(f"[Counter]连铸机出坯启动，炉次为{heat.heat_no}")
                        self.current_heat = heat
                        self.current_heat.init_strands_status(_get_heat_init_strand(self.current_heat.heat_no))
                        self.current_heat.begin_use()
                        self.current_heat_point.inject(int(self.current_heat.heat_no))
        except Exception as e:
            self._thread_event.clear()
            self.logger.error(f"换炉线程意外退出:{e}")
    
    def loop_start(self):
        if self._thread and self._thread.is_alive():
            if self._thread_event.is_set():
                return
            raise ChildProcessError("换炉线程异常持续运行")
        self._thread = threading.Thread(target=self.loop_forever)
        self._thread_event.set()
        self._thread.start()

    def loop_stop(self):
        self._thread_event.clear()
        if self._thread and self._thread.is_alive():
            self._thread.join(3)

    def get_current_heat_info(self) -> dict:
        # 当换炉进行中时，获取炉次信息阻塞等待
        with self.change_heat_lock:
            return self.current_heat
        
    def __del__(self):
        self.loop_stop()
        
class BatchBillet:
    def __init__(self, heat: BatchHeat,
                 cutting_sig_point_list: list[Statepoint], short_sizing_point_list: list[Statepoint],
                 sizing_point_list: list[Statepoint], drawing_speed_point_list: list[Statepoint],
                 logger: logging.Logger):
        self.heat = heat
        self.cutting_sig_point_list = cutting_sig_point_list
        self.short_sizing_point_list = short_sizing_point_list
        self.sizing_point_list = sizing_point_list
        self.drawing_speed_point_list = drawing_speed_point_list
        self.logger = logger

        self.output_billet_queues = [queue.Queue(2) for _ in range(8)]
        self.last_enqueue: list[Billet] = [None for _ in range(8)]

        self._backfill_on_off = threading.Event()
        self.dspeed_integrate_point_list = None

        for i in range(8):
            self.cutting_sig_point_list[i].set_excite_action(lambda index=i: self.cutting_action(index))

    def cutting_action(self, index: int):
        # 先获取实时信息
        sizing = self.sizing_point_list[index].data
        drawing_speed = self.drawing_speed_point_list[index].data

        # 维护拉速回填机制，顺便利用拉速判断短尺
        if self._backfill_on_off.is_set():
            predict_length = self.dspeed_integrate_point_list[index].data
            self.dspeed_integrate_point_list[index].data = 0
            predict_length *= 1000
            if predict_length < 3000:
                predict_length += sizing
            self.logger.debug(f"[Counter]{index+1}流钢坯预测长度为{predict_length}")
            if abs(self.short_sizing_point_list[index].data - predict_length) < abs(sizing - predict_length):
                sizing = self.short_sizing_point_list[index].data

        self.logger.info(f"[Counter]{index+1}流钢坯开始切割")

        billet = Billet()
        billet.sizing = sizing
        billet.drawing_speed = drawing_speed
        billet.strand_no = index + 1
        billet.heat = self.heat.get_current_heat_info()
        if billet.heat:
            self.trans_billet(billet)
        else:
            self.logger.warning(f"[Counter]{index+1}流钢坯未获取到炉次信息")

    def trans_billet(self, billet: Billet):
        try:
            self.output_billet_queues[billet.strand_no-1].put(billet, True, 10)
            self.last_enqueue[billet.strand_no-1] = billet
        except Exception as e:
            self.logger.error(f"{billet.strand_no}流队列加入钢坯失败:{e}")

    def set_dspeed_backfill(self, dspeed_integrate_point_list: list[Statepoint]):
        if self.dspeed_integrate_point_list:
            raise ValueError("拉速补偿数据点不可重复设置")
        self.dspeed_integrate_point_list = dspeed_integrate_point_list.copy()
        for i in range(8):
            self.dspeed_integrate_point_list[i].set_convertor(lambda data, index=i: self.dspeed_convertor(data, index))
            self.dspeed_integrate_point_list[i].data = 0
            self.dspeed_integrate_point_list[i].set_excite_action(lambda index=i: self.dspeed_backfill_action(index))

    def open_dspeed_backfill(self):
        if not self.drawing_speed_point_list:
            raise ValueError("未设置拉速补偿数据点")
        if self._backfill_on_off.is_set():
            return
        self._backfill_on_off.set()
        for i in self.drawing_speed_point_list:
            i.data = 0
            i.set_state(0)
            i.allow_update()

    def close_dspeed_backfill(self):
        self._backfill_on_off.clear()
        if self.drawing_speed_point_list:
            for i in self.drawing_speed_point_list:
                i.allow_update(0)

    def dspeed_convertor(self, data, index):
        return data * 1000 >= self.sizing_point_list[index].data

    def dspeed_backfill_action(self, index: int):
        # 维护内部状态
        self.dspeed_integrate_point_list[index].data = 0
        self.dspeed_integrate_point_list[index].set_state(False)
        
        in_time = datetime.datetime.now()
        while datetime.datetime.now() - in_time < datetime.timedelta(seconds=20):
            if self.last_enqueue[index] and datetime.datetime.now() - self.last_enqueue[index].start_cutting_time < datetime.timedelta(seconds=90):
                return
            time.sleep(0.5)
        
        self.logger.info(f"[Counter]{index+1}流拉速判据回填钢坯")
        billet = Billet(in_time)
        billet.sizing = self.sizing_point_list[index].data
        billet.drawing_speed = self.drawing_speed_point_list[index].data
        billet.strand_no = index + 1
        # 炉次信息可能因换炉阻塞，上面先获取实时信息
        billet.heat = self.heat.get_current_heat_info()
        if billet.heat:
            self.trans_billet(billet)
        else:
            self.logger.warning(f"[Counter]{index+1}流钢坯未获取到炉次信息")


class Batcher:
    @property
    def data_mqtt(self):
        if self._data_mqtt.cli == None:
            raise ValueError('The MQTT connection to MES has not been initialized.')
        return self._data_mqtt
    
    @property
    def data_s7(self):
        if self._data_s7 == None:
            raise ValueError('The S7 connection to the casting machine PLC has not been initialized.')
        return self._data_s7
    
    @property
    def sender(self):
        if self._sender == None:
            raise ValueError('The sender has not been set.')
        return self._sender

    def __init__(self, data_mqtt: Mqttdata, data_s7: S7data, data_web: Mqttdata, ccmNo, logger: logging.Logger, sender: Sender, data_nuo: MysqlData, data_qbc: MysqlDataSizing):
        self._data_mqtt = data_mqtt
        self._data_s7 = data_s7
        self.data_web = data_web
        self._sender = sender
        self.logger = logger
        self.ccm_no = ccmNo
        self.data_nuo = data_nuo
        self.data_qbc = data_qbc

        self.logger.info(f"[Counter]分炉分坯模块:{ccmNo}号机模块启动")

        self.ladle_weight_1 = self.data_s7.make_point('大包重量1')
        self.ladle_weight_2 = self.data_s7.make_point('大包重量2')

        self.begin_pour_sig = self.data_mqtt.make_point(f'{ccmNo}#开浇信号')
        self.end_pour_sig = self.data_mqtt.make_point(f'{ccmNo}#停浇信号')
        self.nuo_heat_point = self.data_nuo.make_point(f"{ccmNo}#到站信息")
        self.manual_change_heat_sig = data_web.make_point(f"{ccmNo}#手动换炉")

        self.begin_cutting_sig = [self.data_s7.make_point(f'L{i+1}切割信号[0]') for i in range(8)]
        self.end_cutting_sig = [self.data_s7.make_point(f'L{i+1}切割信号[1]') for i in range(8)]
        if ccmNo == '5':
            self.length_cutting = [self.data_qbc.make_point(f'{i+1}流定尺') for i in range(8)]
        elif ccmNo == '6':
            self.length_cutting = [self.data_s7.make_point(f'L{i+1}定尺') for i in range(8)]
        self.short_length_cutting = [self.data_s7.make_point(f'L{i+1}短尺') for i in range(8)]
        self.drawing_speed = [self.data_s7.make_point(f'L{i+1}拉速') for i in range(8)]

        # 拉速补偿点位
        self.dspeed_backfill_point = [self.data_s7.make_point(f"L{i+1}拉速", Integration_speed_mpmin) for i in range(8)]
        
        # 主逻辑对象
        self.batch_heat = BatchHeat(self.ccm_no, self.manual_change_heat_sig,
                                    self.nuo_heat_point, self.begin_pour_sig,
                                    self.ladle_weight_1, self.ladle_weight_2,
                                    self.sender, self.logger)
        self.batch_heat.loop_start()
        
        self.batch_billet = BatchBillet(self.batch_heat, self.begin_cutting_sig, self.short_length_cutting,
                                        self.length_cutting, self.drawing_speed, self.logger)
        self.batch_billet.set_dspeed_backfill(self.dspeed_backfill_point)
        self.batch_billet.open_dspeed_backfill()