pwarren/pyFroniusReg
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

DataType rebuild working

Browse source
This commit is contained in:
Paul Warren 2024-10-11 11:10:47 +11:00
parent 9ae239bc18
commit 348febb0d2
1 changed files with 36 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
src/pyfroniusreg/froniusreg.py
View file

@ -13,7 +13,7 @@ from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
#from pymodbus.other_message import *
#from pymodbus.mei_message import *
class DataType(Enum):
class DataType():
String8 = 1
String16 = 2
String32 = 3
@ -24,24 +24,26 @@ class DataType(Enum):
Float32 = 8
UInt64 = 7
# Returns the length (amount) of the registers.
# This corresponds to the value from the Fronius Excel list (column "Size").
# This refers to how many registers the Mobus function read_holding_registers()
# must read to get the complete value
def getRegisterLength(self):
def __init__(self, width, decode, add):
self._width = width
self._decode = decode
self._add = add
if (self == DataType.String8) or (self == DataType.UInt64):
return int(4)
elif (self == DataType.String16):
return int(8)
elif (self == DataType.String32):
return int(16)
elif (self == DataType.Int16) or (self == DataType.UInt16):
return int(1)
elif (self == DataType.Int32) or (self == DataType.UInt32) or (self == DataType.Float32):
return int(2)
@property
def width(self):
return self._width
def decode_from_register(self, value):
def decode_from_register(self, registers):
decoder = BinaryPayloadDecoder.fromRegisters(registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
return self._decode(decoder)
def encode_to_buffer(self, value):
encoder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
self._add(encoder, value)
return encoder.build()
def old_decode_from_register(self, value):
decoder = BinaryPayloadDecoder.fromRegisters(value.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
if (self == DataType.String8) or (self == DataType.String16) or (self == DataType.String32):
@ -65,7 +67,7 @@ class DataType(Enum):
else:
return str(decoder.decode_bits())
def encode_to_buffer(self, value):
def old_encode_to_buffer(self, value):
encoder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
if (self == DataType.String8) or (self == DataType.String16) or (self == DataType.String32):
@ -92,18 +94,20 @@ class DataType(Enum):
else:
return encoder.add_bits(value).build()
# Constants
string8 = DataType.String8
string16 = DataType.String16
string32 = DataType.String32
int16 = DataType.Int16
uint16 = DataType.UInt16
int32 = DataType.Int32
uint32 = DataType.UInt32
float32 = DataType.Float32
uint64 = DataType.UInt64
def decode_string(decoder, value):
return str(decoder.decode_string(16).decode('utf-8'))
string8 = DataType(4, decode_string, BinaryPayloadBuilder.add_string)
string16 = DataType(8, decode_string, BinaryPayloadBuilder.add_string)
string32 = DataType(16, decode_string, BinaryPayloadBuilder.add_string)
int16 = DataType(1,BinaryPayloadDecoder.decode_16bit_int, BinaryPayloadBuilder.add_16bit_int)
uint16 = DataType(1,BinaryPayloadDecoder.decode_16bit_uint, BinaryPayloadBuilder.add_16bit_uint)
int32 = DataType(2,BinaryPayloadDecoder.decode_32bit_int, BinaryPayloadBuilder.add_32bit_int)
uint32 = DataType(2,BinaryPayloadDecoder.decode_32bit_uint, BinaryPayloadBuilder.add_32bit_uint)
float32 = DataType(2,BinaryPayloadDecoder.decode_32bit_float, BinaryPayloadBuilder.add_32bit_float)
uint64 = DataType(4,BinaryPayloadDecoder.decode_64bit_uint, BinaryPayloadBuilder.add_64bit_uint)
class registerReadError(Exception):
pass
@ -123,18 +127,18 @@ class FroniusReg:
def __getRegisterValue(self, modbusClient):
modbusValue = modbusClient.read_holding_registers(self.address-1,
self.datatype.getRegisterLength(),
self.datatype.width,
slave=self.unit)
if(modbusValue.isError()):
raise registerReadError("Unable to read from Fronius Register: %d, %s" % (self.id, self.description))
if(modbusValue is None):
raise registerReadError("It's NONE!")
return self.datatype.decode_from_register(modbusValue)
return self.datatype.decode_from_register(modbusValue.registers)
def __setRegisterValue(self, modbusClient, value):
modbusValue = modbusClient.write_registers(self.address-1,
self.datatype.encode_to_buffer(value),
count=self.datatype.getRegisterLength(),
count=self.datatype.width,
slave=self.unit,
skip_encode=True)
return modbusValue