Code: Alles auswählen
from pymodbus.client.sync import ModbusTcpClient
client = ModbusTcpClient('192.168.1.50', port=502)
client.connect()
response = client.read_holding_registers(44, 8, unit=1)
if response.isError():
print("Error reading registers")
else:
firmware = ''.join([chr(x >> 8) + chr(x & 0xFF) for x in response.registers]).strip()
print("Firmware Version:", firmware)
client.close()Meinst du so:ChristophR hat geschrieben: Fr Mär 21, 2025 7:00 pm Kann hier vielleicht jemand helfen, wie ich einen String per Modbus auslesen kann?
Als ModbusDataType gibt es den leider nicht, habe schon probiert das hinzuzufügen, kommt aber nichts sinnvolles raus.
Ich brauche das zur Ermittlung der Firmware.
https://knowledge-center.solaredge.com/ ... l-note.pdf
Address | Size | Name | Type | Description
40044 | 8 | C_Version | String(16) | SolarEdge Specific
Da kommt dann z.B. sowas raus:
0004.0022.0039
mit 2 NULL Werten am Ende, damit es 16 Byte sind.
Ich habe es jetzt mal so probiert, da der Verbindungsaufbau ja bereits an anderer Stelle passiert:Basti hat geschrieben: Fr Mär 21, 2025 7:13 pmCode: Alles auswählen
from pymodbus.client.sync import ModbusTcpClient client = ModbusTcpClient('192.168.1.50', port=502) client.connect() response = client.read_holding_registers(44, 8, unit=1) if response.isError(): print("Error reading registers") else: firmware = ''.join([chr(x >> 8) + chr(x & 0xFF) for x in response.registers]).strip() print("Firmware Version:", firmware) client.close()
Code: Alles auswählen
response = self.__tcp_client.read_holding_registers(40044, 8, unit=unit)
self.inverter_firmware = ''.join([chr(x >> 8) + chr(x & 0xFF) for x in response.registers]).strip()
log.debug(f"Response: {response} Firmware: {self.inverter_firmware}")Code: Alles auswählen
2025-03-21 22:54:56,871 - {helpermodules.logger:181} - {ERROR:set power limit 4} - Uncaught exception in threading.excepthook:
Traceback (most recent call last):
File "/var/www/html/openWB/packages/modules/common/modbus.py", line 101, in __read_registers
number_of_addresses = sum(divide_rounding_up(
File "/var/www/html/openWB/packages/modules/common/modbus.py", line 102, in <genexpr>
t.bits, _MODBUS_HOLDING_REGISTER_SIZE) for t in types)
AttributeError: 'int' object has no attribute 'bits'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/usr/lib/python3.9/threading.py", line 954, in _bootstrap_inner
self.run()
File "/usr/lib/python3.9/threading.py", line 892, in run
self._target(*self._args, **self._kwargs)
File "/var/www/html/openWB/packages/modules/devices/solaredge/solaredge/bat.py", line 97, in set_power_limit
response = self.__tcp_client.read_holding_registers(40044, 8, unit=unit)
File "/var/www/html/openWB/packages/modules/common/modbus.py", line 138, in read_holding_registers
return self.__read_registers(
File "/var/www/html/openWB/packages/modules/common/modbus.py", line 121, in __read_registers
raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e
Exception: modules.common.modbus <class 'AttributeError'> 'int' object has no attribute 'bits'Code: Alles auswählen
#!/usr/bin/env python3
import logging
from typing import Dict, Union, Optional
from pymodbus.constants import Endian
from control import data
from dataclass_utils import dataclass_from_dict
from modules.common import modbus
from modules.common.abstract_device import AbstractBat
from modules.common.component_state import BatState
from modules.common.component_type import ComponentDescriptor
from modules.common.fault_state import ComponentInfo, FaultState
from modules.common.modbus import ModbusDataType
from modules.common.simcount import SimCounter
from modules.common.store import get_bat_value_store
from modules.devices.solaredge.solaredge.config import SolaredgeBatSetup
import pymodbus
log = logging.getLogger(__name__)
FLOAT32_UNSUPPORTED = -0xffffff00000000000000000000000000
class SolaredgeBat(AbstractBat):
# Define all possible registers with their data types
REGISTERS = {
"Battery1StateOfEnergy": (0xf584, ModbusDataType.FLOAT_32,), # Dezimal 62852
"Battery1InstantaneousPower": (0xf574, ModbusDataType.FLOAT_32,), # Dezimal 62836
"StorageControlMode": (0xe004, ModbusDataType.UINT_16,),
"StorageChargeDischargeDefaultMode": (0xe00a, ModbusDataType.UINT_16,),
"RemoteControlCommandMode": (0xe00d, ModbusDataType.UINT_16,),
"RemoteControlCommandDischargeLimit": (0xe010, ModbusDataType.FLOAT_32,),
}
def __init__(self,
device_id: int,
component_config: Union[Dict, SolaredgeBatSetup],
tcp_client: modbus.ModbusTcpClient_) -> None:
self.__device_id = device_id
self.component_config = dataclass_from_dict(SolaredgeBatSetup, component_config)
self.__tcp_client = tcp_client
self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="speicher")
self.store = get_bat_value_store(self.component_config.id)
self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
self.last_mode = 'undefined'
def update(self) -> None:
self.store.set(self.read_state())
firmware = self._read_firmware_version(self.component_config.configuration.modbus_id)
if firmware:
log.info(f"SolarEdge Firmware-Version (Unit {self.component_config.configuration.modbus_id}): {firmware}")
def read_state(self) -> BatState:
unit = self.component_config.configuration.modbus_id
registers_to_read = [
"Battery1InstantaneousPower",
"Battery1StateOfEnergy",
]
values = self._read_registers(registers_to_read, unit)
if values["Battery1InstantaneousPower"] == FLOAT32_UNSUPPORTED:
values["Battery1InstantaneousPower"] = 0
imported, exported = self.sim_counter.sim_count(values["Battery1InstantaneousPower"])
bat_state = BatState(
power=values["Battery1InstantaneousPower"],
soc=values["Battery1StateOfEnergy"],
imported=imported,
exported=exported
)
log.debug(f"Bat {self.__tcp_client.address}: {bat_state}")
return bat_state
def set_power_limit(self, power_limit: Optional[int]) -> None:
unit = self.component_config.configuration.modbus_id
PowerLimitMode = data.data.bat_all_data.data.config.power_limit_mode
if PowerLimitMode == 'no_limit':
return
if power_limit is None:
if self.last_mode is not None:
log.debug("Keine Speichersteuerung gefordert, Steuerung deaktivieren.")
values_to_write = {
"RemoteControlCommandDischargeLimit": 5000,
"StorageChargeDischargeDefaultMode": 0,
"RemoteControlCommandMode": 0,
"StorageControlMode": 2,
}
self._write_registers(values_to_write, unit)
self.last_mode = None
else:
return
elif power_limit >= 0 and self.last_mode != 'stop':
log.debug("Speichersteuerung aktivieren. Speicher-Entladung sperren.")
values_to_write = {
"StorageControlMode": 4,
"StorageChargeDischargeDefaultMode": 1,
"RemoteControlCommandMode": 1,
}
self._write_registers(values_to_write, unit)
self.last_mode = 'stop'
def _read_registers(self, register_names: list, unit: int) -> Dict[str, Union[int, float]]:
values = {}
for key in register_names:
address, data_type = self.REGISTERS[key]
values[key] = self.__tcp_client.read_holding_registers(
address, data_type, wordorder=Endian.Little, unit=unit
)
log.debug(f"Bat raw values {self.__tcp_client.address}: {values}")
return values
def _write_registers(self, values_to_write: Dict[str, Union[int, float]], unit: int) -> None:
for key, value in values_to_write.items():
address, data_type = self.REGISTERS[key]
encoded_value = self._encode_value(value, data_type)
self.__tcp_client.write_registers(address, encoded_value, unit=unit)
log.debug(f"Neuer Wert {encoded_value} in Register {address} geschrieben.")
def _encode_value(self, value: Union[int, float], data_type: ModbusDataType) -> list:
builder = pymodbus.payload.BinaryPayloadBuilder(
byteorder=pymodbus.constants.Endian.Big,
wordorder=pymodbus.constants.Endian.Little
)
encode_methods = {
ModbusDataType.UINT_32: builder.add_32bit_uint,
ModbusDataType.INT_32: builder.add_32bit_int,
ModbusDataType.UINT_16: builder.add_16bit_uint,
ModbusDataType.INT_16: builder.add_16bit_int,
ModbusDataType.FLOAT_32: builder.add_32bit_float,
}
if data_type == ModbusDataType.FLOAT_32:
encode_methods[data_type](float(value))
else:
encode_methods[data_type](int(value))
return builder.to_registers()
def _read_firmware_version(self, unit: int) -> Optional[str]:
try:
address = 43 # 40044 - 40001 = 43
length = 8
result = self.__tcp_client.client.read_holding_registers(address=address, count=length, unit=unit)
if result.isError():
log.warning(f"Fehler beim Lesen der Firmware-Version von Unit {unit}")
return None
from pymodbus.payload import BinaryPayloadDecoder
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers, byteorder=Endian.Big, wordorder=Endian.Big
)
firmware = decoder.decode_string(length * 2).decode('ascii').strip('\x00')
return firmware
except Exception as e:
log.warning(f"Ausnahme beim Lesen der Firmware: {e}")
return None
component_descriptor = ComponentDescriptor(configuration_factory=SolaredgeBatSetup)Hallo Basti,
Code: Alles auswählen
#!/usr/bin/env python3
import logging
from typing import Dict, Union, Optional
from pymodbus.constants import Endian
from control import data
from dataclass_utils import dataclass_from_dict
from modules.common import modbus
from modules.common.abstract_device import AbstractBat
from modules.common.component_state import BatState
from modules.common.component_type import ComponentDescriptor
from modules.common.fault_state import ComponentInfo, FaultState
from modules.common.modbus import ModbusDataType
from modules.common.simcount import SimCounter
from modules.common.store import get_bat_value_store
from modules.devices.solaredge.solaredge.config import SolaredgeBatSetup
import pymodbus
log = logging.getLogger(__name__)
FLOAT32_UNSUPPORTED = -0xffffff00000000000000000000000000
class SolaredgeBat(AbstractBat):
# Define all possible registers with their data types
REGISTERS = {
"Battery1StateOfEnergy": (0xf584, ModbusDataType.FLOAT_32,), # Dezimal 62852
"Battery1InstantaneousPower": (0xf574, ModbusDataType.FLOAT_32,), # Dezimal 62836
"StorageControlMode": (0xe004, ModbusDataType.UINT_16,),
"StorageChargeDischargeDefaultMode": (0xe00a, ModbusDataType.UINT_16,),
"RemoteControlCommandMode": (0xe00d, ModbusDataType.UINT_16,),
"RemoteControlCommandDischargeLimit": (0xe010, ModbusDataType.FLOAT_32,),
}
def __init__(self,
device_id: int,
component_config: Union[Dict, SolaredgeBatSetup],
tcp_client: modbus.ModbusTcpClient_) -> None:
self.__device_id = device_id
self.component_config = dataclass_from_dict(SolaredgeBatSetup, component_config)
self.__tcp_client = tcp_client
self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="speicher")
self.store = get_bat_value_store(self.component_config.id)
self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
self.last_mode = 'undefined'
def update(self) -> None:
self.store.set(self.read_state())
def read_state(self) -> BatState:
unit = self.component_config.configuration.modbus_id
registers_to_read = [
"Battery1InstantaneousPower",
"Battery1StateOfEnergy",
]
values = self._read_registers(registers_to_read, unit)
if values["Battery1InstantaneousPower"] == FLOAT32_UNSUPPORTED:
values["Battery1InstantaneousPower"] = 0
imported, exported = self.sim_counter.sim_count(values["Battery1InstantaneousPower"])
bat_state = BatState(
power=values["Battery1InstantaneousPower"],
soc=values["Battery1StateOfEnergy"],
imported=imported,
exported=exported
)
log.debug(f"Bat {self.__tcp_client.address}: {bat_state}")
return bat_state
def set_power_limit(self, power_limit: Optional[int]) -> None:
unit = self.component_config.configuration.modbus_id
PowerLimitMode = data.data.bat_all_data.data.config.power_limit_mode
firmware = self._read_firmware_version(unit)
if firmware:
log.info(f"SolarEdge Firmware-Version (Unit {unit}): {firmware}")
if PowerLimitMode == 'no_limit':
return
if power_limit is None:
if self.last_mode is not None:
log.debug("Keine Speichersteuerung gefordert, Steuerung deaktivieren.")
values_to_write = {
"RemoteControlCommandDischargeLimit": 5000,
"StorageChargeDischargeDefaultMode": 0,
"RemoteControlCommandMode": 0,
"StorageControlMode": 2,
}
self._write_registers(values_to_write, unit)
self.last_mode = None
else:
return
elif power_limit >= 0 and self.last_mode != 'stop':
log.debug("Speichersteuerung aktivieren. Speicher-Entladung sperren.")
values_to_write = {
"StorageControlMode": 4,
"StorageChargeDischargeDefaultMode": 1,
"RemoteControlCommandMode": 1,
}
self._write_registers(values_to_write, unit)
self.last_mode = 'stop'
def _read_registers(self, register_names: list, unit: int) -> Dict[str, Union[int, float]]:
values = {}
for key in register_names:
address, data_type = self.REGISTERS[key]
values[key] = self.__tcp_client.read_holding_registers(
address, data_type, wordorder=Endian.Little, unit=unit
)
log.debug(f"Bat raw values {self.__tcp_client.address}: {values}")
return values
def _write_registers(self, values_to_write: Dict[str, Union[int, float]], unit: int) -> None:
for key, value in values_to_write.items():
address, data_type = self.REGISTERS[key]
encoded_value = self._encode_value(value, data_type)
self.__tcp_client.write_registers(address, encoded_value, unit=unit)
log.debug(f"Neuer Wert {encoded_value} in Register {address} geschrieben.")
def _encode_value(self, value: Union[int, float], data_type: ModbusDataType) -> list:
builder = pymodbus.payload.BinaryPayloadBuilder(
byteorder=pymodbus.constants.Endian.Big,
wordorder=pymodbus.constants.Endian.Little
)
encode_methods = {
ModbusDataType.UINT_32: builder.add_32bit_uint,
ModbusDataType.INT_32: builder.add_32bit_int,
ModbusDataType.UINT_16: builder.add_16bit_uint,
ModbusDataType.INT_16: builder.add_16bit_int,
ModbusDataType.FLOAT_32: builder.add_32bit_float,
}
if data_type == ModbusDataType.FLOAT_32:
encode_methods[data_type](float(value))
else:
encode_methods[data_type](int(value))
return builder.to_registers()
def _read_firmware_version(self, unit: int) -> Optional[str]:
try:
address = 43 # 40044 - 40001 = 43
length = 8
result = self.__tcp_client.read_holding_registers(address=address, count=length, unit=unit)
if result.isError():
log.warning(f"Fehler beim Lesen der Firmware-Version von Unit {unit}")
return None
from pymodbus.payload import BinaryPayloadDecoder
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers, byteorder=Endian.Big, wordorder=Endian.Big
)
firmware = decoder.decode_string(length * 2).decode('ascii').strip('\x00')
return firmware
except Exception as e:
log.warning(f"Ausnahme beim Lesen der Firmware: {e}")
return None
component_descriptor = ComponentDescriptor(configuration_factory=SolaredgeBatSetup)Code: Alles auswählen
def _read_firmware_version(self, unit: int) -> Optional[str]:
try:
address = 43 # 40044 - 40001 = 43
length = 8
# Direkt auf den pymodbus-Client zugreifen!
result = self.__tcp_client.client.read_holding_registers(address=address, count=length, unit=unit)
if result.isError():
log.warning(f"Fehler beim Lesen der Firmware-Version von Unit {unit}")
return None
from pymodbus.payload import BinaryPayloadDecoder
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers, byteorder=Endian.Big, wordorder=Endian.Big
)
firmware = decoder.decode_string(length * 2).decode('ascii').strip('\x00')
return firmware
except Exception as e:
log.warning(f"Ausnahme beim Lesen der Firmware: {e}")
return NoneCode: Alles auswählen
from pymodbus.client.sync import ModbusTcpClient
client = ModbusTcpClient('192.168.1.199', port=502)
client.connect()
response = client.read_holding_registers(40044, 8, unit=1)
if response.isError():
print("Error reading registers")
else:
firmware = ''.join([chr(x >> 8) + chr(x & 0xFF) for x in response.registers]).strip()
print("Firmware Version:", firmware)
client.close()
Code: Alles auswählen
openwb@raspberrypi:~ $ python test.py
Firmware Version: 0004.0022.0044
Erst ab Firmware 4.20.36 wird Time of Use unterstützt.snoerenberg hat geschrieben: Sa Mär 22, 2025 7:09 pm @Christoph - wieso willst du die Firmware des Meters auslesen?
Oder willst du die Firmware der Batterie oder des Wechselrichters auslesen?
Gruß Stephan
Code: Alles auswählen
from pymodbus.client.sync import ModbusTcpClient
client = ModbusTcpClient('192.168.1.199', port=502)
client.connect()
response = client.read_holding_registers(address=40036, count=16, unit=1)
print("Read register: 40036")
if response.isError():
print("Error reading registers")
else:
firmware = ''.join([chr(x >> 8) + chr(x & 0xFF) for x in response.registers]).strip()
print("Firmware Version:", firmware)
client.close()
Bei self.__tcp_client.client.read_holding_registers(...Basti hat geschrieben: Sa Mär 22, 2025 7:01 pmCode: Alles auswählen
def _read_firmware_version(self, unit: int) -> Optional[str]: try: address = 43 # 40044 - 40001 = 43 length = 8 # Direkt auf den pymodbus-Client zugreifen! result = self.__tcp_client.client.read_holding_registers(address=address, count=length, unit=unit) if result.isError(): log.warning(f"Fehler beim Lesen der Firmware-Version von Unit {unit}") return None from pymodbus.payload import BinaryPayloadDecoder decoder = BinaryPayloadDecoder.fromRegisters( result.registers, byteorder=Endian.Big, wordorder=Endian.Big ) firmware = decoder.decode_string(length * 2).decode('ascii').strip('\x00') return firmware except Exception as e: log.warning(f"Ausnahme beim Lesen der Firmware: {e}") return None