0 medlemmar och 2 gäster tittar på detta ämne.
from pyModbusTCP.client import ModbusClientfrom pyModbusTCP import utilsimport numpy as np# TCP auto connect on modbus request, close after itnibe_IP = "192.168.x.xx"def get_nibe_currents(IP): error = False current = np.zeros(4, dtype = float) try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 46/47 BE3, 48/49 BE2, 50/51, BE1 regs = c.read_input_registers(46, 6) if regs: regs_32bit = utils.word_list_to_long(regs[::-1]) current[1] = regs_32bit[0] / 10 current[2] = regs_32bit[1] / 10 current[3] = regs_32bit[2] / 10 current[0] = np.sum(current[1 : ]) # Total currents else: print("read error") else: print('Fatal error') error = False return(current)def get_nibe_power(IP): error = False power = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 2166/2167 regs = c.read_input_registers(2166, 2) if regs: regs_32bit = utils.word_list_to_long(regs[::-1]) power = regs_32bit[0] else: print("read error") else: print('Fatal error') error = False return(power)def get_nibe_DM(IP): error = False DM = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 11/12 regs = c.read_holding_registers(11, 2) if regs: regs_32bit = utils.word_list_to_long(regs[::-1]) regs_32bit = utils.get_2comp(regs_32bit[0], val_size = 32) DM = regs_32bit / 10 else: print("read error") else: print('Fatal error') error = False return(DM)def get_nibe_temp_out(IP): error = False temp_out = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 1 regs = c.read_input_registers(1, 1) if regs: regs_16bit = utils.get_2comp(regs[0], val_size = 16) temp_out = regs_16bit / 10 else: print("read error") else: print('Fatal error') error = False return(temp_out)def get_nibe_temp_in(IP): error = False temp_out = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 26 regs = c.read_input_registers(26, 1) if regs: regs_16bit = utils.get_2comp(regs[0], val_size = 16) temp_in = regs_16bit / 10 else: print("read error") else: print('Fatal error') error = False return(temp_in)def get_nibe_brine(IP): error = False brine_in = 0 brine_out = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 10/11 regs = c.read_input_registers(10, 2) if regs: regs_16bit = utils.get_list_2comp(regs, val_size = 16) brine_in = regs_16bit[0] / 10 brine_out = regs_16bit[1] / 10 else: print("read error") else: print('Fatal error') error = False return(brine_in, brine_out)def get_nibe_comp_freq(IP): error = False comp_freq = 0 try: c = ModbusClient(host = IP, auto_open = True, auto_close = True, timeout = 5) except ValueError: print("Error with host or port params") error = True if error == False: # reg 1046 regs = c.read_input_registers(1046, 1) if regs: regs_16bit = regs[0] comp_freq = regs_16bit / 10 else: print("read error") else: print('Fatal error') error = False return(comp_freq)print('Temp out: ' + str(get_nibe_temp_out(nibe_IP)) + ' degC')print('Temp in: ' + str(get_nibe_temp_in(nibe_IP)) + ' degC')print('Brine in/out: ' + str(get_nibe_brine(nibe_IP)) + ' degC')print('Power: ' + str(get_nibe_power(nibe_IP)) + ' W')print('Comp freq: ' + str(get_nibe_comp_freq(nibe_IP)) + ' Hz')print('Degree minutes: ' + str(get_nibe_DM(nibe_IP)))print('Ltot, L1, L2, L3: ' + str(get_nibe_currents(nibe_IP)) + ' A')
import requestsrequests.packages.urllib3.disable_warnings()from requests.auth import HTTPBasicAuthimport numpy as npfloat_formatter = "{:.1f}".formatnp.set_printoptions(formatter={'float_kind':float_formatter})nibe_IP = "https://192.168.x.xx:8443"Serial_NO = "xxxxxxxxxxxxxxxx"# Get device points from Nibedef get_nibe_point(url, point_id, device_id): error = False auth = HTTPBasicAuth('username', 'password') headers = { "accept": "*/*", "content-type": "application/json" } try: response = requests.get(url + '/api/v1/devices/' + str(device_id) + '/points/' + str(point_id), headers = headers, auth = auth, timeout = 5, verify = False) response.raise_for_status() except requests.exceptions.HTTPError as errh: print("HTTP Error") print(errh.args[0]) error = True except requests.exceptions.ReadTimeout: print("Time out") error = True except requests.exceptions.ConnectionError: print("Connection error") error = True except requests.exceptions.RequestException: print("Exception request") error = True if error == False: if response.status_code == 200: message = response.json() elif response.status_code == 429: print('Too many requests !!!!!!') else: message = 'Failed' else: message = 'Fatal error' error = False return(message) def get_nibe_currents(url, device_id): current = np.zeros(4, dtype = float) data = np.zeros(4, dtype = dict) BE = np.array([27816, 27815, 27814]) i = 1 for point in BE: data[i] = get_nibe_point(url, point, device_id) if data[i] == 'Failed' or data[i] == 'Fatal error': current[i] = 999 else: current[i] = data[i]['value']['integerValue'] / data[i]['metadata']['divisor'] i += 1 current[0] = current[0] = np.sum(current[1 : ]) # Total currents return(current)def get_nibe_power(url, device_id): power = np.zeros(3, dtype = float) data = np.zeros(3, dtype = dict) P = np.array([22130, 1801, 1932]) i = 0 for point in P: data[i] = get_nibe_point(url, point, device_id) if data[i] == 'Failed' or data[i] == 'Fatal error': power[i] = 999 else: power[i] = data[i]['value']['integerValue'] / data[i]['metadata']['divisor'] i += 1 return(power)def get_nibe_DM(url, device_id): dm = 0 point = 781 data = get_nibe_point(url, point, device_id) if data == 'Failed' or data == 'Fatal error': dm = 999 else: dm = data['value']['integerValue'] / data['metadata']['divisor'] return(dm)def get_nibe_temp_out(url,device_id): temp_out = 0 point = 4 data = get_nibe_point(url, point, device_id) if data == 'Failed' or data == 'Fatal error': temp_out = 999 else: temp_out = data['value']['integerValue'] / data['metadata']['divisor'] return(temp_out)def get_nibe_temp_in(url, device_id): temp_in = 0 point = 29 data = get_nibe_point(url, point, device_id) if data == 'Failed' or data == 'Fatal error': temp_in = 999 else: temp_in = data['value']['integerValue'] / data['metadata']['divisor'] return(temp_in)def get_nibe_brine(url, device_id): brine = np.zeros(2, dtype = float) data = np.zeros(3, dtype = dict) BR = np.array([13,14]) i = 0 for point in BR: data[i] = get_nibe_point(url, point, device_id) if data[i] == 'Failed' or data[i] == 'Fatal error': brine[i] = 999 else: brine[i] = data[i]['value']['integerValue'] / data[i]['metadata']['divisor'] i += 1 return(brine)def get_nibe_comp_freq(url, device_id): freq = 0 point = 1799 data = get_nibe_point(url, point, device_id) if data == 'Failed' or data == 'Fatal error': freq = 999 else: freq = data['value']['integerValue'] / data['metadata']['divisor'] return(freq)print('Temp out: ' + str(get_nibe_temp_out(nibe_IP, Serial_NO)) + ' degC')print('Temp in: ' + str(get_nibe_temp_in(nibe_IP, Serial_NO)) + ' degC')print('Brine in/out: ' + str(get_nibe_brine(nibe_IP, Serial_NO)) + ' degC')print('Power (Instant, CP, CP AVG): ' + str(get_nibe_power(nibe_IP, Serial_NO)) + ' W')print('Comp freq: ' + str(get_nibe_comp_freq(nibe_IP, Serial_NO)) + ' Hz')print('Degree minutes: ' + str(get_nibe_DM(nibe_IP, Serial_NO)))print('Ltot, L1, L2, L3: ' + str(get_nibe_currents(nibe_IP, Serial_NO)) + ' A')