##------------------------------------------------------------------------------- # Name: FlowMeter Driver # Purpose: Provide an Interface for the serial communication with an SF06 chip # based Sensirion Flow Meter # Author: Jonathan G # # Last Version:14.12.2021 # Copyright: (c) Institute of technical biocatalysis TUHH # Licence: GNU GPL-v3 #------------------------------------------------------------------------------- # # Based upon the SHDLC communicator provided by Sensirion # ------------------------------------------------------------------------------- # Name: SHDLC Command Library for Python # Purpose: Demonstration of SHDLC communication with RS485 adapter cable # over serial interface. # # Author: nsaratz # # Created: 04. 10. 2012 # Copyright: (c) Sensirion AG 2012 # Licence: All Rights Reserved #------------------------------------------------------------------------------- from SHDLC_communicator import SHDLC_SerialConnection import time import asyncio #import matplotlib.pyplot as plt --> not in use class FlowCmds: SensorStatus : int = int(0x30) StartConti : int = int(0x33) StopConti : int = int(0x34) LastMeasure : int = int(0x35) GetBuffer : int = int(0x36) SensorPart : int = int(0x50) ScaleCheck : int = int(0x53) SensorSerial : int = int(0x54) SensorReset : int = int(0x65) class Flowmeter(SHDLC_SerialConnection): def __init__(self,COM_Port, print_debug_info = False, address = 0): """ Init class. (Duh) """ self.permRecord = False self.SignalNumber = 3 self.address = address self.started_measurement = False self.connected = False self.MeasureIntervall = 1000 self.healthy = False self.COM_Port = COM_Port self.FlowScaleFactor = 10 self.TempScaleFactor = 200 #print('Flowmeter init start') super().__init__(print_debug_info=print_debug_info) #print('Flowmeter init done') async def initialize(self): """ Procedure to initialize the actual sensor and not just the python object. """ self.connect(self.COM_Port) await asyncio.sleep(0.5) self.performIntegrityCheck() def calculateMeasureInterval(self,interval): """ Procedure to calculate and create the array to set the interval time in ms. """ hexstring = hex(interval).replace('0x','') if len(hexstring)>2: hexHead = int(hexstring[:-2],16) hexTail = int(hexstring[-2:],16) else: hexHead = 0 hexTail = int(hexstring[-2:],16) return [hexHead,hexTail] def connect(self,COM_Port_number): """ Procedure to connect to a COM port connected flow meter. """ super().open_connection(COM_Port_number) self.connected = True async def start_contiMeasure(self,MeasureInterval=100): """ Procedure to start the continuous measurement cycle. """ if self.started_measurement == True: return 'Measurement already started' if MeasureInterval<0: raise ValueError("Negative Measurement intervals are phycially impossible") tempHexCode = hex(MeasureInterval).replace('0x','') if len(tempHexCode)>4: raise ValueError("Measure Interval out of range of 16 bit interval") else: HexCode = tempHexCode for i in range(4-len(tempHexCode)): HexCode = '0'+HexCode HexHead = int(HexCode[:2],16) HexTail = int(HexCode[-2:],16) await self.send_a_receive(self.address,FlowCmds.StartConti,[HexHead,HexTail,54,8]) self.started_measurement = True async def stop_contiMeasure(self): """ Routine to stop the continuous measurement of flow rates. The buffer is NOT deleted and has to be cleared separately or emptied """ if not self.started_measurement: raise RuntimeError("Cannot stop a measurement that is not running") await self.send_a_receive(self.address,FlowCmds.StopConti,[]) async def send_a_receive(self, address, command, parameters): """ Local implementation of the send and recieve procedure of the communicator in case in the future manipulation has to be done. """ msg = super().SHDLC_send_and_receive(address, command, parameters) return msg async def clear_buffer(self): """ Clears the buffer of the flow meter, deleting any and all data remaining. Should only be called as part of the startup routine """ msg = await self.send_a_receive(self.address,FlowCmds.GetBuffer,[2]) async def get_last_flow(self): """ Returns the last and only the last flow measurement. Use the average flow measurement instead. """ param = 0 if not self.permRecord: param |= (1<<0) if self.SignalNumber == 3: param |= (1<<1) msg = await self.send_a_receive(self.address,FlowCmds.LastMeasure,[param]) data = msg[0] errors = msg[1] if errors != 0: await self.analyzeErrors(errors) return await self.processFlowData(data) async def get_conti_flow_measurement(self): msg,errors = await self.send_a_receive(self.address,0x36,[3]) header = msg[:8] data = msg[8:] lostPackages = int('0x'+''.join([str(a) for a in header[:3]]),16) remainingPackages = int('0x'+''.join([str(a) for a in header[4:5]]),16) InterlacedData = int('0x'+str(header[6])+str(header[7]),16) if lostPackages != 0 and self.debuginfo == True: print('Packages were lost due to an overflow') if remainingPackages !=0 and self.debuginfo == True: print('Some packages remain in the buffer') if len(data)%InterlacedData !=0: print('Data is missing') raise ValueError('Data is missing') else: dataPackages = [] for i in range(0,len(data)-6,6): dataPackages.append(data[i:i+6]) flowRates,temperatures,flags =self.processListFlowData(dataPackages) return flowRates,temperatures,flags def processListFlowData(self,dataSet): flowRates = [] temperatures = [] flags = [] for data in dataSet: if len(data) == 6: flowRates.append(self.processFlowData(data[:2])) temperatures.append(self.processTempData(data[2:4])) flags.append(self.convertSignalFlags(data[4:6])) elif len(data) == 2: flowRates.append(self.processFlowData((data))) else: raise ValueError('') return flowRates,temperatures,flags def convertSignalFlags(self,data): if len(data)!=2: raise ValueError('The signal flags have to be 2 bytes') DecData = data[0]*256 + data[1] BinData = bin(DecData) flag = 0 ''' if BinData[-1] == '1': flag |= (1<<0) if BinData[-2] == '1': flag|= (1<<1) if BinData[-6] == '1': flag|=(1<<2) ''' return 4 def processTempData(self,data): """ Calculates the physical temperature values """ if len(data) != 2: raise ValueError("The temperature measurements should always return 2 bytes") val = data[0] * 256 + data[1] val = val/self.TempScaleFactor return val def processFlowData(self,data): """ Calculates the physical flow values """ if len(data) != 2: raise ValueError("The flow measurements should always return 2 bytes") m = data[0] * 256 + data[1] if m&(2**15) == 2**15: val = m - 2**16 else: val = m val = val/self.FlowScaleFactor return val def analyzeErrors(self,errors): """ This function should translate the error codes to readable statements and hopefully take appropriate action """ raise NotImplementedError('Error analyzing is not yet available') def performIntegrityCheck(self): """ This procedure should perform an integrity check to make sure everything works as intended. """ pass