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

Open
LKuemmel wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Solaredge: use modbus bulk reader for bat #3198

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
028672c
solaredge: use modbus bulk reader for bat
LKuemmel Mar 11, 2026
71dac08
fixes
LKuemmel Mar 12, 2026
5a43adf
flake8
LKuemmel Mar 12, 2026
c1a755f
fix count of read registers
LKuemmel Apr 22, 2026
de6af51
review
LKuemmel Apr 22, 2026
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
84 changes: 48 additions & 36 deletions packages/control/process.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,44 +64,56 @@ def process_algorithm_results(self) -> None:
log.exception("Fehler im Process-Modul für Ladepunkt "+str(cp))
if data.data.bat_all_data.data.set.set_limit:
for bat_component in get_controllable_bat_components():
modules_threads.append(
Thread(
target=set_power_limit_wrapper,
args=(bat_component,
data.data.bat_data[f"bat{bat_component.component_config.id}"].data.set.power_limit),
name=f"set power limit {bat_component.component_config.id}"))
try:
modules_threads.append(
Thread(
target=set_power_limit_wrapper,
args=(bat_component,
data.data.bat_data[
f"bat{bat_component.component_config.id}"].data.set.power_limit),
name=f"set power limit {bat_component.component_config.id}"))
except Exception:
log.exception(f"Fehler im Process-Modul für Speicher {bat_component.component_config.id}")
for action in data.data.io_actions.actions.values():
if isinstance(action, DimmingDirectControl):
for d in action.config.configuration.devices:
if d["type"] == "io":
data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[d["digital_output"]] = (
action.dimming_via_direct_control()[0] is None # active output (True) if no dimming
)
if isinstance(action, DimmingIo):
for d in action.config.configuration.devices:
if d["type"] == "io":
data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[d["digital_output"]] = (
not action.dimming_active() # active output (True) if no dimming
)
if isinstance(action, StepwiseControlIo):
# check if passthrough is enabled
if action.config.configuration.passthrough_enabled:
# find output pattern by value
for pattern in action.config.configuration.output_pattern:
if pattern["value"] == action.control_stepwise()[0]:
# set digital outputs according to matching output_pattern
for output in pattern["matrix"].keys():
data.data.io_states[
f"io_states{action.config.configuration.io_device}"
].data.set.digital_output[output] = pattern["matrix"][output]
try:
if isinstance(action, DimmingDirectControl):
for d in action.config.configuration.devices:
if d["type"] == "io":
data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[
d["digital_output"]] = (
action.dimming_via_direct_control()[0] is None # active output (True) if no dimming
)
if isinstance(action, DimmingIo):
for d in action.config.configuration.devices:
if d["type"] == "io":
data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[
d["digital_output"]] = (
not action.dimming_active() # active output (True) if no dimming
)
if isinstance(action, StepwiseControlIo):
# check if passthrough is enabled
if action.config.configuration.passthrough_enabled:
# find output pattern by value
for pattern in action.config.configuration.output_pattern:
if pattern["value"] == action.control_stepwise()[0]:
# set digital outputs according to matching output_pattern
for output in pattern["matrix"].keys():
data.data.io_states[
f"io_states{action.config.configuration.io_device}"
].data.set.digital_output[output] = pattern["matrix"][output]
except Exception:
log.exception(f"Fehler im Process-Modul für IO-Aktion {action.config.id}")
for io in data.data.system_data.values():
if isinstance(io, AbstractIoDevice):
modules_threads.append(
Thread(
target=io.write,
args=(data.data.io_states[f"io_states{io.config.id}"].data.set.analog_output,
data.data.io_states[f"io_states{io.config.id}"].data.set.digital_output,),
name=f"set output io{io.config.id}"))
try:
if isinstance(io, AbstractIoDevice):
modules_threads.append(
Thread(
target=io.write,
args=(data.data.io_states[f"io_states{io.config.id}"].data.set.analog_output,
data.data.io_states[f"io_states{io.config.id}"].data.set.digital_output,),
name=f"set output io{io.config.id}"))
except Exception:
log.exception(f"Fehler im Process-Modul für IO-Gerät {io.config.id}")
if modules_threads:
joined_thread_handler(modules_threads, 3)
except Exception:
Expand Down
Loading
Loading