from audioop import add import re import time import asyncio import sys import json import requests import mariadb from smbus2 import SMBus # -- BME280 -- from ctypes import c_short from ctypes import c_byte from ctypes import c_ubyte # -- Kimo (VolStrom) -- from pyModbusTCP.client import ModbusClient from pymodbus.constants import Endian from pymodbus.payload import BinaryPayloadDecoder from pymodbus.payload import BinaryPayloadBuilder # -- INA219 (Strom) -- from ina219 import INA219 from ina219 import DeviceRangeError class bcolors: MAGENTA = '\033[95m' BLUE = '\033[94m' CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' buses = {} # -------------- SDP810 ------------------------------------------ async def init_sen_sdp810(addr, bus_nr): bus = buses[bus_nr] bus.write_i2c_block_data(addr, 0x3F, [0xF9]) #Stop any cont measurement of the sensor time.sleep(0.1) bus.write_i2c_block_data(addr, 0x36, [0x1E]) time.sleep(0.1) # bus.close() async def query_sen_sdp810(addr, bus_nr): bus = buses[bus_nr] data = bus.read_i2c_block_data(addr, 0, 9) press = -twos_comp((256 * data[0] + data[1]), 16) / 240.0 temp = twos_comp((256 * data[3] + data[4]), 16) / 200 # bus.close() return press def twos_comp(val, bits): if (val & (1 << (bits - 1))) != 0: # if sign bit is set e.g., 8bit: 128-255# val = val - (1 << bits) # compute negative value return val # -------------- BMP280 ------------------------------------------ async def init_sen_bmp280(addr, bus_nr): pass async def query_sen_bmp280(addr, bus_nr): # print(f' * {addr}, {bus_nr}') bus = buses[bus_nr] REG_DATA = 0xF7 REG_CONTROL = 0xF4 REG_CONFIG = 0xF5 REG_CONTROL_HUM = 0xF2 REG_HUM_MSB = 0xFD REG_HUM_LSB = 0xFE # Oversample setting - page 27 OVERSAMPLE_TEMP = 2 OVERSAMPLE_PRES = 2 MODE = 1 # Oversample setting for humidity register - page 26 OVERSAMPLE_HUM = 2 bus.write_byte_data(addr, REG_CONTROL_HUM, OVERSAMPLE_HUM) control = OVERSAMPLE_TEMP<<5 | OVERSAMPLE_PRES<<2 | MODE bus.write_byte_data(addr, REG_CONTROL, control) # Read blocks of calibration data from EEPROM # See Page 22 data sheet cal1 = bus.read_i2c_block_data(addr, 0x88, 24) cal2 = bus.read_i2c_block_data(addr, 0xA1, 1) cal3 = bus.read_i2c_block_data(addr, 0xE1, 7) # Convert byte data to word values dig_T1 = getUShort(cal1, 0) dig_T2 = getShort(cal1, 2) dig_T3 = getShort(cal1, 4) dig_P1 = getUShort(cal1, 6) dig_P2 = getShort(cal1, 8) dig_P3 = getShort(cal1, 10) dig_P4 = getShort(cal1, 12) dig_P5 = getShort(cal1, 14) dig_P6 = getShort(cal1, 16) dig_P7 = getShort(cal1, 18) dig_P8 = getShort(cal1, 20) dig_P9 = getShort(cal1, 22) dig_H1 = getUChar(cal2, 0) dig_H2 = getShort(cal3, 0) dig_H3 = getUChar(cal3, 2) dig_H4 = getChar(cal3, 3) dig_H4 = (dig_H4 << 24) >> 20 dig_H4 = dig_H4 | (getChar(cal3, 4) & 0x0F) dig_H5 = getChar(cal3, 5) dig_H5 = (dig_H5 << 24) >> 20 dig_H5 = dig_H5 | (getUChar(cal3, 4) >> 4 & 0x0F) dig_H6 = getChar(cal3, 6) # Wait in ms (Datasheet Appendix B: Measurement time and current calculation) wait_time = 1.25 + (2.3 * OVERSAMPLE_TEMP) + ((2.3 * OVERSAMPLE_PRES) + 0.575) + ((2.3 * OVERSAMPLE_HUM)+0.575) time.sleep(wait_time/1000) # Wait the required time # Read temperature/pressure/humidity data = bus.read_i2c_block_data(addr, REG_DATA, 8) pres_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4) temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4) hum_raw = (data[6] << 8) | data[7] #Refine temperature var1 = ((((temp_raw>>3)-(dig_T1<<1)))*(dig_T2)) >> 11 var2 = (((((temp_raw>>4) - (dig_T1)) * ((temp_raw>>4) - (dig_T1))) >> 12) * (dig_T3)) >> 14 t_fine = var1+var2 temperature = float(((t_fine * 5) + 128) >> 8) # Refine pressure and adjust for temperature var1 = t_fine / 2.0 - 64000.0 var2 = var1 * var1 * dig_P6 / 32768.0 var2 = var2 + var1 * dig_P5 * 2.0 var2 = var2 / 4.0 + dig_P4 * 65536.0 var1 = (dig_P3 * var1 * var1 / 524288.0 + dig_P2 * var1) / 524288.0 var1 = (1.0 + var1 / 32768.0) * dig_P1 if var1 == 0: pressure=0 else: pressure = 1048576.0 - pres_raw pressure = ((pressure - var2 / 4096.0) * 6250.0) / var1 var1 = dig_P9 * pressure * pressure / 2147483648.0 var2 = pressure * dig_P8 / 32768.0 pressure = pressure + (var1 + var2 + dig_P7) / 16.0 # Refine humidity humidity = t_fine - 76800.0 humidity = (hum_raw - (dig_H4 * 64.0 + dig_H5 / 16384.0 * humidity)) * (dig_H2 / 65536.0 * (1.0 + dig_H6 / 67108864.0 * humidity * (1.0 + dig_H3 / 67108864.0 * humidity))) humidity = humidity * (1.0 - dig_H1 * humidity / 524288.0) if humidity > 100: humidity = 100 elif humidity < 0: humidity = 0 # bus.close() return temperature / 100.0, pressure / 100.0, humidity def getShort(data, index): return c_short((data[index+1] << 8) + data[index]).value def getUShort(data, index): return (data[index+1] << 8) + data[index] def getChar(data,index): result = data[index] if result > 127: result -= 256 return result def getUChar(data,index): result = data[index] & 0xFF return result # -------------- EMC2101 ------------------------------------------ EMC2101_I2CADDR_DEFAULT = 0x4C EMC2101_CHIP_ID = 0x16 EMC2101_ALT_CHIP_ID = 0x28 EMC2101_WHOAMI = 0xFD EMC2101_STATUS = 0x02 EMC2101_REG_CONFIG = 0x03 EMC2101_REG_DATA_RATE = 0x04 EMC2101_TEMP_FORCE = 0x0C EMC2101_TACH_LSB = 0x46 EMC2101_TACH_MSB = 0x47 EMC2101_TACH_LIMIT_LSB = 0x48 EMC2101_TACH_LIMIT_MSB = 0x49 EMC2101_FAN_CONFIG = 0x4A EMC2101_FAN_SPINUP = 0x4B EMC2101_REG_FAN_SETTING = 0x4C EMC2101_PWM_FREQ = 0x4D EMC2101_PWM_DIV = 0x4E EMC2101_LUT_HYSTERESIS = 0x4F MAX_LUT_SPEED = 0x3F # 6-bit value MAX_LUT_TEMP = 0x7F # 7-bit async def init_sen_fan_rpm(addr, bus_nr): pass async def query_sen_fan_rpm(addr, bus_nr): bus = buses[bus_nr] tach_lsb = bus.read_byte_data(EMC2101_I2CADDR_DEFAULT, EMC2101_TACH_LSB) tach_msb = bus.read_byte_data(EMC2101_I2CADDR_DEFAULT, EMC2101_TACH_MSB) rpm = 5400000 / (tach_msb * 256 + tach_lsb) # bus.close() return rpm # -------------- Kimo ------------------------------------------ MODBUS_SERVER_HOST = "192.168.178.150" MODBUS_SERVER_PORT = 502 async def init_sen_kimo_volflow(addr, bus_nr): global modbusC310 global isModbusVolFlowConnected global hasErrorOccured # try: modbusC310 = ModbusClient(MODBUS_SERVER_HOST, MODBUS_SERVER_PORT) # modbusC310 = ModbusClient() # modbusC310.host(MODBUS_SERVER_HOST) # modbusC310.port(MODBUS_SERVER_PORT) if modbusC310.open(): print("Connected to " + MODBUS_SERVER_HOST + ":" + str(MODBUS_SERVER_PORT)) isModbusVolFlowConnected = True else: print("Unable to connect to " + MODBUS_SERVER_HOST + ":" + str(MODBUS_SERVER_PORT)) isModbusVolFlowConnected = False # except: # hasErrorOccured = True # isModbusVolFlowConnected = False # print("ERROR -> Unable to connect to " + MODBUS_SERVER_HOST + ":" + str(MODBUS_SERVER_PORT)) async def query_sen_kimo_volflow(addr, bus_nr): global modbusC310 global hasErrorOccured volFlow = 0 try: regs = modbusC310.read_holding_registers(7040, 4) #print(regs) decoder = BinaryPayloadDecoder.fromRegisters(regs, Endian.Big, wordorder=Endian.Little) volFlow = decoder.decode_32bit_float() except: hasErrorOccured = True print("ERROR - getVolumeflow -> ", sys.exc_info()) return volFlow # def getFlowSpeed(): # global modbusC310 # global hasErrorOccured # volFlow = 0 # try: # regs = modbusC310.read_holding_registers(7040, 4) # Kanal 2 (m/s) # #print(regs) # decoder = BinaryPayloadDecoder.fromRegisters(regs, Endian.Big, wordorder=Endian.Little) # volFlow = decoder.decode_32bit_float() # except: # hasErrorOccured = True # print("ERROR - getVolumeflow -> ", sys.exc_info()) # return volFlow # -------------- INA219 ------------------------------------------ SHUNT_OHMS = 0.1 INA_MEAS_PAUSE = 0.1 # in s async def init_sen_ina219(addr, bus_nr): global ina ina = INA219(SHUNT_OHMS, busnum = bus_nr) ina.configure() asyncio.create_task(query_sen_ina219_highres()) async def query_sen_ina219_highres(): global ina global curAvg curAvg = 0 curs = [] while True: current = ina.current() curs.append(current) if (len(curs) == 20): curAvg = sum(curs) / len(curs) curs.pop(0) await asyncio.sleep(INA_MEAS_PAUSE); async def query_sen_ina219(addr, bus_nr): global curAvg # current = ina.current() # return current return curAvg # -------------- Fan-Rapsi ------------------------------------------ async def init_sen_raspi(addr, bus_nr): pass async def query_sen_raspi(addr, bus_nr): res = requests.get('http://192.168.178.101/hub/sen_socket_adapter.php') return res.json() # -------------- Mux (tca9548a) ------------------------------------------ MUX_CHANNELS = [0b00000001, 0b00000010, 0b00000100, 0b00001000, 0b00010000, 0b00100000, 0b01000000, 0b10000000] def sel_mux_channel(addr, bus_nr, channel): # print(f'{bus_nr} {addr} {channel}') # print(f'{type(bus_nr)} {type(addr)} {type(channel)} {type(MUX_CHANNELS[channel])}') # print(f'{type(bus_nr)}') # bus = SMBus(bus_nr) bus = buses[bus_nr] bus.write_byte(addr, MUX_CHANNELS[channel]) time.sleep(0.01) # ----------------------------------------------------------------- def write_val_to_db(session_pid, sensor_pid, value): global conn with conn.cursor(buffered = True) as cur: cur.execute('INSERT INTO data.sensor_data(sensor, meas_session, val) VALUES(?, ?, ?)', (sensor_pid, session_pid, value)) # ----------------------------------------------------------------- async def main(): global conn sensors = [] try: conn = mariadb.connect(user = 'meas_www', password = '6yWmVAw+a70wGToAFoJnWA', host = 'localhost', port = 3306) # conn.autocommit = True with conn.cursor(buffered = True) as cur: cur.execute('SELECT pid, i2c_bus_nr FROM site.sensor_location WHERE com_type = 0 AND i2c_bus_nr IS NOT NULL') for (pid, i2c_bus_nr) in cur: buses[i2c_bus_nr] = SMBus(int(i2c_bus_nr)) # print(buses) with conn.cursor(buffered = True) as cur: # cur.execute('SELECT s_phys.pid, s_type.pid, s_phys.i2c_address, loc.pid, loc.i2c_bus_nr, s_type.query_func_info FROM sensor.sensor_physical s_phys, sensor.sensor_type s_type, sensor.sensor_location loc WHERE s_phys.sensor_type = s_type.pid AND s_phys.location = loc.pid AND s_phys.is_enabled = 1') cur.execute('SELECT s_phys.pid, s_type.pid, s_phys.i2c_address, loc.pid, loc.i2c_bus_nr, s_type.query_func_info, s_phys.mux_channel FROM site.sensor_physical s_phys, sensor.sensor_type s_type, site.sensor_location loc WHERE s_phys.sensor_type = s_type.pid AND s_phys.location = loc.pid AND s_phys.is_enabled = 1') # for (sen) in cur: # sensors.append(dict(sen)) # print(dict(sen)) for (sen_pyhs_pid, sen_type_pid, i2c_address, loc_pid, i2c_bus_nr, query_func_info, mux_channel) in cur: with conn.cursor(buffered = True) as cur2: # cur2.execute('SELECT pid, meas_type FROM sensor.sensor WHERE sensor_physical = ? ORDER BY meas_type', (sen_pyhs_pid, )) cur2.execute('SELECT pid, meas_type FROM site.sensor WHERE sensor_physical = ? ORDER BY meas_type', (sen_pyhs_pid, )) sens = [] for (sen_pid, sen_type) in cur2: sens.append({'sen_pid' : sen_pid, 'sen_type' : sen_type}) print(f' {sen_pid} - {sen_type}') sensors.append({'sen_pyhs_pid' : sen_pyhs_pid, 'sens' : sens, 'sen_type_pid' : sen_type_pid, 'i2c_address' : i2c_address, 'loc_pid' : loc_pid, 'i2c_bus_nr' : i2c_bus_nr, 'query_func_info' : query_func_info, 'mux_channel': mux_channel }) print(f'{sen_pyhs_pid}, {sen_type_pid} : {loc_pid} ({i2c_bus_nr} - {i2c_address}, {mux_channel}) - {query_func_info}') except mariadb.Error as e: print(f'Error: {e}') print() print('>> initializing sensors...') for sen in sensors: init_func_name = 'init_' + sen['query_func_info'] print(f' sensor {sen["sen_pyhs_pid"]} : {sen["i2c_bus_nr"]}, {sen["i2c_address"]}, {sen["mux_channel"]} > {init_func_name}') if sen['i2c_address'].isnumeric(): i2c_address = int('0x' + sen['i2c_address'], 16) else: i2c_address = None if sen['mux_channel'] is not None: sel_mux_channel(0x71, sen['i2c_bus_nr'], int(sen['mux_channel'])) # print('----------------') # print(locals()) # print('----------------') # print(' --> ' + locals()[init_func_name]) # print(f' initializing sensor {sen["sen_pyhs_pid"]} ({init_func_name})') # meas_data = await locals()[init_func_name](i2c_address, sen['i2c_bus_nr']) # print(' --> ' + str(globals()[init_func_name])) meas_data = await globals()[init_func_name](i2c_address, sen['i2c_bus_nr']) print('') print('>> starting continuous measurement...') meas_session_id = 0 # while (False): while True: with conn.cursor(buffered = True) as cur: cur.execute('INSERT INTO data.meas_session() VALUES()') meas_session_id = cur.lastrowid meas_start_time = time.time() print(f'---- {meas_session_id} -----------------------') for sen in sensors: query_func_name = 'query_' + sen['query_func_info'] if sen['i2c_address'].isnumeric(): i2c_address = int('0x' + sen['i2c_address'], 16) else: i2c_address = None if sen['mux_channel'] is not None: sel_mux_channel(0x71, sen['i2c_bus_nr'], int(sen['mux_channel'])) # print('---> ' + query_func_name) meas_data = await globals()[query_func_name](i2c_address, sen['i2c_bus_nr']) # meas_data = await locals()[query_func_name](i2c_address, sen['i2c_bus_nr']) print(f' {sen["sen_pyhs_pid"]} ({query_func_name}) > {meas_data}') if sen['sen_type_pid'] == 2: # bmp280 write_val_to_db(meas_session_id, sen['sens'][1]['sen_pid'], meas_data[0]) # Temp write_val_to_db(meas_session_id, sen['sens'][0]['sen_pid'], meas_data[1]) # Druck abs write_val_to_db(meas_session_id, sen['sens'][2]['sen_pid'], meas_data[2]) # Feuchte rel pass elif sen['sen_type_pid'] == 8: # Fan-Raspi write_val_to_db(meas_session_id, sen['sens'][0]['sen_pid'], meas_data['current']) write_val_to_db(meas_session_id, sen['sens'][1]['sen_pid'], meas_data['rpm']) write_val_to_db(meas_session_id, sen['sens'][2]['sen_pid'], meas_data['pwm']) # elif sen['sen_type_pid'] == 3: # sdp810 # write_val_to_db(meas_session_id, sen['sens'][0]['sen_pid'], meas_data) else: write_val_to_db(meas_session_id, sen['sens'][0]['sen_pid'], meas_data) conn.commit() meas_end_time = time.time() meas_duration = meas_end_time - meas_start_time meas_pause = 1 - meas_duration if meas_pause < 0: meas_pause = 0 print(f'* took {(meas_duration * 1000):.1f} ms - pausing {(meas_pause * 1000):.1f} ms') await asyncio.sleep(meas_pause) # time.sleep(meas_pause) asyncio.run(main())