word_swap detection

nibe/connection/encoders.py

@@ -1,4 +1,5 @@
 from binascii import hexlify
+from typing import Optional
 
 from construct import (
     Construct,
@@ -25,8 +26,8 @@
     "s32": Int32sl,
 }
 
-parser_map_word_swaped = parser_map.copy()
-parser_map_word_swaped.update(
+parser_map_word_swapped = parser_map.copy()
+parser_map_word_swapped.update(
     {
         "u32": WordSwapped(Int32ul),
         "s32": WordSwapped(Int32sl),
@@ -37,8 +38,10 @@
 class CoilDataEncoder:
     """Encode and decode coil data."""
 
-    def __init__(self, word_swap: bool = True) -> None:
-        self._word_swap = word_swap
+    word_swap: Optional[bool] = None
+
+    def __init__(self, word_swap: Optional[bool] = None):
+        self.word_swap = word_swap
 
     def encode(self, coil_data: CoilData) -> bytes:
         """Encode coil data to bytes.
@@ -48,7 +51,7 @@ def encode(self, coil_data: CoilData) -> bytes:
             coil_data.validate()
 
             return self._pad(self._get_parser(coil_data.coil), coil_data.raw_value)
-        except (ConstructError, ValidationError) as e:
+        except (ValueError, ConstructError, ValidationError) as e:
             raise EncodeException(
                 f"Failed to encode {coil_data.coil.name} coil for value: {coil_data.value}, exception: {e}"
             )
@@ -67,7 +70,7 @@ def decode(self, coil: Coil, raw: bytes) -> CoilData:
                 return CoilData(coil, None)
 
             return CoilData.from_raw_value(coil, value)
-        except (AssertionError, ConstructError, ValidationError) as e:
+        except (ValueError, AssertionError, ConstructError, ValidationError) as e:
             raise DecodeException(
                 f"Failed to decode {coil.name} coil from raw: {hexlify(raw).decode('utf-8')}, exception: {e}"
             ) from e
@@ -89,10 +92,13 @@ def _is_hitting_integer_limit(self, coil: Coil, int_value: int):
         return False
 
     def _get_parser(self, coil: Coil) -> Construct:
-        if self._word_swap:
+        if coil.size in ["u32", "s32"] and self.word_swap is None:
+            raise ValueError("Word swap is not set, cannot parse 32 bit integers")
+
+        if self.word_swap:  # yes, it is visa versa
             return parser_map[coil.size]
         else:
-            return parser_map_word_swaped[coil.size]
+            return parser_map_word_swapped[coil.size]
 
     def _pad(self, parser: Construct, value: int) -> bytes:
         return Padded(4, parser).build(value)

nibe/connection/mixins.py

@@ -0,0 +1,33 @@
+from enum import Enum
+
+
+class ConnectionStatus(Enum):
+    """Connection status of the NibeGW connection."""
+
+    UNKNOWN = "unknown"
+    INITIALIZING = "initializing"
+    LISTENING = "listening"
+    CONNECTED = "connected"
+    DISCONNECTED = "disconnected"
+
+    def __str__(self):
+        return self.value
+
+
+class ConnectionStatusMixin:
+    CONNECTION_STATUS_EVENT = "connection_status"
+    _status: ConnectionStatus = ConnectionStatus.UNKNOWN
+
+    @property
+    def status(self) -> ConnectionStatus:
+        """Get the current connection status"""
+        return self._status
+
+    @status.setter
+    def status(self, status: ConnectionStatus):
+        if status != self._status:
+            self._status = status
+            self.notify_event_listeners(self.CONNECTION_STATUS_EVENT, status=status)
+
+    def notify_event_listeners(self, *args, **kwargs):
+        pass

nibe/connection/nibegw.py

@@ -52,6 +52,7 @@
 from nibe.coil import Coil, CoilData
 from nibe.connection import DEFAULT_TIMEOUT, READ_PRODUCT_INFO_TIMEOUT, Connection
 from nibe.connection.encoders import CoilDataEncoder
+from nibe.connection.mixins import ConnectionStatus, ConnectionStatusMixin
 from nibe.event_server import EventServer
 from nibe.exceptions import (
     AddressInUseException,
@@ -75,33 +76,18 @@
 logger = logging.getLogger("nibe").getChild(__name__)
 
 
-class ConnectionStatus(Enum):
-    """Connection status of the NibeGW connection."""
-
-    UNKNOWN = "unknown"
-    INITIALIZING = "initializing"
-    LISTENING = "listening"
-    CONNECTED = "connected"
-    DISCONNECTED = "disconnected"
-
-    def __str__(self):
-        return self.value
-
-
 @dataclass
 class CoilAction:
     coil: Coil
     future: Future
 
 
-class NibeGW(asyncio.DatagramProtocol, Connection, EventServer):
+class NibeGW(asyncio.DatagramProtocol, Connection, EventServer, ConnectionStatusMixin):
     """NibeGW connection."""
 
-    CONNECTION_STATUS_EVENT = "connection_status"
     PRODUCT_INFO_EVENT = "product_info"
     _futures: Dict[str, Future]
     _registered_reads: Dict[str, CoilAction]
-    _status: ConnectionStatus
 
     def __init__(
         self,
@@ -125,7 +111,6 @@ def __init__(
         self._remote_write_port = remote_write_port
 
         self._transport = None
-        self._status = ConnectionStatus.UNKNOWN
 
         self._send_lock = asyncio.Lock()
         self._futures = {}
@@ -148,7 +133,7 @@ def __init__(
     async def start(self):
         logger.info(f"Starting UDP server on port {self._listening_port}")
 
-        self._set_status(ConnectionStatus.INITIALIZING)
+        self.status = ConnectionStatus.INITIALIZING
 
         family, type, proto, _, sockaddr = socket.getaddrinfo(
             self._listening_ip,
@@ -182,9 +167,12 @@ async def start(self):
 
         await asyncio.get_event_loop().create_datagram_endpoint(lambda: self, sock=sock)
 
+        if self._heatpump.word_swap is None:
+            await self.detect_word_swap()
+
     def connection_made(self, transport):
         """Callback when connection is made."""
-        self._set_status(ConnectionStatus.LISTENING)
+        self.status = ConnectionStatus.LISTENING
         self._transport = transport
 
     def datagram_received(self, data: bytes, addr):
@@ -197,7 +185,7 @@ def datagram_received(self, data: bytes, addr):
                 logger.debug("Pump discovered at %s", addr)
                 self._remote_ip = addr[0]
 
-            self._set_status(ConnectionStatus.CONNECTED)
+            self.status = ConnectionStatus.CONNECTED
 
             logger.debug(msg.fields.value)
             cmd = msg.fields.value.cmd
@@ -245,6 +233,18 @@ async def read_product_info(
         finally:
             del self._futures["product_info"]
 
+    async def detect_word_swap(self, timeout: float = DEFAULT_TIMEOUT) -> None:
+        """Read word swap setting."""
+        try:
+            coil = self._heatpump.get_coil_by_address(48852)
+            assert coil.is_boolean, "Coil is not boolean"
+            coil_data = await self.read_coil(coil, timeout)
+            self.coil_encoder.word_swap = coil_data.value == "ON"
+            self._heatpump.word_swap = self.coil_encoder.word_swap
+            logger.info(f"Word swap setting detected: {coil_data.value}")
+        except Exception as e:
+            logger.warning(f"Failed to detect word swap setting: {e}")
+
     async def read_coil(self, coil: Coil, timeout: float = DEFAULT_TIMEOUT) -> CoilData:
         async with self._send_lock:
             assert self._transport, "Transport is closed"
@@ -377,21 +377,11 @@ def error_received(self, exc):
         """Handle errors from the transport"""
         logger.error(exc)
 
-    @property
-    def status(self) -> ConnectionStatus:
-        """Get the current connection status"""
-        return self._status
-
     @property
     def remote_ip(self) -> Optional[str]:
         """Get the remote IP address"""
         return self._remote_ip
 
-    def _set_status(self, status: ConnectionStatus):
-        if status != self._status:
-            self._status = status
-            self.notify_event_listeners(self.CONNECTION_STATUS_EVENT, status=status)
-
     def _on_rmu_data(self, value: Container):
         data = value.data
         self._on_coil_value(40004, data.bt1_outdoor_temperature)
@@ -465,7 +455,7 @@ async def stop(self):
             self._transport.close()
             self._transport = None
         await asyncio.sleep(0)
-        self._set_status(ConnectionStatus.DISCONNECTED)
+        self.status = ConnectionStatus.DISCONNECTED
 
 
 def xor8(data: bytes) -> int:

nibe/heatpump.py

@@ -113,7 +113,7 @@ class HeatPump(EventServer):
 
     _address_to_coil: Dict[str, Coil]
     _name_to_coil: Dict[str, Coil]
-    word_swap: bool = True
+    word_swap: Optional[bool] = None
     _product_info: Union[ProductInfo, None] = None
     _model: Optional[Model] = None
 
@@ -167,15 +167,11 @@ async def _load_coils(self):
         self._address_to_coil = {}
         for k, v in data.items():
             try:
-                self._address_to_coil[k] = self._make_coil(address=int(k), **v)
+                self._address_to_coil[k] = Coil(address=int(k), **v)
             except (AssertionError, TypeError) as e:
                 logger.warning(f"Failed to register coil {k}: {e}")
         self._name_to_coil = {c.name: c for _, c in self._address_to_coil.items()}
 
-    def _make_coil(self, address: int, **kwargs):
-        kwargs["word_swap"] = self.word_swap
-        return Coil(address, **kwargs)
-
     async def initialize(self):
         """Initialize the heat pump"""
         if not isinstance(self._model, Model) and isinstance(

tests/connection/test_modbus.py

@@ -20,6 +20,7 @@ def fixture_modbus_client():
 @pytest.fixture(name="heatpump")
 async def fixture_heatpump():
     heatpump = HeatPump(Model.S1255)
+    heatpump.word_swap = True
     await heatpump.initialize()
     yield heatpump
 

tests/connection/test_nibegw.py

@@ -17,15 +17,16 @@ async def asyncSetUp(self) -> None:
         self.loop = asyncio.get_running_loop()
 
         self.heatpump = HeatPump(Model.F1255)
+        self.heatpump.word_swap = True
         await self.heatpump.initialize()
         self.nibegw = NibeGW(self.heatpump, "127.0.0.1")
 
         self.transport = Mock()
-        assert self.nibegw.status == "unknown"
+        assert self.nibegw.status == ConnectionStatus.UNKNOWN
         self.nibegw.connection_made(self.transport)
 
     async def test_status(self):
-        assert self.nibegw.status == "listening"
+        assert self.nibegw.status == ConnectionStatus.LISTENING
 
         connection_status_handler_mock = Mock()
         self.nibegw.subscribe(
@@ -38,7 +39,7 @@ async def test_status(self):
 
         await self.nibegw.read_coil(coil)
 
-        assert self.nibegw.status == "connected"
+        assert self.nibegw.status == ConnectionStatus.CONNECTED
         connection_status_handler_mock.assert_called_once_with(
             status=ConnectionStatus.CONNECTED
         )