BarLine1Sizing/models/steel_fit.py

from utils.statepoint import Statepoint
from models.data_sender import Sender
import numpy as np
from scipy import interpolate, integrate
from collections import deque
from models.cip_data import CIPData
from utils.s7data import S7data
import datetime, logging, time, threading, queue

def interp1d(x, y, logger: logging.Logger):
    if x[-1] < datetime.datetime.now().timestamp() - 300:
        logger.error("连续5分钟未采集到变化数值，表示为-1")
        return lambda data: -1
    if len(x) < 4:
        logger.warning("采集到的数量低于4个，表示为平均值")
        avg = np.mean(y)
        return lambda data: avg
    try:
        return interpolate.interp1d(x, y, kind='cubic')
    except Exception as e:
        logger.error(f"三次样条插值失败:{e}")
        return lambda data: -1

class BufferPoint(Statepoint):
    def __init__(self, initvalue = None, initstate = False, maxlen: int | None = 3000):
        super().__init__(deque(maxlen = maxlen), initstate)

    def inject(self, data):
        self.data.append((data, datetime.datetime.now().timestamp()))

    def get_buffer(self):
        res = self.data.copy()
        last = res[-1][0]
        res.append((last, datetime.datetime.now().timestamp() + 0.001))
        return res

class billet_data_gatherer:
    def __init__(self, dspeed_point: BufferPoint, cutting_sig_point: Statepoint, sizing_point: Statepoint, flow_rate_point_list: list[BufferPoint],
                 logger: logging.Logger, strand_no: int, result_queue: queue.Queue):
        self.dspeed_point = dspeed_point
        self.cutting_sig_point = cutting_sig_point
        self.sizing_point = sizing_point
        self.flow_rate_point_list = flow_rate_point_list
        self.logger = logger
        self.strand_no = strand_no
        self.result_queue = result_queue

        self.MOLD_TO_CUTTER_DISTANCE = 28
        self.CRITICAL_ZONE_LENGTH = 12

        self.cutting_sig_point.set_excite_action(self.cutting_action)
        
    def cutting_action(self):
        cutting_time = datetime.datetime.now().timestamp()
        sizing = self.sizing_point.data / 1000

        self.logger.debug(f"{self.strand_no}流开始切割")

        time.sleep(30)
        dspeed_buffer: deque = self.dspeed_point.get_buffer()
        flow_rate_buffer_list: list[deque] = [self.flow_rate_point_list[i].get_buffer() for i in range(5)]

        if len(dspeed_buffer) < 10:
            self.logger.debug(f"{self.strand_no}流已统计数据量不足，无法计算")
            return

        data_tuple, time_tuple = zip(*dspeed_buffer)

        x = np.array(time_tuple)
        y = np.array(data_tuple) / 60
        vt_func = interp1d(x, y, self.logger)
        entry_time = self._binary_search_start(vt_func, cutting_time, sizing + self.MOLD_TO_CUTTER_DISTANCE)

        if entry_time == None:
            self.logger.debug(f"{self.strand_no}流已统计数据量不足，无法计算")
            return
        
        exit_time = self._binary_search_end(vt_func, entry_time, sizing + self.CRITICAL_ZONE_LENGTH)
        dspeed_avg = (sizing + self.CRITICAL_ZONE_LENGTH) / (exit_time - entry_time) * 60

        self.create_data(cutting_time, entry_time, exit_time, self.flow_rate_total(flow_rate_buffer_list, entry_time, exit_time), dspeed_avg)
        
    def _binary_search_start(self, func, upper_limit, target):
        left = func.x.min()
        right = func.x.max()

        if self._get_distance(func, left, upper_limit) < target:
            return

        while abs(right - left) > 0.01:
            mid = (left + right) / 2
            if self._get_distance(func, mid, upper_limit) >= target:
                left = mid
            else:
                right = mid
        
        return (left + right) / 2
    
    def _binary_search_end(self, func, lower_limit, target):
        left = func.x.min()
        right = func.x.max()

        while abs(right - left) > 0.01:
            mid = (left + right) / 2
            if self._get_distance(func, lower_limit, mid) >= target:
                right = mid
            else:
                left = mid
        
        return (left + right) / 2

    def _get_distance(self, vt_func, lower, upper):
        return integrate.quad(vt_func, lower, upper)[0]
    
    def flow_rate_total(self, deque_list: list[deque], start_time, end_time):
        res = []
        for dequei in deque_list:
            data_tuple, time_tuple = zip(*dequei)
            x = np.array(time_tuple)
            y = np.array(data_tuple) / 3600
            vt_func = interp1d(x, y, self.logger)
            total = integrate.quad(vt_func, start_time, end_time)[0]
            res.append(total)

        return sum(res)
    
    def create_data(self, cutting_time, entry_time, exit_time, water_total, dspeed_avg):
        self.logger.debug(f"{self.strand_no}流钢坯计算结果：")
        self.logger.debug(f'\t切割时间：{datetime.datetime.fromtimestamp(cutting_time).strftime("%Y-%m-%d %H:%M:%S")}')
        self.logger.debug(f'\t进入时间：{datetime.datetime.fromtimestamp(entry_time).strftime("%Y-%m-%d %H:%M:%S")}')
        self.logger.debug(f'\t离开时间：{datetime.datetime.fromtimestamp(exit_time).strftime("%Y-%m-%d %H:%M:%S")}')
        self.logger.debug(f'\t水量总计：{water_total}')
        self.logger.debug(f'\t平均拉速：{dspeed_avg}')
        self.result_queue.put((self.strand_no, float(cutting_time), float(entry_time), float(exit_time), float(water_total), float(dspeed_avg)))

class SteelFit:
    def __init__(self, s7_data_20: S7data, s7_data_215: S7data, cip_data: CIPData, sender: Sender, logger: logging.Logger):
        self.water_temperature_buffer: BufferPoint = cip_data.make_point("5#二冷水总管温度", BufferPoint)
        self.water_pressure_buffer: BufferPoint = cip_data.make_point("5#二冷水总管压力", BufferPoint)
        self.steel_temperature_buffer: BufferPoint = s7_data_20.make_point("中间包连续测温温度", BufferPoint)
        self.water_temperature_difference_buffer: BufferPoint = cip_data.make_point("5#结晶器水温差", BufferPoint)

        self.dspeed_buffer = [s7_data_20.make_point(f"{i}流结晶器拉速", BufferPoint) for i in range(1, 9)]
        self.cutting_sig_point = [s7_data_215.make_point(f"L{i}切割信号[0]") for i in range(1, 9)]
        self.sizing_point = [s7_data_20.make_point(f"{i}流定尺") for i in range(1, 9)]
        self.flow_rate_point_list = [[cip_data.make_point(f"5#水流量-{i}流-{j}段", BufferPoint) for j in range(1, 6)] for i in range(1, 9)]

        self.sender = sender
        self.logger = logger
        self.task_queue = queue.Queue()

        self.billet_data_gatherer_list = [
            billet_data_gatherer(
                self.dspeed_buffer[i],
                self.cutting_sig_point[i],
                self.sizing_point[i],
                self.flow_rate_point_list[i],
                logger,
                i + 1,
                self.task_queue
            )
            for i in range(8)
        ]
        
        self.thread_run = True
        self.thread = threading.Thread(target=self.loop_process)
        self.thread.start()

    def loop_process(self):
        while self.thread_run:
            try:
                task_tuple = self.task_queue.get(True, 1)
                tmp_dict = {}
                tmp_dict["strand_no"] = task_tuple[0]
                tmp_dict["cutting_time"] = datetime.datetime.fromtimestamp(task_tuple[1])
                tmp_dict["entry_time"] = datetime.datetime.fromtimestamp(task_tuple[2])
                tmp_dict["exit_time"] = datetime.datetime.fromtimestamp(task_tuple[3])
                
                cal_res = self.cal_data(task_tuple[2], task_tuple[3])
                tmp_dict["water_temperature"] = cal_res[0]
                tmp_dict["water_pressure"] = cal_res[1]
                tmp_dict["water_volume"] = task_tuple[4]
                tmp_dict["water_pressure_sd"] = cal_res[2]
                tmp_dict["steel_temperature"] = cal_res[3]
                tmp_dict["drawing_speed"] = task_tuple[5]
                tmp_dict["water_temperature_difference"] = cal_res[4]

                self.sender.upload_billet(tmp_dict)
            except queue.Empty:
                pass
            except Exception as e:
                self.logger.error(f"铸机数据计算过程中出现意外:{e}")

    def cal_data(self, entry_time, exit_time):
        wt = self.interval_avg(self.water_temperature_buffer.get_buffer(), entry_time, exit_time)
        wp = self.interval_avg(self.water_pressure_buffer.get_buffer(), entry_time, exit_time)
        wps = self.interval_sd(self.water_pressure_buffer.get_buffer(), entry_time, exit_time)
        st = self.interval_avg(self.steel_temperature_buffer.get_buffer(), entry_time, exit_time)
        wtd = self.interval_avg(self.water_temperature_difference_buffer.get_buffer(), entry_time, exit_time)

        return (float(wt), float(wp), float(wps), float(st), float(wtd))

    def interval_avg(self, buffer, left, right):
        data_tuple, time_tuple = zip(*buffer)
        x = np.array(time_tuple)
        y = np.array(data_tuple)
        func = interp1d(x, y, self.logger)
        inte = integrate.quad(func, left, right)[0]

        return inte / (right - left)

    def interval_sd(self, buffer, left, right):
        data_tuple, time_tuple = zip(*buffer)
        x = np.array(time_tuple)
        y = np.array(data_tuple)
        func = interp1d(x, y, self.logger)
        inte = integrate.quad(func, left, right)[0]
        avg = inte / (right - left)

        func2 = lambda x: (func(x) - avg) ** 2
        inte2 = integrate.quad(func2, left, right)[0]
        avg2 = inte2 / (right - left)

        return avg2 ** 0.5


if __name__ == "__main__":
    from utils.s7data import S7Client, S7data
    from utils.logger import Logger

    # 配置S7连接
    s7_1 = S7Client()
    s7_1.connect("172.16.1.20", 0, 0)
    data_1 = S7data("conf/s7@172.16.1.20.csv")
    data_1.set_S7Client(s7_1)
    data_1.auto_update_group()
    
    s7_2 = S7Client()
    s7_2.connect("172.16.1.21", 0, 0)
    data_2 = S7data("conf/s7@172.16.1.21.csv")
    data_2.set_S7Client(s7_2)
    data_2.auto_update_group()

    s7_3 = S7Client()
    s7_3.connect("192.168.1.215", 0, 0)
    data_3 = S7data("conf/s7@192.168.1.215.csv")
    data_3.set_S7Client(s7_3)
    data_3.auto_update_group()

    # 配置CIP连接
    cip_data = CIPData("192.168.3.100")
    cip_data.start_update()
    
    # 配置日志模块
    logger = Logger('test')
    logger.screen_on()

    class C:
        def upload_billet(self, arg_dict: dict):
            pass

    # 钢坯拟合模块
    steel_fit = SteelFit(data_1, data_3, cip_data, C(), logger)