Skip to content

Solaredge: use modbus bulk reader for bat #3198

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LKuemmel merged 1 commit into from
Jun 3, 2026
Merged

Solaredge: use modbus bulk reader for bat #3198

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 73 additions & 105 deletions packages/modules/devices/solaredge/solaredge/bat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,8 @@
#!/usr/bin/env python3
from enum import IntEnum
import logging

from typing import Any, TypedDict, Dict, Union, Optional, Tuple


from pymodbus.constants import Endian
import pymodbus


from modules.common import modbus
from modules.common.abstract_device import AbstractBat
Expand Down Expand Up @@ -36,21 +32,28 @@ class KwargsDict(TypedDict):
client: modbus.ModbusTcpClient_


class SolaredgeBat(AbstractBat):
# Define all possible registers with their data types
REGISTERS = {
"Battery1StateOfEnergy": (0xe184, ModbusDataType.FLOAT_32,), # Mirror: 0xf584
"Battery1InstantaneousPower": (0xe174, ModbusDataType.FLOAT_32,), # Mirror: 0xf574
"Battery2StateOfEnergy": (0xe284, ModbusDataType.FLOAT_32,),
"Battery2InstantaneousPower": (0xe274, ModbusDataType.FLOAT_32,),
"StorageControlMode": (0xe004, ModbusDataType.UINT_16,),
"StorageBackupReserved": (0xe008, ModbusDataType.FLOAT_32,),
"RemoteControlCommandModeDefault": (0xe00a, ModbusDataType.UINT_16,),
"RemoteControlCommandMode": (0xe00d, ModbusDataType.UINT_16,),
"RemoteControlChargeLimit": (0xe00e, ModbusDataType.FLOAT_32,),
"RemoteControlDischargeLimit": (0xe010, ModbusDataType.FLOAT_32,),
}
class Registers(IntEnum):
STORAGE_CONTROL_MODE = 0xe004
REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG = 0xe00a
REMOTE_CONTROL_COMMAND_MODE = 0xe00d
REMOTE_CONTROL_CHARGE_LIMIT = 0xe00e
REMOTE_CONTROL_DISCHARGE_LIMIT = 0xe010
BAT_1_POWER = 0xe174
BAT_1_SOC = 0xe184
BAT_2_POWER = 0xe274
BAT_2_SOC = 0xe284


WRITING_DATA_TYPES = {
Registers.STORAGE_CONTROL_MODE: ModbusDataType.UINT_16,
Registers.REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG: ModbusDataType.UINT_16,
Registers.REMOTE_CONTROL_COMMAND_MODE: ModbusDataType.UINT_16,
Registers.REMOTE_CONTROL_CHARGE_LIMIT: ModbusDataType.FLOAT_32,
Registers.REMOTE_CONTROL_DISCHARGE_LIMIT: ModbusDataType.FLOAT_32,
}


class SolaredgeBat(AbstractBat):
def __init__(self, component_config: SolaredgeBatSetup, **kwargs: Any) -> None:
self.component_config = component_config
self.kwargs: KwargsDict = kwargs
Expand Down Expand Up @@ -79,32 +82,19 @@ def read_state(self):

def get_values(self) -> Tuple[float, float]:
unit = self.component_config.configuration.modbus_id
# Use 1 as fallback if battery_index is not set
battery_index = getattr(self.component_config.configuration, "battery_index", 1)

# Define base registers for Battery 1 in hex
base_soc_reg = 0xE184 # Battery 1 SoC
base_power_reg = 0xE174 # Battery 1 Power
offset = 0x100 # 256 bytes in hex

# Adjust registers based on battery_index
if battery_index == 1:
soc_reg = base_soc_reg
power_reg = base_power_reg
elif battery_index == 2:
soc_reg = base_soc_reg + offset # 0xE284
power_reg = base_power_reg + offset # 0xE274
else:
raise ValueError(f"Invalid battery_index: {battery_index}. Must be 1 or 2.")

# Read SoC and Power from the appropriate registers
soc = self.__tcp_client.read_holding_registers(
soc_reg, ModbusDataType.FLOAT_32, wordorder=Endian.Little, unit=unit
)
power = self.__tcp_client.read_holding_registers(
power_reg, ModbusDataType.FLOAT_32, wordorder=Endian.Little, unit=unit
battery_index = self.component_config.configuration.battery_index

Copilot AI Apr 22, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

battery_index is no longer validated: any value other than 1 will silently select Battery 2 registers. This changes behavior vs. the previous explicit ValueError and can mask configuration mistakes (e.g. battery_index=0 reads the wrong battery). Consider validating that battery_index is either 1 or 2 (raise a clear exception or fall back to 1) before selecting the register base.

Suggested change
battery_index = self.component_config.configuration.battery_index
battery_index = self.component_config.configuration.battery_index
if battery_index not in (1, 2):
raise ValueError(f"Invalid battery_index {battery_index}. Expected 1 or 2.")

Copilot uses AI. Check for mistakes.
power_reg = Registers.BAT_1_POWER if battery_index == 1 else Registers.BAT_2_POWER
soc_reg = Registers.BAT_1_SOC if battery_index == 1 else Registers.BAT_2_SOC
bulk = (
(power_reg, ModbusDataType.FLOAT_32),
(soc_reg, ModbusDataType.FLOAT_32),
)

resp = self.__tcp_client.read_holding_registers_bulk(
power_reg, 18, mapping=bulk, wordorder=Endian.Little, unit=unit)
log.debug(f"Bat raw values {self.__tcp_client.address}: {resp}")
power = resp[power_reg]
soc = resp[soc_reg]
# Handle unsupported case
if power == FLOAT32_UNSUPPORTED:
power = 0
Expand All @@ -118,110 +108,88 @@ def get_imported_exported(self, power: float) -> Tuple[float, float]:

def set_power_limit(self, power_limit: Optional[int]) -> None:
unit = self.component_config.configuration.modbus_id
# Use 1 as fallback if battery_index is not set
battery_index = getattr(self.component_config.configuration, "battery_index", 1)

registers_to_read = [
"StorageControlMode",
"RemoteControlCommandMode",
"RemoteControlChargeLimit",
"RemoteControlDischargeLimit",
]
try:
values = self._read_registers(registers_to_read, unit)
except pymodbus.exceptions.ModbusException as e:
log.error(f"Failed to read registers: {e}")
self.fault_state.error(f"Modbus read error: {e}")
return
battery_index = self.component_config.configuration.battery_index

bulk = (
(Registers.STORAGE_CONTROL_MODE, ModbusDataType.UINT_16),
(Registers.REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG, ModbusDataType.UINT_16),
(Registers.REMOTE_CONTROL_COMMAND_MODE, ModbusDataType.UINT_16),
(Registers.REMOTE_CONTROL_CHARGE_LIMIT, ModbusDataType.FLOAT_32),
(Registers.REMOTE_CONTROL_DISCHARGE_LIMIT, ModbusDataType.FLOAT_32),
)

values = self.__tcp_client.read_holding_registers_bulk(
Registers.STORAGE_CONTROL_MODE, 14, wordorder=Endian.Little, mapping=bulk, unit=unit)
log.debug(f"Bat raw values {self.__tcp_client.address}: {values}")

if power_limit is None: # No Bat Control should be used.
if values["StorageControlMode"] == CONTROL_MODE_MSC:
if values[Registers.STORAGE_CONTROL_MODE] == CONTROL_MODE_MSC:
log.debug(f"Speicher{battery_index}:Keine Steuerung gefordert, bereits deaktiviert.")
else:
# Disable Bat Control
values_to_write = {
"RemoteControlChargeLimit": MAX_CHARGEDISCHARGE_LIMIT,
"RemoteControlDischargeLimit": MAX_CHARGEDISCHARGE_LIMIT,
"RemoteControlCommandModeDefault": REMOTE_CONTROL_COMMAND_MODE_DEFAULT,
"RemoteControlCommandMode": REMOTE_CONTROL_COMMAND_MODE_DEFAULT,
"StorageControlMode": CONTROL_MODE_MSC,
Registers.REMOTE_CONTROL_CHARGE_LIMIT: MAX_CHARGEDISCHARGE_LIMIT,
Registers.REMOTE_CONTROL_DISCHARGE_LIMIT: MAX_CHARGEDISCHARGE_LIMIT,
Registers.REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG: REMOTE_CONTROL_COMMAND_MODE_DEFAULT,
Registers.REMOTE_CONTROL_COMMAND_MODE: REMOTE_CONTROL_COMMAND_MODE_DEFAULT,
Registers.STORAGE_CONTROL_MODE: CONTROL_MODE_MSC,
}
self._write_registers(values_to_write, unit)
log.debug(f"Speicher{battery_index}:Keine Steuerung gefordert, Steuerung deaktiviert.")

elif power_limit <= 0: # Limit Discharge Mode should be used.
if (values["StorageControlMode"] == CONTROL_MODE_REMOTE and
values["RemoteControlCommandMode"] == REMOTE_CONTROL_COMMAND_MODE_MSC):
if (values[Registers.STORAGE_CONTROL_MODE] == CONTROL_MODE_REMOTE and
values[Registers.REMOTE_CONTROL_COMMAND_MODE] == REMOTE_CONTROL_COMMAND_MODE_MSC):
# Remote Control and Discharge Mode already active.
discharge_limit = int(values["RemoteControlDischargeLimit"])
discharge_limit = int(values[Registers.REMOTE_CONTROL_DISCHARGE_LIMIT])
if discharge_limit not in range(int(abs(power_limit)) - 10, int(abs(power_limit)) + 10):
# Send Limit only if difference is more than 10W, needed with more than 1 battery.
values_to_write = {
"RemoteControlDischargeLimit": int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
Registers.REMOTE_CONTROL_DISCHARGE_LIMIT: int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
}
self._write_registers(values_to_write, unit)
log.debug(f"Entlade-Limit Speicher{battery_index}: {int(abs(power_limit))}W.")
else:
log.debug(f"Entlade-Limit Speicher{battery_index}: Abweichung unter +/- 10W.")
else: # Enable Remote Control and Discharge Mode.
values_to_write = {
"StorageControlMode": CONTROL_MODE_REMOTE,
"RemoteControlCommandModeDefault": REMOTE_CONTROL_COMMAND_MODE_MSC,
"RemoteControlCommandMode": REMOTE_CONTROL_COMMAND_MODE_MSC,
"RemoteControlDischargeLimit": int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
Registers.STORAGE_CONTROL_MODE: CONTROL_MODE_REMOTE,
Registers.REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG: REMOTE_CONTROL_COMMAND_MODE_MSC,
Registers.REMOTE_CONTROL_COMMAND_MODE: REMOTE_CONTROL_COMMAND_MODE_MSC,
Registers.REMOTE_CONTROL_DISCHARGE_LIMIT: int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
}
self._write_registers(values_to_write, unit)
log.debug(f"Entlade-Limit aktiviert, Speicher{battery_index}: {int(abs(power_limit))}W.")

elif power_limit > 0: # Charge Mode should be used
if (values["StorageControlMode"] == CONTROL_MODE_REMOTE and
values["RemoteControlCommandMode"] == REMOTE_CONTROL_COMMAND_MODE_CHARGE):
if (values[Registers.STORAGE_CONTROL_MODE] == CONTROL_MODE_REMOTE and
values[Registers.REMOTE_CONTROL_COMMAND_MODE] == REMOTE_CONTROL_COMMAND_MODE_CHARGE):
# Remote Control and Charge Mode already active.
charge_limit = int(values["RemoteControlChargeLimit"])
charge_limit = int(values[Registers.REMOTE_CONTROL_CHARGE_LIMIT])
if charge_limit not in range(int(abs(power_limit)) - 10, int(abs(power_limit)) + 10):
# Send Limit only if difference is more than 10W.
values_to_write = {
"RemoteControlChargeLimit": int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
Registers.REMOTE_CONTROL_CHARGE_LIMIT: int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
}
self._write_registers(values_to_write, unit)
log.debug(f"Ladung Speicher{battery_index}: {int(abs(power_limit))}W.")
else:
log.debug(f"Ladung Speicher{battery_index}: Abweichung unter +/- 10W.")
else: # Enable Remote Control and Charge Mode.
values_to_write = {
"StorageControlMode": CONTROL_MODE_REMOTE,
"RemoteControlCommandModeDefault": REMOTE_CONTROL_COMMAND_MODE_CHARGE,
"RemoteControlCommandMode": REMOTE_CONTROL_COMMAND_MODE_CHARGE,
"RemoteControlChargeLimit": int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
Registers.STORAGE_CONTROL_MODE: CONTROL_MODE_REMOTE,
Registers.REMOTE_CONTROL_COMMAND_MODE_DEFAULT_REG: REMOTE_CONTROL_COMMAND_MODE_CHARGE,
Registers.REMOTE_CONTROL_COMMAND_MODE: REMOTE_CONTROL_COMMAND_MODE_CHARGE,
Registers.REMOTE_CONTROL_CHARGE_LIMIT: int(min(abs(power_limit), MAX_CHARGEDISCHARGE_LIMIT))
}
self._write_registers(values_to_write, unit)
log.debug(f"Aktivierung Ladung Speicher{battery_index}: {int(abs(power_limit))}W.")

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]
try:
values[key] = self.__tcp_client.read_holding_registers(
address, data_type, wordorder=Endian.Little, unit=unit
)
except pymodbus.exceptions.ModbusException as e:
log.error(f"Failed to read register {key} at address {address}: {e}")
self.fault_state.error(f"Modbus read error: {e}")
values[key] = 0 # Fallback value
log.debug(f"Bat raw values {self.__tcp_client.address}: {values}")
return values
# TODO: Optimize to read multiple contiguous registers in a single request if supported by ModbusTcpClient_

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]
try:
self.__tcp_client.write_register(address, value, data_type, wordorder=Endian.Little, unit=unit)
log.debug(f"Neuer Wert {value} in Register {address} geschrieben.")
except pymodbus.exceptions.ModbusException as e:
log.error(f"Failed to write register {key} at address {address}: {e}")
self.fault_state.error(f"Modbus write error: {e}")
def _write_registers(self, values_to_write: Dict[Registers, Union[int, float]], unit: int) -> None:
for address, value in values_to_write.items():
self.__tcp_client.write_register(
address, value, WRITING_DATA_TYPES[address], wordorder=Endian.Little, unit=unit)
log.debug(f"Neuer Wert {value} in Register {address} geschrieben.")

def power_limit_controllable(self) -> bool:
return True
Expand Down
Loading