Skip to content

v.1.0.5 #61

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
TypQxQ merged 1 commit into from
Apr 27, 2025
Merged

v.1.0.5 #61

Show file tree
Hide file tree
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
2 changes: 1 addition & 1 deletion custom_components/sigen/manifest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,5 +21,5 @@
"loggers": ["custom_components.sigen"],
"quality_scale": "custom",
"requirements": ["pymodbus>=3.0.0"],
"version": "1.0.4"
"version": "1.0.5"
}
127 changes: 0 additions & 127 deletions custom_components/sigen/modbus.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -569,23 +569,6 @@ async def async_write_registers(
if register_type in [RegisterType.HOLDING, RegisterType.WRITE_ONLY]:
tried_offset = False
last_error = None
# Try offset addressing first if address >= 40001
# if address >= 40001:
# offset_address = address - 40001
# _LOGGER.debug(
# "Trying write_registers at offset address %s (orig %s) with values %s for slave %s",
# offset_address, address, values, slave_id
# )
# result = await client.write_registers(
# address=offset_address, values=values, slave=slave_id
# )
# if not result.isError():
# _LOGGER.debug("Successfully wrote to registers at offset address %s", offset_address)
# return
# _LOGGER.debug("Offset write failed, will try original address. Error: %s", result)
# last_error = result
# tried_offset = True
# Try original address if not tried yet or offset failed
_LOGGER.debug(
"Trying write_registers at original address %s with values %s for slave %s",
address, values, slave_id
Expand Down Expand Up @@ -1001,116 +984,6 @@ async def async_write_parameter(
slave_id, value, register_def.address,
register_def.register_type, register_def.data_type, register_def.gain
)

# === Plant-Specific Handling ===
# Certain plant registers require special handling due to Modbus quirks
if device_type == "plant":
key = self._get_connection_key(device_info) # Get connection key for client/lock
# plant_info is guaranteed to be a dict here

# Special handling for plant_remote_ems_enable register
# if register_name == "plant_remote_ems_enable":
# _LOGGER.debug("Special handling for plant_remote_ems_enable register")
# approaches = [
# {"function": "write_registers", "address": register_def.address,
# "values": [int(value)]},
# {"function": "write_register", "address": register_def.address,
# "value": int(value)},
# {"function": "write_registers", "address": register_def.address - 40001,
# "values": [int(value)]},
# {"function": "write_register", "address": register_def.address - 40001,
# "value": int(value)},
# ]
# last_error = None
# success = False
# try:
# async with self._locks[key]:
# client = await self._get_client(device_info)
# for approach in approaches:
# try:
# _LOGGER.debug(
# "Plant write attempt: %s", approach['description'] \
# if 'description' in approach \
# else f"{approach['function']} @ {approach['address']}")
# if approach["function"] == "write_registers":
# result = await client.write_registers(
# address=approach["address"],
# values=approach["values"],
# slave=slave_id
# )
# else:
# result = await client.write_register(
# address=approach["address"],
# value=approach["value"],
# slave=slave_id
# )
# if not result.isError():
# success = True
# break
# last_error = result
# except Exception as ex_inner:
# last_error = ex_inner
# if not success:
# raise SigenergyModbusError("Failed plant_remote_ems_enable write after "
# f"all attempts. Last error: {last_error}")
# return # Success
# except (ConnectionException, ModbusException, SigenergyModbusError) as ex_outer:
# self._connected[key] = False
# raise SigenergyModbusError(f"Error during special plant write: {ex_outer}") \
# from ex_outer
# except Exception as ex_outer:
# raise SigenergyModbusError("Unexpected error during special plant write:"
# f" {ex_outer}") from ex_outer

# Special handling for U32/S32 registers
# if register_def.data_type in [DataType.U32, DataType.S32]:
# _LOGGER.debug("Special handling for U32/S32 register %s", register_name)
# encoded_values = self._encode_value(value=value, data_type=register_def.data_type,
# gain=register_def.gain)
# _LOGGER.debug("Encoded values for %s: %s", register_name, encoded_values)
# approaches = [
# {"address": register_def.address},
# {"address": register_def.address - 40001},
# {"address": register_def.address - 40000},
# {"address": register_def.address % 10000},
# ]
# last_error = None
# success = False
# try:
# async with self._locks[key]:
# client = await self._get_client(device_info)
# for approach in approaches:
# try:
# _LOGGER.debug("Plant U32/S32 write attempt: write_registers @ %s",
# approach['address'])
# result = await client.write_registers(
# address=approach["address"],
# values=encoded_values,
# slave=slave_id
# )
# if not result.isError():
# success = True
# break
# last_error = result
# except Exception as ex_inner:
# last_error = ex_inner
# if not success:
# raise SigenergyModbusError(
# f"Failed U32/S32 write for {register_name} after all attempts. "
# f"Last error: {last_error}"
# )
# return # Success
# except (ConnectionException, ModbusException, SigenergyModbusError) as ex_outer:
# self._connected[key] = False
# raise SigenergyModbusError("Error during special plant U32/S32 write:"\
# f" {ex_outer}") from ex_outer
# except Exception as ex_outer:
# raise SigenergyModbusError("Unexpected error during special plant " \
# f"U32/S32 write: {ex_outer}") from ex_outer

# === General Write Logic ===
# (Executes if device_type is not 'plant' or if it's a plant register
# without special handling)
encoded_values = self._encode_value(
value=value,
data_type=register_def.data_type,
Expand Down
16 changes: 8 additions & 8 deletions custom_components/sigen/sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -565,14 +565,14 @@ def native_value(self) -> Any:
}.get(value, f"Unknown: {value}")
if self.entity_description.key == "ac_charger_system_state":
return {
0: "System Init",
1: "A1/A2",
2: "B1",
3: "B2",
4: "C1",
5: "C2",
6: "F",
7: "E",
0: "Initializing",
1: "Not Connected",
2: "Reserving - Charger & EV Not Ready",
3: "Preparing - Charger Ready, EV Not Ready",
4: "EV Ready, Charger Not Ready",
5: "Charging",
6: "Fault",
7: "Error",
}.get(value, f"Unknown: {value}")
if self.entity_description.key == "inverter_output_type":
return {0: "Three Phase", 1: "Single Phase"}.get(value, f"Unknown: {value}")
Expand Down
2 changes: 1 addition & 1 deletion custom_components/sigen/switch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ class SigenergySwitchEntityDescription(SwitchEntityDescription):
name="AC Charger Power",
icon="mdi:ev-station",
# identifier here will be ac_charger_name
is_on_fn=lambda data, identifier: data["ac_chargers"].get(identifier, {}).get("ac_charger_system_state") > 0,
is_on_fn=lambda data, identifier: data["ac_chargers"].get(identifier, {}).get("ac_charger_system_state") not in ("Initializing", "Fault", "Error", "Not Connected"),
turn_on_fn=lambda coordinator, identifier: coordinator.async_write_parameter("ac_charger", identifier, "ac_charger_start_stop", 0),
turn_off_fn=lambda coordinator, identifier: coordinator.async_write_parameter("ac_charger", identifier, "ac_charger_start_stop", 1),
),
Expand Down