Add spa state

pybalboa/__main__.py

@@ -1,30 +1,44 @@
 """Main entry."""
 
+import argparse
 import asyncio
 import logging
 import sys
 from enum import IntEnum
+from typing import Union
 
 try:
     from . import SpaClient, SpaConnectionError, SpaControl, __version__
+    from .enums import SpaState
 except ImportError:
     from pybalboa import SpaClient, SpaConnectionError, SpaControl, __version__
+    from pybalboa.enums import SpaState
 
 
-def usage() -> None:
-    """Print uage instructions."""
-    print(f"Usage: {sys.argv[0]} <ip/host> <flag>")
-    print("\tip/host:\tip address of spa (required)")
-    print("\t-d, --debug:\tenable debug logs (optional)")
+async def run_discovery(first_spa: bool = True) -> None:
+    """Attempt to discover a spa and try some commands."""
+    spas = await SpaClient.discover(first_spa)
+    for spa in spas:
+        await connect_and_listen(spa=spa)
 
 
-async def connect_and_listen(host: str) -> None:
+async def connect_and_listen(
+    host: Union[str, None] = None, spa: Union[SpaClient, None] = None
+) -> None:
     """Connect to the spa and try some commands."""
     print("******** Testing spa connection and configuration **********")
     try:
-        async with SpaClient(host) as spa:
+        if host:
+            spa = SpaClient(host)
+        if not spa:
+            print("No spa provided")
+            return
+        async with spa:
             if not await spa.async_configuration_loaded():
-                print("Config not loaded, something is wrong!")
+                if spa.state == SpaState.TEST_MODE:
+                    print("Config not loaded, spa is in test mode!")
+                else:
+                    print("Config not loaded, something is wrong!")
                 return
 
             print()
@@ -176,14 +190,31 @@ async def _state_check() -> None:
 
 
 if __name__ == "__main__":
-    if (args := len(sys.argv)) < 2:
-        usage()
-        sys.exit(1)
+    parser = argparse.ArgumentParser(
+        description="Connect to a spa and listen for updates.",
+        usage=f"{sys.argv[0]} [host] [-d | --debug] [--all]",
+    )
+    parser.add_argument("host", nargs="?", help="Spa IP address or hostname (optional)")
+    parser.add_argument(
+        "-d", "--debug", action="store_true", help="Enable debug logging"
+    )
+    parser.add_argument(
+        "--all",
+        action="store_true",
+        help="Discover all available spas instead of just the first one.",
+    )
 
-    if args > 2 and sys.argv[2] in ("-d", "--debug"):
+    args = parser.parse_args()
+
+    if args.debug:
         logging.basicConfig(level=logging.DEBUG)
 
     print(f"pybalboa version: {__version__}")
-    asyncio.run(connect_and_listen(sys.argv[1]))
+
+    if args.host:
+        asyncio.run(connect_and_listen(args.host))
+    else:
+        print("No host provided. Running in discovery mode...")
+        asyncio.run(run_discovery(first_spa=not args.all))
 
     sys.exit(0)

pybalboa/client.py

@@ -17,6 +17,7 @@
     LowHighRange,
     MessageType,
     SettingsCode,
+    SpaState,
     TemperatureUnit,
     ToggleItemCode,
     WiFiState,
@@ -109,14 +110,15 @@ def __init__(
         self._filter_cycle_2_end: time | None = None
 
         # status update
-        self._time_offset: timedelta = timedelta(0)
         self._accessibility_type: AccessibilityType = AccessibilityType.NONE
         self._filter_cycle_1_running: bool = False
         self._filter_cycle_2_running: bool = False
         self._heat_state: HeatState = HeatState.OFF
         self._is_24_hour: bool = False
+        self._state: SpaState = SpaState.UNKNOWN
         self._time_hour: int = 0
         self._time_minute: int = 0
+        self._time_offset: timedelta = timedelta(0)
         self._temperature_unit: TemperatureUnit = TemperatureUnit.FAHRENHEIT
         self._temperature: float | None = None
         self._target_temperature: float | None = None
@@ -250,6 +252,11 @@ def software_version(self) -> str | None:
         """Return the software version."""
         return self._software_version
 
+    @property
+    def state(self) -> SpaState:
+        """Return the spa state."""
+        return self._state
+
     @property
     def temperature_unit(self) -> TemperatureUnit:
         """Return the temperatre unit."""
@@ -643,7 +650,8 @@ def _parse_status_update(self, data: bytes, reprocess: bool = False) -> None:
 
         Byte  | Data
         ---------------------------
-        00-01 | ? ?
+        00    | spa state ?
+        01    | initialization mode ?
         02    | current temperature
         03    | current hour
         04    | current minute
@@ -667,6 +675,7 @@ def _parse_status_update(self, data: bytes, reprocess: bool = False) -> None:
             return
 
         self._previous_status = data
+        self._state = SpaState(data[0])
         self._time_hour = data[3]
         self._time_minute = data[4]
         if not reprocess:

pybalboa/enums.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import logging
 from enum import Enum, IntEnum
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class MessageType(IntEnum):
     """Message type."""
@@ -78,6 +81,24 @@ class HeatState(IntEnum):
     HEAT_WAITING = 2
 
 
+class SpaState(IntEnum):
+    """Spa state."""
+
+    RUNNING = 0x00
+    INITIALIZING = 0x01
+    HOLD_MODE = 0x05
+    AB_TEMPS_ON = 0x14
+    TEST_MODE = 0x17
+
+    UNKNOWN = -1
+
+    @classmethod
+    def _missing_(cls, value: object) -> SpaState:
+        """Handle unknown values by returning UNKNOWN instead of raising an error."""
+        _LOGGER.warning("Received unknown value %s for %s", value, cls.__name__)
+        return cls.UNKNOWN
+
+
 class TemperatureUnit(IntEnum):
     """Tempeature unit."""