Skip to content
Merged
Changes from all commits
Commits
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
30 changes: 25 additions & 5 deletions apps/predbat/solis.py
Original file line number Diff line number Diff line change
Expand Up @@ -2736,6 +2736,16 @@ def is_tou_v2_mode(self, sn):

# ==================== Component Lifecycle ====================

async def startup_reset_registers(self, device_sn):
"""
Reset the startup registers for the given device serial number.
"""
current_mode = self.get_current_solis_mode_value(device_sn)
new_mode = current_mode | (1 << SOLIS_BIT_BACKUP_MODE)
await self.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(new_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")
await self.read_and_write_cid(device_sn, SOLIS_CID_BATTERY_RESERVE_SOC, "5", field_description=f"Write reserve SOC to 5%")
await self.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(current_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")

Comment on lines +2742 to +2748

Copilot AI Apr 1, 2026

Copy link

Choose a reason for hiding this comment

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

The return values from read_and_write_cid() inside startup_reset_registers() are ignored, so failures won’t affect poll_success and may leave the inverter in a partially-updated state (e.g., backup bit set but reserve SOC write failed). Capture the results and either revert/restore the original mode on failure or propagate the failure up so startup can mark the poll unsuccessful.

Suggested change
"""
current_mode = self.get_current_solis_mode_value(device_sn)
new_mode = current_mode | (1 << SOLIS_BIT_BACKUP_MODE)
await self.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(new_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")
await self.read_and_write_cid(device_sn, SOLIS_CID_BATTERY_RESERVE_SOC, "5", field_description="Test write reserve SOC to 5%")
new_mode = current_mode & ~(1 << SOLIS_BIT_BACKUP_MODE)
await self.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(new_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")
This briefly enables backup mode, writes a low reserve SOC, then restores
the original storage mode. Any failure during these writes will attempt
to restore the original mode and will raise an exception so callers can
treat startup as unsuccessful.
"""
current_mode = self.get_current_solis_mode_value(device_sn)
# Enable backup bit temporarily
backup_mode = current_mode | (1 << SOLIS_BIT_BACKUP_MODE)
ok = await self.read_and_write_cid(
device_sn,
SOLIS_CID_STORAGE_MODE,
str(backup_mode),
field_description=f"battery reserve to (mode: {current_mode} -> {backup_mode})",
)
if not ok:
self.log(f"Warn: Failed to set backup mode for inverter {device_sn} (mode {current_mode} -> {backup_mode}) during startup_reset_registers")
raise RuntimeError(f"Solis startup_reset_registers failed to set backup mode for {device_sn}")
# Test write reserve SOC to 5%
ok = await self.read_and_write_cid(
device_sn,
SOLIS_CID_BATTERY_RESERVE_SOC,
"5",
field_description="Test write reserve SOC to 5%",
)
if not ok:
self.log(f"Warn: Failed to write reserve SOC for inverter {device_sn} during startup_reset_registers, attempting to restore original storage mode")
# Best-effort restore of original mode before propagating failure
try:
await self.read_and_write_cid(
device_sn,
SOLIS_CID_STORAGE_MODE,
str(current_mode),
field_description=f"restore storage mode after reserve SOC write failure (mode: {backup_mode} -> {current_mode})",
)
except Exception:
# Swallow secondary failure here; original exception is more important
self.log(f"Warn: Additional failure while restoring storage mode for inverter {device_sn} after reserve SOC write failure")
raise RuntimeError(f"Solis startup_reset_registers failed to write reserve SOC for {device_sn}")
# Clear backup bit and restore original mode
restored_mode = current_mode & ~(1 << SOLIS_BIT_BACKUP_MODE)
ok = await self.read_and_write_cid(
device_sn,
SOLIS_CID_STORAGE_MODE,
str(restored_mode),
field_description=f"battery reserve to (mode: {backup_mode} -> {restored_mode})",
)
if not ok:
self.log(f"Warn: Failed to restore storage mode for inverter {device_sn} during startup_reset_registers, attempting to reset to original mode")
try:
await self.read_and_write_cid(
device_sn,
SOLIS_CID_STORAGE_MODE,
str(current_mode),
field_description=f"restore original storage mode after failure (mode: {restored_mode} -> {current_mode})",
)
except Exception:
self.log(f"Warn: Additional failure while restoring original storage mode for inverter {device_sn}")
raise RuntimeError(f"Solis startup_reset_registers failed to restore storage mode for {device_sn}")

Copilot uses AI. Check for mistakes.
async def run(self, seconds, first):
"""Main run cycle called every 5 seconds"""
poll_success = True
Expand Down Expand Up @@ -2780,6 +2790,8 @@ async def run(self, seconds, first):
self.log(f"Solis API: Inverter {sn} is in Time of Use V2 mode")
else:
self.log(f"Solis API: Inverter {sn} is in standard Time of Use mode")
if self.control_enable:
await self.startup_reset_registers(sn) # Reset registers on startup to ensure we have write access and correct initial state

if not self.inverter_sn:
self.log("Error: Solis API: No inverters to manage after discovery")
Expand Down Expand Up @@ -2892,8 +2904,8 @@ def dashboard_item(self, entity_id, state=None, attributes=None, app=None):
print(f" Attributes: {json.dumps(attributes, indent=2)}")
self.set_state_wrapper(entity_id, state, attributes)

def get_arg(self, key, default=None):
return self.args.get(key, default)
def get_arg(self, arg, default=None, indirect=True, combine=False, attribute=None, index=None, domain=None, can_override=True, required_unit=None):
return self.args.get(arg, default)

def set_arg(self, key, value):
self.args[key] = value
Expand Down Expand Up @@ -2928,9 +2940,17 @@ async def test_solis_api(key_id, secret): # pragma: no cover
print("Calling run() once...")
await solis_api.run(seconds=0, first=True)
for device_sn, values in solis_api.cached_values.items():
await solis_api.set_storage_mode_if_needed(device_sn, "Feed-in priority")
await solis_api.set_storage_mode_if_needed(device_sn, "Self-Use")
# await solis_api.read_and_write_cid(device_sn, SOLIS_CID_BATTERY_RESERVE_SOC, "12", field_description="Test write reserve SOC to 12%")
await solis_api.read_cid(device_sn, SOLIS_CID_STORAGE_MODE) # Ensure we have the latest value for storage mode
await solis_api.read_cid(device_sn, SOLIS_CID_BATTERY_RESERVE_SOC) # Ensure we have the latest value for battery reserve SOC
pass # Just print the cached values for inspection
#await solis_api.set_storage_mode_if_needed(device_sn, "Feed-in priority")
#await solis_api.set_storage_mode_if_needed(device_sn, "Self-Use")
#current_mode = solis_api.get_current_solis_mode_value(device_sn)
#new_mode = current_mode | (1 << SOLIS_BIT_BACKUP_MODE)
#await solis_api.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(new_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")
#await solis_api.read_and_write_cid(device_sn, SOLIS_CID_BATTERY_RESERVE_SOC, "5", field_description="Test write reserve SOC to 5%")
#new_mode = current_mode & ~(1 << SOLIS_BIT_BACKUP_MODE)
#await solis_api.read_and_write_cid(device_sn, SOLIS_CID_STORAGE_MODE, str(new_mode), field_description=f"battery reserve to (mode: {current_mode} -> {new_mode})")
print("Run completed successfully")

await solis_api.final()
Expand Down
Loading