Installation de 7 Wallboxes chez 2 clients à Esch/Alzette
Marque GARO/PCE GLB
2 stations comme contrôleurs et 5 stations contrôlées
Il faut juste construire un petit barrage pour produire de l’énergie verte 😉
Schlagwort: ModBUS
Kategorien
SAIA-ALD1B5FD00A3A00
Programme en Python pour lire les valeurs
#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""
Modbus TestKit: Implementation of Modbus protocol in python
(C)2009 - Luc Jean - luc.jean@gmail.com
(C)2009 - Apidev - http://www.apidev.fr
This is distributed under GNU LGPL license, see license.txt
"""
# https://code.google.com/archive/p/modbus-tk/wikis/ModbusMasterExample.wiki
# To install dependencies:
# pip install modbus-tk
# pip install pyserial
import serial
import modbus_tk
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu
from datetime import datetime
# Settings for Saia Burgess ALD1B5FD
PORT = 'COM5'
PARITY = 'E'
BAUDRATE = 4800
address = 6
address = int(input("Enter the address of your module: "))
print ("Asking values for address:?", address)
def main():
"""main"""
now = datetime.now()
logger = modbus_tk.utils.create_logger("console")
print ("Testing of Saia Burgess ALD1B5FD")
print("by Electro-Tech s.à r.l.")
print("15, rue Centrale")
print("L-4499 Limpach")
print("email: serge.klein AT electro-tech.lu")
print("Date of test: ",now.strftime("%B %d, %Y %H:%M:%S"))
try:
#Connect to the slave
master = modbus_rtu.RtuMaster(
serial.Serial(port=PORT, baudrate=BAUDRATE, bytesize=8, parity=PARITY, stopbits=1, xonxoff=0)
)
master.set_timeout(5.0)
master.set_verbose(False)
logger.info("connected")
#data_format makes possible to extract the data like defined in the
#struct python module documentation
# Get FW Version
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 0, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 0, 1)
tempValue=temp[0]
print ("Firmware:", tempValue)
# Get number of supported registers
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 1, 1)
tempValue=temp[0]
print ("Number of supported registers:", tempValue)
# Get BAUDRATE
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 4, 2)
tempValue=temp[0]
print ("Baudrate:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 6, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 7, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 8, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 9, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 10, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 11, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 12, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 13, 1)
tempValue=temp[0]
print ("Type /ASN function:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 14, 1)
tempValue=temp[0]
print ("HW vers. modif.:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 15, 2)
tempValue=temp[0]+temp[1]
print ("Serial number:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 17, 1)
tempValue=temp[0]
print ("Serial number:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 21, 1)
tempValue=temp[0]
print ("Status:", tempValue)
# Get Type /ASN function
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 22, 1)
tempValue=temp[0]
print ("Response timeout [ms]:", tempValue)
# Get address
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 23, 1)
tempValue=temp[0]
print ("Device address:", tempValue)
# Get error register
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 24, 1)
tempValue=temp[0]
print ("Error register:", tempValue)
# Tariff register
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 26, 1)
tempValue=temp[0]
print ("Tariff register:", tempValue)
# Counter energy total tariff 1
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 27, 2)
tempValue=temp[0]*65536+temp[1]
print ("Counter energy total tariff 1 [kWh]:", tempValue*0.01)
# Counter energy partial tariff 1
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 29, 2)
tempValue=temp[0]*65536+temp[1]
print ("Counter energy partial tariff 1 [kWh]:", tempValue*0.01)
# Get URMS phase 1
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 35, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 35, 1)
tempValue=temp[0]
print ("URMS phase 1 [V]:", tempValue)
# Get IRMS phase 1
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 36, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 36, 1)
tempValue=temp[0]
print ("IRMS phase 1 [A]:", tempValue*0.1)
# Get PRMS phase 1
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 37, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 37, 1)
tempValue=temp[0]
print ("PRMS phase 1 [kW]:", tempValue*0.01)
# Get QRMS phase 1
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 38, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 38, 1)
tempValue=temp[0]
print ("QRMS phase 1 [kvar]:", tempValue*0.01)
# Get cos phi phase 1
#logger.info(master.execute(address, cst.READ_HOLDING_REGISTERS, 39, 1))
temp=master.execute(address, cst.READ_HOLDING_REGISTERS, 39, 1)
tempValue=temp[0]
print ("cosphi phase 1 :", tempValue*0.01)
#send some queries
#logger.info(master.execute(1, cst.READ_COILS, 0, 10))
#logger.info(master.execute(1, cst.READ_DISCRETE_INPUTS, 0, 8))
#logger.info(master.execute(1, cst.READ_INPUT_REGISTERS, 100, 3))
#logger.info(master.execute(1, cst.READ_HOLDING_REGISTERS, 100, 12))
#logger.info(master.execute(1, cst.WRITE_SINGLE_COIL, 7, output_value=1))
#logger.info(master.execute(1, cst.WRITE_SINGLE_REGISTER, 100, output_value=54))
#logger.info(master.execute(1, cst.WRITE_MULTIPLE_COILS, 0, output_value=[1, 1, 0, 1, 1, 0, 1, 1]))
#logger.info(master.execute(1, cst.WRITE_MULTIPLE_REGISTERS, 100, output_value=xrange(12)))
except modbus_tk.modbus.ModbusError as exc:
logger.error("%s- Code=%d", exc, exc.get_exception_code())
if __name__ == "__main__":
main()
Kategorien
Eastron SDM 120 Modbus
Programme en Python pour lire les valeurs
#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""
Modbus TestKit: Implementation of Modbus protocol in python
(C)2009 - Luc Jean - luc.jean@gmail.com
(C)2009 - Apidev - http://www.apidev.fr
This is distributed under GNU LGPL license, see license.txt
"""
# https://code.google.com/archive/p/modbus-tk/wikis/ModbusMasterExample.wiki
# To install dependencies:
# pip install modbus-tk
# pip install pyserial
import serial
import struct
import modbus_tk
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu
from datetime import datetime
from pymodbus.pdu import ModbusRequest
from pymodbus.client.sync import ModbusSerialClient as ModbusClient #initialize a serial RTU client instance
from pymodbus.transaction import ModbusRtuFramer
from pymodbus.constants import Endian # For 32-bit float numbers (2 registers / 4 bytes)
from pymodbus.payload import BinaryPayloadDecoder # For 32-bit float numers (2 registers / 4 bytes)
from pymodbus.payload import BinaryPayloadBuilder # Write 32-bit floats in register
# Settings for SDM120M
METHOD = "rtu"
PORT = 'COM5'
PARITY = 'N'
BAUDRATE = 2400
STOPBITS = 1
BYTESIZE = 8
TIMEOUT = 1
RETRIES = 2
DEBUG = False
address = 1
address = int(input("Enter the address of your module: "))
print ("Asking values for address:?", address)
def main():
"""main"""
now = datetime.now()
logger = modbus_tk.utils.create_logger("console")
print ("Testing of Eastron SDM120M")
print("by Electro-Tech s.à r.l.")
print("15, rue Centrale")
print("L-4499 Limpach")
print("email: serge.klein AT electro-tech.lu")
print("Date of test: ",now.strftime("%B %d, %Y %H:%M:%S"))
#Connect to the slave
master = modbus_rtu.RtuMaster(
serial.Serial(port=PORT, baudrate=BAUDRATE, bytesize=8, parity=PARITY, stopbits=1, xonxoff=0)
)
master.set_timeout(5.0)
master.set_verbose(False)
logger.info("connected")
# Get Voltage version 1
#https://forum-raspberrypi.de/forum/thread/37754-umwandlung-array-of-int-4-byte-in-float-nach-ieee-754/?postID=317842#post317842
try:
print ("Get voltage version 1")
print ("---------------------")
if DEBUG == True : logger.info(master.execute(address, cst.READ_INPUT_REGISTERS, 0, 2))
temp=master.execute(address, cst.READ_INPUT_REGISTERS, 0, 2)
inputArray=[temp[1], temp[0]]
int32Val = temp[1] + (temp[0] << 16)
newFloat = struct.unpack('f', struct.pack('i', int32Val))[0]
print("inputArray='{}' ==> int32Val='{}' ==> newFloat='{:.2f}' Volt".format(inputArray, int32Val, newFloat))
master.close()
except modbus_tk.modbus.ModbusError as exc:
logger.error("%s- Code=%d", exc, exc.get_exception_code())
try:
#Connect to the slave
client = ModbusClient(method = METHOD, port = PORT, stopbits = STOPBITS, bytesize = BYTESIZE, parity = PARITY, baudrate = BAUDRATE, timeout = TIMEOUT, retries = RETRIES)
connection = client.connect()
print("Connected.")
### Get Voltage version 2
print ("Get voltage version 2")
print ("---------------------")
data = client.read_input_registers(0, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big) # endian=Endian.Little / endian=Endian.Big
temp = round(decoder.decode_32bit_float(),3)
print("Voltage [V]", temp)
### Get Current
data = client.read_input_registers(6, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Current [A]", temp)
### Get active power
data = client.read_input_registers(12, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Active power [W]", temp)
### Get apparent power
data = client.read_input_registers(18, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Apparent power [VA]", temp)
### Get reactive power
data = client.read_input_registers(14, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Reactive power [VAr]", temp)
### Get power factor
data = client.read_input_registers(12, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Power factor", temp)
### Get Frequency
data = client.read_input_registers(70, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Frequency [Hz]", temp)
### Import active energy
data = client.read_input_registers(72, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Import active energy [kWh]", temp)
### Export active energy
data = client.read_input_registers(74, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Export active energy [kWh]", temp)
### Import reactive energy
data = client.read_input_registers(76, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Import reactive energy [kvarh]", temp)
### Export reactive energy
data = client.read_input_registers(78, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Export reactive energy [kvarh]", temp)
### Get Total active energy
data = client.read_input_registers(342, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float(),3)
print("Total active energy [kWh]", temp)
### Get relay pulse width
data = client.read_holding_registers(12, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float())
print("Relay pulse width [ms]", temp)
### Get network parity Stop
data = client.read_holding_registers(18, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float())
print("Network parity/stop:")
if temp == 0: print("->One stop bit and no parity")
if temp == 1: print("->One stop bit and even parity")
if temp == 2: print("->One stop bit and odd parity")
if temp == 3: print("->Two stop bits and no parity")
### Get device ID
data = client.read_holding_registers(20, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float())
print("Device ID", temp)
### Get Baudrate
data = client.read_holding_registers(28, 2, unit = address)
if DEBUG == True : print ("Data registers", data.registers)
decoder = BinaryPayloadDecoder.fromRegisters(data.registers, byteorder=Endian.Big, wordorder=Endian.Big)
temp = round(decoder.decode_32bit_float())
print("Baudrate:")
if temp == 0: print("->2400 (default)")
if temp == 1: print("->4800")
if temp == 2: print("->9600")
if temp == 3: print("->1200")
client.close()
except:
print ("Modbus connection error.")
if __name__ == "__main__":
main()