Kategorien
Allgemein

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
Allgemein

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()