from dataclasses import dataclass from pathlib import Path from typing import List, Any, Callable from pubsub import pub import numpy as np from enum import Enum import struct import math import time from myconfig import TOKEN, DATA_DIR, FILE_MARK, OUTPUT_DIR, CAL_DIR from myconfig import PURE_WATER_FNAME, SAVE_EXT_NAME,FLOAT_RESERVE_BIT from myexception import MyException from mypath import MyDir class WorkMode(Enum): FILEMODE=1 DEVICEMODE=2 class AlgorithMode(Enum): A720=1 PureWater=2 class PureWaterData: wavelength=[360.0, 365.0, 370.0, 375.0, 380.0, 385.0, 390.0, 395.0, 400.0, 405.0, 410.0, 415.0, 420.0, 425.0, 430.0, 435.0, 440.0, 445.0, 450.0, 455.0, 460.0, 465.0, 470.0, 475.0, 480.0, 485.0, 490.0, 495.0, 500.0, 505.0, 510.0, 515.0, 520.0, 525.0, 530.0, 535.0, 540.0, 545.0, 550.0, 555.0, 560.0, 565.0, 570.0, 575.0, 580.0, 585.0, 590.0, 595.0, 600.0, 605.0, 610.0, 615.0, 620.0, 625.0, 630.0, 635.0, 640.0, 645.0, 650.0, 655.0, 660.0, 665.0, 670.0, 675.0, 680.0, 685.0, 690.0, 695.0, 700.0, 705.0, 710.0, 715.0, 720.0, 725.0] coeff=[0.0158, 0.0149, 0.0142, 0.0133, 0.0125, 0.0119, 0.0114, 0.0108, 0.0104, 0.0101, 0.0098, 0.0096, 0.00924, 0.00928, 0.00925, 0.0094, 0.01025, 0.01121, 0.01272, 0.01292, 0.01299, 0.01311, 0.0135, 0.0142, 0.0154, 0.0161, 0.0174, 0.0196, 0.0226, 0.0277, 0.0345, 0.0416, 0.0428, 0.0435, 0.0451, 0.0469, 0.049, 0.0526, 0.058, 0.061, 0.0632, 0.0655, 0.0708, 0.0784, 0.0908, 0.1111, 0.1362, 0.1682, 0.2234, 0.2587, 0.2653, 0.2687, 0.2764, 0.2842, 0.2924, 0.302, 0.3116, 0.3257, 0.3407, 0.3717, 0.4107, 0.4296, 0.4396, 0.4486, 0.4656, 0.4866, 0.5166, 0.5595, 0.6245, 0.7045, 0.8275, 1.0075, 1.2315, 1.4894] @dataclass class SerialPort : port: str = None baudrate: int = None bytesize: int = None parity: str =None stopbit: int =None def __post_init__(self): pass def set_serial_port_(self, sp:dict): self.port = sp['port'] self.baudrate = sp['baudrate'] self.bytesize = sp['bytesize'] self.parity = sp['parity'] self.stopbit = sp['stopbit'] pass @dataclass class Registers : slaveaddress: int = 1 functioncode: int = 3 DataBeginAddress: int = 2614 SNAddress: int = 2980 # 10 SNLen: int = 5 WLBeginAddress: int = 2102 count: int = 3 snBuf:bytes = b'' wavelengthBuf: bytes = b'' intensityBuf: bytes = b'' def __post_init__(self): pass def set_register(self, rg:dict): self.DataBeginAddress = rg['DataBeginAddress'] self.SNAddress = rg['SNAddress'] self.SNLen = rg['SNLen'] self.WLBeginAddress = rg['WLBeginAddress'] self.count = rg['count'] self.functioncode = rg['functioncode'] self.slaveaddress = rg['slaveaddress'] pass @dataclass class LogSetting : LogInterval: int =None RefreshInterval: int =None def set_log_setting(self, dct:dict): self.LogInterval = dct['LogInterval'] self.RefreshInterval = dct['RefreshInterval'] @dataclass class PlotSetting : LineBegin: int =None LineInterval: int =None def set_plot_setting(self, dct:dict): self.LineBegin = dct['LineBegin'] self.LineInterval = dct['LineInterval'] @dataclass class Retrieve : beginWL: int =None endWL: int=None interval: int=None def set_retrieve(self, dct:dict): self.beginWL = dct['beginWL'] self.endWL = dct['endWL'] self.interval = dct['interval'] @dataclass class Algorithm : A720: int =None PureWater: int =None def set_algorithm(self, dct:dict): self.A720 = dct['A720'] self.PureWater = dct['PureWater'] @dataclass class ConfigViper : SN: str = None lightPath: float = None mode:WorkMode = None filePath: Path = None deviceSN:str = None rawWavelength: list = None rawIntensity: list = None Wavelength: list = None Intensity: list = None # raw_wavelength_np = np.array([]) purewaterWavelength: list = None purewaterAttenuation: list = None purewaterAttAfterInterp: np.ndarray = None outputWavelength: list = None beginSite:int =None endsite:int = None measureTime:str =None retrieve: Retrieve = None algorithm: Algorithm = None serailPort: SerialPort = None register: Registers = None logSetting: LogSetting = None plotSetting: PlotSetting = None absorptionCoef:list =None def __post_init__(self): self.retrieve = Retrieve() self.algorithm = Algorithm() pass class MyViper(object): def __init__(self, sn: str = None): self.viper = ConfigViper(SN=sn) self.cfg= None self.file_lst =[] self.mydir = MyDir() self.ui_sn = "" self.devicesn_ok = False pass def set_SN(self, sn :str): self.viper.SN = sn self.__prepare_for_save() def set_mode(self, mode:WorkMode=WorkMode.FILEMODE): self.viper.mode = mode if self.viper.mode == WorkMode.FILEMODE: pass if self.viper.mode == WorkMode.DEVICEMODE: self.viper.serailPort =SerialPort() self.viper.serailPort.set_serial_port_( self.cfg['comsetting']) self.viper.register = Registers() self.viper.register.set_register( self.cfg['register']) pass def set_cfg_viper(self, cfg: dict): ''' cfg 从config.yaml 读出来的数据 ''' self.cfg = cfg if self.viper.SN != cfg['device']['UISN']: raise Exception(f" 波长 不匹配") self.viper.lightPath= cfg['device']['UIPath'] self.viper.algorithm.set_algorithm( cfg['algorithm'] ) self.viper.retrieve.set_retrieve(cfg['retrieve'] ) self.viper.logSetting =LogSetting() self.viper.logSetting.set_log_setting(cfg['logsetting'] ) self.viper.plotSetting =PlotSetting() self.viper.plotSetting.set_plot_setting(cfg['plotsetting'] ) # if self.viper.mode == WorkMode.FILEMODE: # pass # if self.viper.mode == WorkMode.DEVICEMODE: # self.viper.serailPort =SerialPort() # self.viper.serailPort.set_serial_port_( cfg['comsetting']) # self.viper.register =Registers() # self.viper.register.set_register( cfg['register']) # pass pass def set_retrieve(self, rtv:dict ): self.viper.retrieve.set_retrieve(rtv ) pass def set_raw_wavelength(self, raw_wavelength ): if self.viper.retrieve is None: return self.viper.rawWavelength = [float(i) for i in raw_wavelength] self.get_begin_end() pass def set_raw_intensity(self, sn, time_str, intensity): if self.viper.retrieve is None: return if sn != self.viper.SN : raise MyException(f" 数据的波长 [{sn}] 与系统波长 [{self.viper.SN}] 不匹配 ") # 取有效波长范围数据 ，因为浊度校正须在720nm左右，去掉720以后波长 intensity = intensity[self.viper.beginSite:self.viper.endsite] # 赋值，并转为浮点 self.viper.rawIntensity = [float(i) for i in intensity ] # 浊度校正 data = self.correction_turbidity( np.array(self.viper.rawIntensity) ) # 计算 吸收系数 data = data * (1000 * math.log(10,math.e) / self.viper.lightPath ) # 对纯水系数系数进行校正 data = self.correction_pure_water( data ) data = data.tolist() self.viper.measureTime = time_str self.viper.absorptionCoef = data # 分发数据 self.distribute_data(self.viper.measureTime,self.viper.absorptionCoef) pass def set_pure_water(self,wavelength, atten): self.viper.purewaterWavelength = [float(i) for i in wavelength] self.viper.purewaterAttenuation = [float(i) for i in atten] # print(f"== {self.viper.purewaterWavelength}") # print(f"== {self.viper.purewaterAttenuation}") if self.viper.rawWavelength is None: raise MyException(f" 没有波长数据。") self.interpo_pure_water() self.__prepare_for_save() def interpo_pure_water(self): self.viper.purewaterAttAfterInterp = np.interp( np.array(self.viper.outputWavelength), np.array(self.viper.purewaterWavelength), np.array(self.viper.purewaterAttenuation)) pass def get_begin_end(self,) -> list: # 读取配置文件 for i in range(len(self.viper.rawWavelength)): # print(f"i {i} {self.viper.rawWavelength[i]}") if self.viper.rawWavelength[i] < self.viper.retrieve.beginWL \ and self.viper.rawWavelength[i+1] > self.viper.retrieve.beginWL: self.viper.beginSite = i+1 pass if self.viper.rawWavelength[i] < self.viper.retrieve.endWL \ and self.viper.rawWavelength[i+1] > self.viper.retrieve.endWL: self.viper.endsite = i+2 break self.get_output_wavelength() msg = "起始波长 : " + \ str(self.viper.outputWavelength[0]) + \ " , 结束波长 : " + str(self.viper.outputWavelength[-1]) self.__set_msg("notice", msg) pub.sendMessage(self.msg) def get_output_wavelength(self,): self.viper.outputWavelength = self.viper.rawWavelength[self.viper.beginSite:self.viper.endsite] pass pass def distribute_data(self, time_str, data, mode =0): self.__set_msg( "data", {"time":time_str, "data":data } ) pub.sendMessage("update", msg=self.msg) # 保存数据 ？？？ self.mydir.setContent(self.viper.absorptionCoef,TOKEN,self.viper.measureTime) self.mydir.writeContent() if mode == 1: print(f" \ wavelenght : {self.viper.outputWavelength[0]} \ coef : {self.viper.absorptionCoef[0]} \ purewater : {self.viper.purewaterAttAfterInterp[0]} \ rawInt : {self.viper.rawIntensity[self.viper.beginSite]} \ ") def deal_measure_time_data(self, sn, res_time, res_data): ''' # res_time ['2011-01-28 00:00:32', '2011-01-28 00:01:04', '2011-01-28 00:04:05', '2011-01-28 00:04:17'] # res_data [ [[,,,]], [[,,,]], [[,,,]], [[,,,]]] 取 res_data[0] ''' # log.info(f" -> time : {res_time}",__name__, "deal_one_measure_time_data") # log.info(f" -> datalen : {len(res_data)} ",__name__, "deal_one_measure_time_data") if sn != self.viper.SN: raise MyException(f" wrong SN file {sn} !! [SN={self.viper.SN}]") for i in range(len(res_time)): self.set_raw_intensity( sn,res_time[i], res_data[i][0] ) self.__set_msg( "notice", "文件处理完毕" ) pub.sendMessage( "update", msg = self.msg ) def correction_turbidity(self, data:np.ndarray ): '''浊度校正, 吸光度 0 : 默认11项平均 1 : 720 2 : 不浊度校正 ''' # log.debug( "correction_turbidity .....",__name__, 'correction_turbidity' ) if self.viper.algorithm.A720 == 0: count = data.shape[0] tmp = 0.0 for i in range(count-11,count,1): tmp = tmp + data[i] tmp = tmp/11 # print(f"tmp ....{tmp}") return data - tmp pass if self.viper.algorithm.A720 == 1: count = data.shape[0] tmp = data[count] - (self.viper.outputWavelength[count]-720) * (data[count] -data[count-1]) \ / (self.viper.outputWavelength[count]-self.viper.outputWavelength[count-1]) # print(f"tmp .... {tmp}") return data-tmp pass if self.viper.algorithm.A720 == 2: return data return data pass def correction_pure_water(self, data:np.ndarray ): '''纯水校正 0 : 不变 1 : 加纯水衰减系数 ''' print(f" ==== {data} ") print(f" ==== {self.viper.purewaterAttAfterInterp} ") if self.viper.algorithm.PureWater == 0: return data pass if self.viper.algorithm.PureWater == 1: return data + self.viper.purewaterAttAfterInterp pass return data pass def __prepare_for_save(self,) -> bool: dir = Path().joinpath(DATA_DIR, OUTPUT_DIR) self.mydir.setBaseDir(dir) self.mydir.newDirIfNot() self.mydir.newFileIfNot(self.viper.SN+SAVE_EXT_NAME) self.mydir.setHeader(self.viper.outputWavelength, TOKEN, self.viper.SN) if self.mydir.checkHeader() == 0: self.mydir.writeHeader() if self.mydir.checkHeader() == -1: # self.popDialog(" 文件头不一致， 请备份到其他目录，并在该目录下删除") raise MyException(" 文件头不一致， 请备份到其他目录，并在该目录下删除") def __set_msg(self, typ, d): self.msg = {} self.msg.update( {"type":typ} ) self.msg.update( {"data":d} ) pass def get_device_sn_from_buf(self ): print( f" 000000 {self.viper.register.snBuf}") self.viper.deviceSN = self.viper.register.snBuf[-6:-2].decode() if self.viper.SN == self.viper.deviceSN: self.devicesn_ok = True def get_raw_wavelength_from_buf(self ): rawWavelength = self.convert_buf_2_float(self.viper.register.wavelengthBuf,FLOAT_RESERVE_BIT ) self.set_raw_wavelength( rawWavelength) print(f" === {len(self.viper.rawWavelength)} {self.viper.rawWavelength}" ) def get_raw_intensity_from_buf(self ): intens =self.convert_buf_2_float(self.viper.register.intensityBuf,FLOAT_RESERVE_BIT ) time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) self.set_raw_intensity(self.viper.deviceSN,time_str ,intens) def convert_buf_2_float(self, buff, bit = 3 ,byteOrder= "big" )-> None: res = [] len_ = len(buff) if len_%4 != 0: return res if byteOrder == "big": for i in range( int(len_/4) ): tmp = struct.unpack(">f", buff[i*4: i*4+4] ) res.append( round(tmp[0],bit) ) else: for i in range( int(len_/4) ): tmp = struct.unpack(">f", buff[i*4: i*4+4] ) res.append(round(tmp[0],bit)) return res pass def convert_str_2_float_list(self, lst )-> None: res = [] for l in lst: res.append(float(l)) return res pass