Synchronize confirmed NEO writes with Home Assistant (output power limit sometimes "unknown")

README.md

@@ -21,6 +21,8 @@ Join us at [#grobro:matrix.org](https://matrix.to/#/#grobro:matrix.org)
 - Proxies messages to the Growatt Cloud to keep the ShinePhone app functional (optional)
 - Enables a local-only setup - keeping your device off the cloud
 - Supports Home Assistant MQTT auto-discovery
+- Publishes all holding-register state topics with MQTT retention so Home Assistant restores the latest confirmed control values after restarts
+- Handles the observed NEO output-power-limit encoding where raw response value `1` represents `100%`
 - Containerized and configurable via environment variables
 
 ---

grobro/grobro/client.py

@@ -49,6 +49,12 @@ def _extract_device_id(topic: str) -> str:
     """
     return _DEVICE_ID_RE.sub("", topic.split("/")[-1])
 
+def _normalize_neo_output_power_limit(value: float | int) -> float | int:
+    # NEO encodes its 100% output-power limit as raw value 1 in read responses.
+    return 100 if value == 1 else value
+
+
+
 
 LOG = logging.getLogger(__name__)
 HA_BASE_TOPIC = os.getenv("HA_BASE_TOPIC", "homeassistant")
@@ -381,9 +387,33 @@ def __on_message(self, client, userdata, msg: MQTTMessage):
                         self.on_config(device_id, config)
                         return
 
+            # NEO preset-single responses carry the confirmed register value after
+            # a one-byte status field, unlike standard Modbus register blocks.
+            preset_response = GrowattModbusFunctionSingle.parse_response_grobro(unscrambled)
+            if preset_response and device_id.startswith("QMN"):
+                state = HomeAssistantHoldingRegisterInput(device_id=device_id)
+                for name, register in KNOWN_NEO_REGISTERS.holding_registers.items():
+                    if register.growatt.position.register_no != preset_response.register_no:
+                        continue
+                    value = register.growatt.data.parse(preset_response.value.to_bytes(2, "big"))
+                    if name == "output_power_limit":
+                        value = _normalize_neo_output_power_limit(value)
+                    if value is not None:
+                        state.payload.append(
+                            HomeAssistantHoldingRegisterValue(
+                                name=name,
+                                value=value,
+                                register=register.homeassistant,
+                            )
+                        )
+                    break
+                if state.payload:
+                    self.on_holding_register_input(state)
+                return
+
             # Generic modbus message
             modbus_message = GrowattModbusMessage.parse_grobro(unscrambled)
-            LOG.debug("Received modbus message: %s", modbus_message)
+            LOG.debug("Received Modbus response: %s", modbus_message)
 
             if modbus_message:
                 known_registers = None
@@ -422,6 +452,8 @@ def __on_message(self, client, userdata, msg: MQTTMessage):
                         value = register.growatt.data.parse(data_raw)
                         if value is None:
                             continue
+                        if modbus_device_id.startswith("QMN") and name == "output_power_limit":
+                            value = _normalize_neo_output_power_limit(value)
                         if register.homeassistant.type=="switch":
                             value = "ON" if value==1 else "OFF"
                         state.payload.append(

grobro/ha/client.py

@@ -368,7 +368,7 @@ def publish_holding_register_input(self, ha_input: HomeAssistantHoldingRegisterI
             LOG.debug("HA: publish: %s", ha_input)
             for value in ha_input.payload:
                 topic = f"{HA_BASE_TOPIC}/{value.register_def.type}/grobro/{ha_input.device_id}/{value.name}/get"
-                self._client.publish(topic, value.value, retain=PUBLISH_SENSORS_RETAINED)
+                self._client.publish(topic, value.value, retain=True)
         except Exception as e:
             LOG.error(f"HA: publish msg: {e}")
 

grobro/model/modbus_function.py

@@ -12,9 +12,9 @@ class GrowattModbusFunctionMultiple(BaseModel):
     to read or write multiple registers.
 
     Structure:
-        - H - 2 byte unknown
-        - H - 2 byte constant 7
-        - H - 2 byte message length (excluding register count, constant and message length)
+        - H - 2 byte transaction ID (echoed in responses)
+        - H - 2 byte Growatt protocol ID (constant: 7)
+        - H - 2 byte frame length from offset 8 (includes trailing CRC)
         - B - 1 byte modbus device address (seems to be constant 1 in mqtt)
         - B - 1 byte function
         - 30s - 30 byte zero-padded device id
@@ -74,9 +74,9 @@ class GrowattModbusFunctionSingle(BaseModel):
     to read or write single registers.
 
     Structure:
-        - H - 2 byte unknown
-        - H - 2 byte constant 7
-        - H - 2 byte message length (excluding register count, constant and message length)
+        - H - 2 byte transaction ID (echoed in responses)
+        - H - 2 byte Growatt protocol ID (observed: 7)
+        - H - 2 byte frame length from offset 8 (includes trailing CRC)
         - B - 1 byte modbus device address (seems to be constant 1 in mqtt)
         - B - 1 byte function
         - 30s - 30 byte zero-padded device id
@@ -111,6 +111,36 @@ def parse_grobro(buffer) -> Optional["GrowattModbusFunctionSingle"]:
             value=value,
         )
 
+    @staticmethod
+    def parse_response_grobro(buffer) -> Optional["GrowattModbusFunctionSingle"]:
+        if len(buffer) < 45:
+            return None
+
+        (
+            _unknown,
+            _constant_7,
+            message_length,
+            _device_address,
+            function,
+            device_id_raw,
+            register,
+            status,
+            value,
+        ) = struct.unpack(">HHHBB30sHBH", buffer[:43])
+        if (
+            message_length != len(buffer[8:])
+            or function != GrowattModbusFunction.PRESET_SINGLE_REGISTER
+            or status != 0
+        ):
+            return None
+
+        return GrowattModbusFunctionSingle(
+            device_id=device_id_raw.decode("ascii", errors="ignore").strip("\x00"),
+            function=function,
+            register_no=register,
+            value=value,
+        )
+
     def build_grobro(self) -> bytes:
         return struct.pack(
             MODBUS_COMMAND_STRUCT,

grobro/model/modbus_message.py

@@ -110,17 +110,17 @@ class GrowattModbusMessage(BaseModel):
     Represents a block of modbus registers sent by the growatt device.
 
     Header Structure:
-        - H - 2 byte unknown
-        - H - 2 byte constant 7
-        - H - 2 byte message length (excluding register count, constant and message length)
+        - H - 2 byte transaction ID (echoed in responses)
+        - H - 2 byte Growatt protocol ID (observed: 7)
+        - H - 2 byte frame length from offset 8 (includes trailing CRC)
         - B - 1 byte modbus device address (seems to be constant 1 in mqtt)
         - B - 1 byte function
         - 30s - 30 byte zero-padded device id
         - optional GrowattModbusMetadata - only present when function == READ_INPUT_REGISTER
         - N register blocks
     """
 
-    unknown: int
+    transaction_id: int
     device_id: str
     metadata: Optional[GrowattMetadata] = None
     function: GrowattModbusFunction
@@ -146,7 +146,7 @@ def get_data(self, pos: GrowattRegisterPosition):
     @staticmethod
     def parse_grobro(buffer) -> Optional["GrowattModbusMessage"]:
         try:
-            (unknown, constant_7, msg_len, constant_1, function, device_id_raw) = (
+            (transaction_id, constant_7, msg_len, constant_1, function, device_id_raw) = (
                 struct.unpack(
                     HEADER_STRUCT,
                     buffer[0:38],
@@ -173,7 +173,7 @@ def parse_grobro(buffer) -> Optional["GrowattModbusMessage"]:
                 offset += block.size()
 
             return GrowattModbusMessage(
-                unknown=unknown,
+                transaction_id=transaction_id,
                 metadata=metadata,
                 device_id=device_id,
                 function=function,
@@ -185,7 +185,7 @@ def parse_grobro(buffer) -> Optional["GrowattModbusMessage"]:
     def build_grobro(self) -> bytes:
         result = struct.pack(
             HEADER_STRUCT,
-            self.unknown,
+            self.transaction_id,
             7,
             self.msg_len,
             1,

tests/model/test_models.py

@@ -171,6 +171,68 @@ def test_parse_function_3_roundtrip(self):
         rebuilt = parsed.build_grobro()
         assert rebuilt == pkt
 
+    def test_parse_preset_single_response(self):
+        packet = struct.pack(
+            ">HHHBB30sHBH",
+            1,
+            7,
+            37,
+            1,
+            GrowattModbusFunction.PRESET_SINGLE_REGISTER,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            3,
+            0,
+            74,
+        ) + b"\x00\x00"
+
+        response = GrowattModbusFunctionSingle.parse_response_grobro(packet)
+
+        assert response is not None
+        assert response.device_id == "QMN000ABC1D2E3FG"
+        assert response.function == GrowattModbusFunction.PRESET_SINGLE_REGISTER
+        assert response.register_no == 3
+        assert response.value == 74
+
+    def test_parse_preset_single_response_rejects_error_status(self):
+        packet = struct.pack(
+            ">HHHBB30sHBH",
+            1,
+            7,
+            37,
+            1,
+            GrowattModbusFunction.PRESET_SINGLE_REGISTER,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            3,
+            1,
+            74,
+        ) + b"\x00\x00"
+
+        assert GrowattModbusFunctionSingle.parse_response_grobro(packet) is None
+
+    @pytest.mark.parametrize(
+        ("message_length", "truncate"),
+        [(36, False), (37, True)],
+    )
+    def test_parse_preset_single_response_rejects_invalid_length(
+        self, message_length, truncate
+    ):
+        packet = struct.pack(
+            ">HHHBB30sHBH",
+            1,
+            7,
+            message_length,
+            1,
+            GrowattModbusFunction.PRESET_SINGLE_REGISTER,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            3,
+            0,
+            74,
+        ) + b"\x00\x00"
+
+        if truncate:
+            packet = packet[:-1]
+
+        assert GrowattModbusFunctionSingle.parse_response_grobro(packet) is None
 
 class TestModbusBlock:
     def test_parse_error(self):
@@ -192,7 +254,7 @@ def test_parse_and_build():
 class TestModbusMessage:
     def test_get_data_match(self):
         msg = GrowattModbusMessage(
-            unknown=0,
+            transaction_id=0,
             device_id="TEST",
             function=GrowattModbusFunction.READ_INPUT_REGISTER,
             register_blocks=[
@@ -209,7 +271,7 @@ def test_get_data_match(self):
 
     def test_get_data_no_match(self):
         msg = GrowattModbusMessage(
-            unknown=0,
+            transaction_id=0,
             device_id="TEST",
             function=GrowattModbusFunction.READ_INPUT_REGISTER,
             register_blocks=[

tests/test_grobro_client.py

@@ -1,11 +1,14 @@
 import os
+import struct
 import tempfile
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 from paho.mqtt.client import MQTTMessage
 
+from grobro.grobro.builder import scramble
+
 from grobro.grobro.client import Client, get_property, dump_message_binary
 import grobro.grobro.client as grobro_client
 from grobro.model.mqtt_config import MQTTConfig
@@ -195,11 +198,79 @@ def test_modbus_single_register_neo(self, client):
         client._client.on_message(None, None, msg)
         client.on_holding_register_input.assert_called_once()
 
-    def test_modbus_prese_single_noah(self, client):
+    def test_modbus_single_register_neo_normalizes_full_power_limit(self, client):
+        response = struct.pack(
+            ">HHHBB30sHHH",
+            1,
+            7,
+            38,
+            1,
+            5,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            3,
+            3,
+            1,
+        ) + b"\x00\x00"
+        msg = _msg("c/33/QMN000ABC1D2E3FG", scramble(response))
+        client._client.on_message(None, None, msg)
+
+        state = client.on_holding_register_input.call_args.args[0]
+        assert state.payload[0].name == "output_power_limit"
+        assert state.payload[0].value == 100
+
+    def test_modbus_preset_single_neo_publishes_confirmed_value(self, client):
+        response = struct.pack(
+            ">HHHBB30sHBH",
+            1,
+            7,
+            37,
+            1,
+            6,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            3,
+            0,
+            74,
+        ) + b"\x00\x00"
+        msg = _msg("c/33/QMN000ABC1D2E3FG", scramble(response))
+
+        client._client.on_message(None, None, msg)
+
+        state = client.on_holding_register_input.call_args.args[0]
+        assert state.payload[0].name == "output_power_limit"
+        assert state.payload[0].value == 74
+
+    def test_modbus_preset_single_neo_ignores_unknown_register(self, client):
+        response = struct.pack(
+            ">HHHBB30sHBH",
+            1,
+            7,
+            37,
+            1,
+            6,
+            b"QMN000ABC1D2E3FG".ljust(30, b"\x00"),
+            65535,
+            0,
+            74,
+        ) + b"\x00\x00"
+        msg = _msg("c/33/QMN000ABC1D2E3FG", scramble(response))
+
+        client._client.on_message(None, None, msg)
+
+        client.on_holding_register_input.assert_not_called()
+
+    def test_modbus_preset_single_noah_uses_generic_parser(self, client):
         data = (Path(DATA_DIR) / "NoahPresetSingle_OutputLimit.bin").read_bytes()
         msg = _msg("c/33/0PVP0000TEST0001", data)
-        client._client.on_message(None, None, msg)
-        # PRESET_SINGLE_REGISTER response is not routed to handlers
+
+        with patch.object(
+            grobro_client.GrowattModbusMessage,
+            "parse_grobro",
+            wraps=grobro_client.GrowattModbusMessage.parse_grobro,
+        ) as parse_grobro:
+            client._client.on_message(None, None, msg)
+
+        parse_grobro.assert_called_once_with(grobro_client.parser.unscramble(data))
+        client.on_holding_register_input.assert_not_called()
 
     def test_modbus_input_noah(self, client):
         data = (Path(DATA_DIR) / "NoahReadInputRegisters_0-124.bin").read_bytes()

tests/test_ha_client.py

@@ -344,6 +344,7 @@ def test_publish_holding_register_input(self, ha_client):
         topic = ha_client._client.publish.call_args[0][0]
         assert "output_power_limit" in topic
         assert "get" in topic
+        assert ha_client._client.publish.call_args.kwargs["retain"] is True
 
     def test_publish_holding_register_error(self, caplog, ha_client):
         ha_client._client.publish.side_effect = Exception("publish error")