Add support for Gen4 devices (#724)

* Add support for Gen4 devices

* Fix test

* Fix the leftovers

Author: bieniu

aioshelly/const.py

@@ -126,6 +126,7 @@
 GEN1 = 1
 GEN2 = 2
 GEN3 = 3
+GEN4 = 4
 
 
 # Firmware 1.9.0 release date
@@ -142,12 +143,16 @@
 # Firmware 1.0.99 release date
 GEN3_MIN_FIRMWARE_DATE = 20231102
 
+# Firmware 1.4.x release date
+GEN4_MIN_FIRMWARE_DATE = 20240902
+
 
 # Fallback for unknown devices
 MIN_FIRMWARE_DATES = {
     GEN1: GEN1_MIN_FIRMWARE_DATE,
     GEN2: GEN2_MIN_FIRMWARE_DATE,
     GEN3: GEN3_MIN_FIRMWARE_DATE,
+    GEN4: GEN4_MIN_FIRMWARE_DATE,
 }
 
 
@@ -952,7 +957,7 @@ class UndefinedType(Enum):
 NOTIFY_WS_CLOSED = "NotifyWebSocketClosed"
 
 BLOCK_GENERATIONS = {GEN1}
-RPC_GENERATIONS = {GEN2, GEN3}
+RPC_GENERATIONS = {GEN2, GEN3, GEN4}
 
 DEFAULT_HTTP_PORT = 80
 PERIODIC_COAP_TYPE_CODE = 30

aioshelly/rpc_device/device.py

@@ -457,7 +457,7 @@ def shelly(self) -> dict[str, Any]:
 
     @property
     def gen(self) -> int:
-        """Device generation: GEN2/3 - RPC."""
+        """Device generation: GEN2/3/4 - RPC."""
         if self._shelly is None:
             raise NotInitialized
 

tests/test_common.py

@@ -26,7 +26,8 @@
 @pytest.mark.parametrize(
     ("gen", "model", "firmware_version", "expected"),
     [
-        (4, "XYZ-G4", "20240913-112054/v1.0.0-gcb84623", False),
+        (5, "XYZ-G5", "20250913-112054/v1.0.0-gcb84623", False),
+        (4, "XYZ-G4", "20240913-112054/v1.0.0-gcb84623", True),
         (1, "SHSW-44", "20230913-112054/v1.14.0-gcb84623", False),
         (1, "SHSW-1", "20230913-112054/v1.14.0-gcb84623", True),
         (2, "SNDC-0D4P10WW", "20230703-112054/0.99.0-gcb84623", False),

tools/common/__init__.py

@@ -42,7 +42,7 @@ async def create_device(
     options: ConnectionOptions,
     gen: int | None,
 ) -> Any:
-    """Create a Gen1/Gen2/Gen3 device."""
+    """Create a device."""
     if gen is None:
         if info := await get_info(
             aiohttp_session, options.ip_address, port=options.port
@@ -170,7 +170,7 @@ def print_block_device(device: BlockDevice) -> None:
 
 
 def print_rpc_device(device: RpcDevice) -> None:
-    """Print RPC (GEN2/3) device data."""
+    """Print RPC (GEN2/3/4) device data."""
     print(f"Status: {device.status}")
     print(f"Event: {device.event}")
     print(f"Connected: {device.connected}")

tools/example.py

@@ -135,6 +135,9 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
     parser.add_argument(
         "--gen3", "-g3", action="store_true", help="Force Gen 3 (RPC) device"
     )
+    parser.add_argument(
+        "--gen4", "-g4", action="store_true", help="Force Gen 4 (RPC) device"
+    )
     parser.add_argument(
         "--debug", "-deb", action="store_true", help="Enable debug level for logging"
     )
@@ -145,7 +148,7 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
         "--update_ws",
         "-uw",
         type=str,
-        help="Update outbound WebSocket (Gen2/3) and exit",
+        help="Update outbound WebSocket (for RPC device) and exit",
     )
     parser.add_argument(
         "--listen_ip_address",
@@ -168,14 +171,14 @@ async def main() -> None:
     await coap_context.initialize(args.coap_port, args.listen_ip_address)
     await ws_context.initialize(args.ws_port, args.ws_api_url)
 
-    if not args.init and not (args.gen1 or args.gen2 or args.gen3):
+    if not args.init and not (args.gen1 or args.gen2 or args.gen3 or args.gen4):
         parser.error("specify gen if no device init at startup")
-    if args.gen1 and args.gen2:
-        parser.error("--gen1 and --gen2 can't be used together")
-    elif args.gen1 and args.gen3:
-        parser.error("--gen1 and --gen3 can't be used together")
-    elif args.gen2 and args.gen3:
-        parser.error("--gen2 and --gen3 can't be used together")
+
+    gen_list = (args.gen1, args.gen2, args.gen3, args.gen4)
+    if len([gen for gen in gen_list if gen]) > 1:
+        parser.error(
+            "You can only use one of --gen1, --gen2, --gen3 or --gen4 at a time"
+        )
 
     gen = None
     if args.gen1:
@@ -184,13 +187,15 @@ async def main() -> None:
         gen = 2
     elif args.gen3:
         gen = 3
+    elif args.gen4:
+        gen = 4
 
     if args.debug:
         logging.basicConfig(level=logging.DEBUG)
         # if gen is in args reduce logging for other gens
         if args.gen1:
             logging.getLogger("aioshelly.rpc_device").setLevel(logging.INFO)
-        elif args.gen2 or args.gen3:
+        elif args.gen2 or args.gen3 or args.gen4:
             logging.getLogger("aioshelly.block_device").setLevel(logging.INFO)
 
     if args.devices:

tools/fixture.py

@@ -287,6 +287,9 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
     parser.add_argument(
         "--gen3", "-g3", action="store_true", help="Force Gen 3 (RPC) device"
     )
+    parser.add_argument(
+        "--gen4", "-g4", action="store_true", help="Force Gen 4 (RPC) device"
+    )
     parser.add_argument(
         "--debug", "-deb", action="store_true", help="Enable debug level for logging"
     )
@@ -306,14 +309,14 @@ async def main() -> None:
     await coap_context.initialize(args.coap_port)
     await ws_context.initialize(args.ws_port, args.ws_api_url)
 
-    if not args.init and not (args.gen1 or args.gen2 or args.gen3):
+    if not args.init and not (args.gen1 or args.gen2 or args.gen3 or args.gen4):
         parser.error("specify gen if no device init at startup")
-    if args.gen1 and args.gen2:
-        parser.error("--gen1 and --gen2 can't be used together")
-    elif args.gen1 and args.gen3:
-        parser.error("--gen1 and --gen3 can't be used together")
-    elif args.gen2 and args.gen3:
-        parser.error("--gen2 and --gen3 can't be used together")
+
+    gen_list = (args.gen1, args.gen2, args.gen3, args.gen4)
+    if len([gen for gen in gen_list if gen]) > 1:
+        parser.error(
+            "You can only use one of --gen1, --gen2, --gen3 or --gen4 at a time"
+        )
 
     gen = None
     if args.gen1:
@@ -322,6 +325,8 @@ async def main() -> None:
         gen = 2
     elif args.gen3:
         gen = 3
+    elif args.gen4:
+        gen = 4
 
     if args.debug:
         logging.basicConfig(level="DEBUG", force=True)