Add Wi‑Fi signal strength support for NEO

grobro/grobro/client.py

@@ -299,11 +299,6 @@ def __on_message(self, client, userdata, msg: MQTTMessage):
                     cfg["value"],
                 )
 
-                # Publish value back to HA (config/.../get)
-                topic = (
-                    f"{HA_BASE_TOPIC}/config/grobro/"
-                    f"{cfg['device_id']}/{cfg['register_no']}/get"
-                )
 
                 value = cfg["value"]
 
@@ -344,13 +339,13 @@ def __on_message(self, client, userdata, msg: MQTTMessage):
                         e,
                     )
 
-                self._client.publish(topic, value, retain=True)
 
                 if self.on_config_read_response:
                     self.on_config_read_response(
                         cfg["device_id"],
                         cfg["register_no"],
-                )
+                        value,
+                    )
                 return
 
             # Config WRITE response (280)

grobro/grobro/parser.py

@@ -22,7 +22,6 @@ def unscramble(decdata: bytes):
     for i, j in zip(range(0, ndecdata - 8), cycle(range(0, nmask))):
         unscrambled += bytes([decdata[i + 8] ^ int(hex_mask[j], 16)])
 
-    # hexdump(unscrambled)
     return unscrambled
 
 
@@ -115,7 +114,10 @@ def find_config_offset(data):
 
 
 def parse_config_message(data: bytes):
-    config_read_struct = struct.Struct(">4sHH16s14sH1xH2x")
+    # Config-read response layout:
+    # header, length, msg type, device id, 14B padding, config type,
+    # 1B padding, register number, value length, value, optional trailing TLVs, CRC.
+    config_read_struct = struct.Struct(">4sHH16s14sH1xHH")
 
     (
         header,
@@ -125,10 +127,22 @@ def parse_config_message(data: bytes):
         _padding,
         config_type,
         register_no,
+        value_len,
     ) = config_read_struct.unpack_from(data)
 
-    # remove trailing checksum
-    value = data[config_read_struct.size:-2].decode("ascii")
+    value_start = config_read_struct.size
+    if msg_type == 0x0118 and value_len == 0:
+        # Config-write packets omit the response value-length field; their
+        # value occupies the remaining payload before the two-byte CRC.
+        value_len = len(data) - value_start - 2
+
+    value_end = value_start + value_len
+    if value_end > len(data) - 2:
+        raise ValueError(
+            f"Config value length {value_len} exceeds packet payload ({len(data)} bytes)"
+        )
+
+    value = data[value_start:value_end].decode("ascii")
 
     return {
         "header": header,
@@ -145,7 +159,7 @@ def parse_config_ack(data: bytes):
     config_ack_struct = struct.Struct(">4sHH16s14sH")
 
     (
-     	header,
+        header,
         msg_len,
         msg_type,
         device_id,
@@ -154,7 +168,7 @@ def parse_config_ack(data: bytes):
     ) = config_ack_struct.unpack_from(data)
 
     return {
-	"header": header,
+        "header": header,
         "message_length": msg_len,
         "message_type": msg_type,
         "device_id": device_id.rstrip(b"\x00").decode("ascii"),
@@ -193,7 +207,6 @@ def parse_noah_0110(data: bytes) -> dict:
     """
     payload = data[24:]
     regs = {}
-    # register data at payload[14:] in the format: reg_lo(1B) + val_hi(1B) or reg(2B) + val(2B)
     body = payload[14:]
     pos = 0
     while pos + 4 <= len(body):
@@ -275,8 +288,7 @@ def parse_noah_fe25(data: bytes) -> dict:
     Payload: zeros except for CRC at end.
     """
     payload = data[24:]
-    # Check first 40 bytes for emptiness (last bytes may be non-payload data)
-    check_len = min(40, len(payload) - 4)  # exclude unknown trailing + CRC
+    check_len = min(40, len(payload) - 4)
     payload_body = payload[:check_len]
     return {
         "message_type": 0xFE25,

grobro/ha/client.py

@@ -932,7 +932,10 @@ def __config_read_timeout(self, device_id: str, register_no: int):
         # continue with next queued register
         self.__kickoff_next_config_read(device_id)
 
-    def handle_config_read_response(self, device_id: str, register_no: int):
+    def handle_config_read_response(self, device_id: str, register_no: int, value: str | int):
+        topic = f"{HA_BASE_TOPIC}/config/grobro/{device_id}/{register_no}/get"
+        self._client.publish(topic, value, retain=True)
+
         with self._config_read_lock:
             inflight = self._config_read_inflight.get(device_id)
             if inflight != register_no:

grobro/model/growatt_registers.py

@@ -146,6 +146,8 @@ class HomeAssistantConfigRegister(BaseModel):
     min: Optional[int] = None
     max: Optional[int] = None
     step: Optional[int] = None
+    state_class: Optional[str] = None
+    device_class: Optional[str] = None
     unit_of_measurement: Optional[str] = None
     icon: Optional[str] = None
 
@@ -162,6 +164,14 @@ class GroBroRegisters(BaseModel):
     KNOWN_NEO_REGISTERS = GroBroRegisters.model_validate(json.load(f))
 with resources.files(__package__).joinpath("growatt_noah_registers.json").open("rb") as f:
     KNOWN_NOAH_REGISTERS = GroBroRegisters.model_validate(json.load(f))
+
+# NEO dataloggers expose Wi-Fi RSSI through config register 76 as NOAH/NEXA do.
+# Keep the HA metadata identical to the established NOAH definition.
+if "wifi_signal_strength" not in KNOWN_NEO_REGISTERS.config_registers:
+    KNOWN_NEO_REGISTERS.config_registers["wifi_signal_strength"] = (
+        KNOWN_NOAH_REGISTERS.config_registers["wifi_signal_strength"].model_copy(deep=True)
+    )
+
 with resources.files(__package__).joinpath("growatt_nexa_registers.json").open("rb") as f:
     KNOWN_NEXA_REGISTERS = GroBroRegisters.model_validate(json.load(f))
 with resources.files(__package__).joinpath("growatt_spf_registers.json").open("rb") as f:

tests/test_grobro_client.py

@@ -171,11 +171,15 @@ def test_config_message_neo_341(self, client):
         client._client.on_message(None, None, msg)
         client.on_config.assert_called_once()
 
-    def test_config_read_response_281(self, client):
+    def test_config_read_response_281_routes_value_to_ha_client(self, client):
         data = (Path(DATA_DIR) / "NeoConfigReadResponse_337.bin").read_bytes()
         msg = _msg("c/33/QMN000ABC1D2E3FG", data)
         client._client.on_message(None, None, msg)
-        client._client.publish.assert_called()  # publishes back to HA topic
+
+        client._client.publish.assert_not_called()
+        client.on_config_read_response.assert_called_once_with(
+            "QMN000ABC1D2E3FG", 4, 5
+        )
 
     def test_config_write_ack_280(self, client):
         data = (Path(DATA_DIR) / "NeoConfigWriteAck_DataInterval.bin").read_bytes()

tests/test_ha_client.py

@@ -475,6 +475,20 @@ def test_discovery_publishes(self, ha_client):
         ha_client._Client__publish_device_discovery("QMN000ABC1D2E3FG")
         assert ha_client._client.publish.called
 
+    def test_neo_wifi_signal_discovery_has_numeric_metadata(self, ha_client):
+        device_id = "QMN000ABC1D2E3FG"
+        ha_client._Client__publish_device_discovery(device_id)
+
+        payload = next(
+            json.loads(call.args[1])
+            for call in ha_client._client.publish.call_args_list
+            if call.args[0] == f"homeassistant/device/{device_id}/config" and call.args[1]
+        )
+
+        component = payload["cmps"][f"grobro_{device_id}_cmd_wifi_signal_strength"]
+        assert component["device_class"] == "signal_strength"
+        assert component["state_class"] == "measurement"
+
     def test_discovery_skip_unchanged_payload(self, ha_client):
         ha_client._Client__publish_device_discovery("QMN000ABC1D2E3FG")
         first_calls = len(ha_client._client.publish.call_args_list)
@@ -673,13 +687,18 @@ def test_kickoff_skips_if_inflight(self):
                 c._Client__kickoff_next_config_read("QMN000ABC1D2E3FG")
                 c.on_config_read.assert_not_called()
 
-    def test_handle_config_read_response(self):
+    def test_handle_config_read_response_publishes_to_target_broker(self):
         with patch("grobro.ha.client.mqtt.Client"):
             with patch("grobro.ha.client.os.listdir", return_value=[]):
                 c = Client(MQTTConfig(host="localhost", port=1883))
                 c._config_read_inflight["QMN000ABC1D2E3FG"] = 1280
                 with patch("grobro.ha.client.Timer"):
-                    c.handle_config_read_response("QMN000ABC1D2E3FG", 1280)
+                    c.handle_config_read_response("QMN000ABC1D2E3FG", 1280, -55)
+                c._client.publish.assert_called_once_with(
+                    "homeassistant/config/grobro/QMN000ABC1D2E3FG/1280/get",
+                    -55,
+                    retain=True,
+                )
                 assert "QMN000ABC1D2E3FG" not in c._config_read_inflight
 
     def test_handle_config_read_response_wrong_register(self):
@@ -688,7 +707,7 @@ def test_handle_config_read_response_wrong_register(self):
                 c = Client(MQTTConfig(host="localhost", port=1883))
                 c._config_read_inflight["QMN000ABC1D2E3FG"] = 1280
                 with patch("grobro.ha.client.Timer"):
-                    c.handle_config_read_response("QMN000ABC1D2E3FG", 999)
+                    c.handle_config_read_response("QMN000ABC1D2E3FG", 999, -55)
                 assert c._config_read_inflight["QMN000ABC1D2E3FG"] == 1280
 
     def test_config_read_timeout(self):

tests/test_parser.py

@@ -27,7 +27,6 @@ def test_parse_noah_6f64():
     assert "2026-05-15T17:12:09.001" in result["timestamp"]
     assert json.loads(result["data"])["t_act"] == 150
 
-    # Should also work via dispatch
     dispatched = parser.parse_noah_message(bytes(data))
     assert dispatched is not None
     assert dispatched["message_type"] == 0x6F64
@@ -62,3 +61,22 @@ def test_parse_config_type_no_params():
     data = b"\xff\xff\x00\x00\x00\xff"
     config = parser.parse_config_type(data, 0)
     assert config.model_dump().get("raw") is not None
+
+
+def test_parse_neo_config_read_uses_declared_value_length():
+    # Real NEO response shape observed for register 76. The Wi-Fi RSSI value is
+    # '-055' and additional bytes follow before the CRC; they must not be
+    # included in the decoded value.
+    data = bytes.fromhex(
+        "00 01 00 07 00 30 01 19 "
+        "51 4d 4e 30 30 30 41 42 43 31 44 32 45 33 46 47 "
+        "00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
+        "00 02 00 00 4c 00 04 2d 30 35 35 00 05 00 01 31 ca f3"
+    )
+
+    result = parser.parse_config_message(data)
+
+    assert result["message_type"] == 0x0119
+    assert result["device_id"] == "QMN000ABC1D2E3FG"
+    assert result["register_no"] == 76
+    assert result["value"] == "-055"