ruff formatter

This commit is contained in:
Paul Warren 2024-10-14 17:02:50 +11:00
parent e6fbb44f2a
commit faf2090fa1
5 changed files with 121 additions and 58 deletions

View file

@ -1,10 +1,12 @@
import nox
import os
@nox.session
def lint(session):
session.install('ruff')
session.run('ruff', 'check', '--exclude', 'examples')
session.install("ruff")
session.run("ruff", "check", "--exclude", "examples")
@nox.session
def build_and_check_dists(session):
@ -13,14 +15,16 @@ def build_and_check_dists(session):
session.run("python", "-m", "build")
session.run("python", "-m", "twine", "check", "dist/*")
@nox.session
def build(session):
lint(session)
build_and_check_dists(session)
@nox.session
def tests(session):
session.install('pytest')
session.install("pytest")
build_and_check_dists(session)
generated_files = os.listdir("dist/")
@ -28,5 +32,4 @@ def tests(session):
session.install(generated_sdist)
session.run('pytest')
session.run("pytest")

View file

@ -6,8 +6,8 @@
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
class DataType():
class DataType:
def __init__(self, width, decode, add):
self._width = width
self._decode = decode
@ -18,10 +18,13 @@ class DataType():
return self._width
def decode_from_register(self, registers):
decoder = BinaryPayloadDecoder.fromRegisters(registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
decoder = BinaryPayloadDecoder.fromRegisters(
registers, byteorder=Endian.BIG, wordorder=Endian.BIG
)
return self._decode(decoder)
def encode_to_buffer(self, value):
print(type(value))
encoder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
self._add(encoder, value)
return encoder.build()
@ -30,40 +33,53 @@ class DataType():
# helper functions for DataType constructors
def decode_string8(decoder):
try:
return str(decoder.decode_string(8).decode('utf-8'))
return str(decoder.decode_string(8).decode("utf-8"))
except UnicodeDecodeError:
return decoder.decode_string(8)
def decode_string16(decoder):
try:
return str(decoder.decode_string(16).decode('utf-8'))
return str(decoder.decode_string(16).decode("utf-8"))
except UnicodeDecodeError:
return decoder.decode_string(16)
def decode_string32(decoder):
try:
return str(decoder.decode_string(32).decode('utf-8'))
return str(decoder.decode_string(32).decode("utf-8"))
except UnicodeDecodeError:
return decoder.decode_string(32)
def encode_16bit_int(encoder, value):
return encoder.add_16bit_int(int(value))
# The various data types that the fronius inverters use
string8 = DataType(4, decode_string8, BinaryPayloadBuilder.add_string)
string16 = DataType(8, decode_string16, BinaryPayloadBuilder.add_string)
string32 = DataType(16, decode_string32, BinaryPayloadBuilder.add_string)
int16 = DataType(1, BinaryPayloadDecoder.decode_16bit_int, encode_16bit_int)
uint16 = DataType(1, BinaryPayloadDecoder.decode_16bit_uint, encode_16bit_int)
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)
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
# Fronius modbus register object.
#
# the 'spreadsheet' refers to the various spreadsheets in the 'gen24-modbus-api-external-docs.zip'
@ -71,6 +87,7 @@ class registerReadError(Exception):
# 'gen24 modbus'
#
# Constructor parameters:
# address: address as specified in the spreadsheet
# datatype: One of the above datatypes, as specified for the address in the spreadsheet
@ -81,7 +98,7 @@ class FroniusReg:
self.address = address
self.datatype = datatype
self.unit = unit
self.description=description
self.description = description
def getValue(self, modbusClient):
return self.__getRegisterValue(modbusClient)
@ -90,31 +107,39 @@ class FroniusReg:
return self.__setRegisterValue(modbusClient, value)
def __getRegisterValue(self, modbusClient):
modbusValue = modbusClient.read_holding_registers(self.address-1,
self.datatype.width,
slave=self.unit)
if(modbusValue.isError()):
raise registerReadError("Unable to read from Fronius Register: %d, %s\n%s" % (self.address, self.description, modbusValue))
if(modbusValue is None):
modbusValue = modbusClient.read_holding_registers(
self.address - 1, self.datatype.width, slave=self.unit
)
if modbusValue.isError():
raise registerReadError(
"Unable to read from Fronius Register: %d, %s\n%s"
% (self.address, self.description, modbusValue)
)
if modbusValue is None:
raise registerReadError("It's NONE!")
return self.datatype.decode_from_register(modbusValue.registers)
def __setRegisterValue(self, modbusClient, value):
modbusValue = modbusClient.write_registers(self.address-1,
modbusValue = modbusClient.write_registers(
self.address - 1,
self.datatype.encode_to_buffer(value),
slave=self.unit,
skip_encode=True)
skip_encode=True,
)
return modbusValue
class ScaledFroniusReg:
def __init__(self, valueReg, scaleReg):
self.valueReg=valueReg
self.scaleReg=scaleReg
self.valueReg = valueReg
self.scaleReg = scaleReg
def getValue(self, modbusClient):
return self.valueReg.getValue(modbusClient) * 10 ** self.scaleReg.getValue(modbusClient)
return self.valueReg.getValue(modbusClient) * 10 ** self.scaleReg.getValue(
modbusClient
)
def setValue(self, modbusClient, value):
return self.valueReg.setValue(modbusClient, value / 10 ** self.scaleReg.getValue(modbusClient))
return self.valueReg.setValue(
modbusClient, value / 10 ** self.scaleReg.getValue(modbusClient)
)

View file

@ -4,12 +4,25 @@ MaxChaRte = froniusreg.FroniusReg(40155, froniusreg.uint16, 1, "Max Charge Rate"
MaxChaRte_SF = froniusreg.FroniusReg(40156, froniusreg.int16, 1, "Max Charge Rate SF")
wChaGra = froniusreg.FroniusReg(40357, froniusreg.uint16, 1, "Max Charge Power")
storageStateOfCharge = froniusreg.FroniusReg(40362, froniusreg.uint16, 1, "Storage State of Charge")
storageStateOfChargeSF = froniusreg.FroniusReg(40376, froniusreg.int16, 1, "Storage State of Charge Scaling Factor")
scaledStateOfCharge = froniusreg.ScaledFroniusReg(storageStateOfCharge, storageStateOfChargeSF)
storageStateOfCharge = froniusreg.FroniusReg(
40362, froniusreg.uint16, 1, "Storage State of Charge"
)
storageStateOfChargeSF = froniusreg.FroniusReg(
40376, froniusreg.int16, 1, "Storage State of Charge Scaling Factor"
)
scaledStateOfCharge = froniusreg.ScaledFroniusReg(
storageStateOfCharge, storageStateOfChargeSF
)
ID = froniusreg.FroniusReg(40003, froniusreg.uint16, 1, "Well-known value. Uniquely identifies this as a sunspec model 'common' (1)")
L = froniusreg.FroniusReg(40004, froniusreg.uint16, 1, "Sunspec model commen register count")
ID = froniusreg.FroniusReg(
40003,
froniusreg.uint16,
1,
"Well-known value. Uniquely identifies this as a sunspec model 'common' (1)",
)
L = froniusreg.FroniusReg(
40004, froniusreg.uint16, 1, "Sunspec model commen register count"
)
Mn = froniusreg.FroniusReg(40005, froniusreg.string16, 1, "Manufacturer")
Md = froniusreg.FroniusReg(40021, froniusreg.string16, 1, "Device Model")
Vr = froniusreg.FroniusReg(40045, froniusreg.string8, 1, "SW version")
@ -19,21 +32,43 @@ DA = froniusreg.FroniusReg(40069, froniusreg.uint16, 1, "Modbus Device Address")
InputID = froniusreg.FroniusReg(40304, froniusreg.uint16, 1, "Input ID")
InputIDString = froniusreg.FroniusReg(40305, froniusreg.string8, 1, "Input ID String")
module3DCW = froniusreg.FroniusReg(40325, froniusreg.uint16, 1, "When the battery is discharged the data-points of the charge input are set to 0")
module3DCW = froniusreg.FroniusReg(
40325,
froniusreg.uint16,
1,
"When the battery is discharged the data-points of the charge input are set to 0",
)
module4DCW = froniusreg.FroniusReg(40345, froniusreg.uint16, 1, "When the battery is charged the data-points of the discharge input are set to 0")
module4DCW = froniusreg.FroniusReg(
40345,
froniusreg.uint16,
1,
"When the battery is charged the data-points of the discharge input are set to 0",
)
DCW_SF = froniusreg.FroniusReg(40268, froniusreg.int16, 1, "DC Power Scaling factor")
OutWRte = froniusreg.FroniusReg(40366, froniusreg.int16, 1, "DischargeRate")
InWRte = froniusreg.FroniusReg(40367, froniusreg.int16, 1, "ChargeRate")
WRteSF = froniusreg.FroniusReg(40379, froniusreg.int16, 1, "ScalingFactor for storage Watts")
StorCtl_Mode = froniusreg.FroniusReg(40359, froniusreg.uint16, 1, "Hold/Charge/Discharge limit")
WRteSF = froniusreg.FroniusReg(
40379, froniusreg.int16, 1, "ScalingFactor for storage Watts"
)
StorCtl_Mode = froniusreg.FroniusReg(
40359, froniusreg.uint16, 1, "Hold/Charge/Discharge limit"
)
MinRsvPct = froniusreg.FroniusReg(40361, froniusreg.uint16, 1, "Reserve Percentage")
InOutWRte_RvrtTms = froniusreg.FroniusReg(40369, froniusreg.uint16, 1, "Revert timer for charge settings")
ChaGriSet = froniusreg.FroniusReg(40371, froniusreg.uint16, 1, "enum16, 0 = PV only, 1 = Grid enabled")
WChaDisChaGra_SF = froniusreg.FroniusReg(40373, froniusreg.int16, 1, "Charge/Discharge Power SF")
MinRsvPct_SF = froniusreg.FroniusReg(40375, froniusreg.int16, 1, "Reserve Percentage Scaling")
InOutWRte_RvrtTms = froniusreg.FroniusReg(
40369, froniusreg.uint16, 1, "Revert timer for charge settings"
)
ChaGriSet = froniusreg.FroniusReg(
40371, froniusreg.uint16, 1, "enum16, 0 = PV only, 1 = Grid enabled"
)
WChaDisChaGra_SF = froniusreg.FroniusReg(
40373, froniusreg.int16, 1, "Charge/Discharge Power SF"
)
MinRsvPct_SF = froniusreg.FroniusReg(
40375, froniusreg.int16, 1, "Reserve Percentage Scaling"
)
scaledOutWRte = froniusreg.ScaledFroniusReg(OutWRte, WRteSF)
scaledInWRte = froniusreg.ScaledFroniusReg(InWRte, WRteSF)

View file

@ -3,13 +3,12 @@ import unittest
import pyfroniusreg.froniusreg as froniusreg
class TestDataTypes(unittest.TestCase):
class TestDataTypes(unittest.TestCase):
def test_int16(self):
int16_buffer = froniusreg.int16.encode_to_buffer(1024)
assert int16_buffer == [b'\x04\x00']
assert int16_buffer == [b"\x04\x00"]
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main()

View file

@ -8,8 +8,8 @@ from pymodbus.client.tcp import ModbusTcpClient
fronius1 = ModbusTcpClient("172.19.107.211", port=502, timeout=10)
fronius1.connect()
class TestRead(unittest.TestCase):
class TestRead(unittest.TestCase):
def test_read_scaled(self):
soc = gen24_registers.scaledStateOfCharge.getValue(fronius1)
assert isinstance(soc, float)
@ -46,6 +46,7 @@ class TestRead(unittest.TestCase):
retval = gen24_registers.scaledInWRte.setValue(fronius1, current)
assert retval is not None
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main()
fronius1.close()