Add support for Gen3 devices

README.md

@@ -67,7 +67,7 @@ if __name__ == "__main__":
     asyncio.run(test_block_device())
 ```
 
-### Gen2 (RPC/WebSocket) device example:
+### Gen2 and Gen3 (RPC/WebSocket) device example:
 
 ```python
 import asyncio
@@ -85,7 +85,7 @@ from aioshelly.rpc_device import RpcDevice, WsServer
 
 
 async def test_rpc_device():
-    """Test Gen2 RPC (WebSocket) based device."""
+    """Test Gen2/Gen3 RPC (WebSocket) based device."""
     options = ConnectionOptions("192.168.1.188", "username", "password")
     ws_context = WsServer()
     await ws_context.initialize(8123)

aioshelly/common.py

@@ -8,15 +8,16 @@
 from dataclasses import dataclass
 from socket import gethostbyname
 from typing import Any, Union
-from yarl import URL
 
 import aiohttp
+from yarl import URL
 
 from .const import (
     CONNECT_ERRORS,
     DEVICE_IO_TIMEOUT,
     GEN1_MIN_FIRMWARE_DATE,
     GEN2_MIN_FIRMWARE_DATE,
+    GEN3_MIN_FIRMWARE_DATE,
 )
 from .exceptions import (
     DeviceConnectionError,
@@ -119,7 +120,9 @@ def shelly_supported_firmware(result: dict[str, Any]) -> bool:
         fw_ver = GEN1_MIN_FIRMWARE_DATE
     else:
         fw_str = result["fw_id"]
-        fw_ver = GEN2_MIN_FIRMWARE_DATE
+        fw_ver = (
+            GEN2_MIN_FIRMWARE_DATE if result["gen"] == 2 else GEN3_MIN_FIRMWARE_DATE
+        )
 
     match = FIRMWARE_PATTERN.search(fw_str)
 

aioshelly/const.py

@@ -97,6 +97,10 @@
 MODEL_PRO_EM3 = "SPEM-003CEBEU"
 MODEL_PRO_EM3_400 = "SPEM-003CEBEU400"
 MODEL_WALL_DISPLAY = "SAWD-0A1XX10EU1"
+# Gen3 RPC based models
+MODEL_PLUS_1_MINI_G3 = "S3SW-001X8EU"
+MODEL_PLUS_1PM_MINI_G3 = "S3SW-001P8EU"
+MODEL_PLUS_PM_MINI_G3 = "S3PM-001PCEU16"
 
 MODEL_NAMES = {
     # Gen1 CoAP based models
@@ -175,7 +179,6 @@
     MODEL_PRO_2_V3: "Shelly Pro 2",
     MODEL_PRO_2PM: "Shelly Pro 2PM",
     MODEL_PRO_2PM_V2: "Shelly Pro 2PM",
-    MODEL_PRO_2PM_V2: "Shelly Pro 2PM",
     MODEL_PRO_3: "Shelly Pro 3",
     MODEL_PRO_4PM: "Shelly Pro 4PM",
     MODEL_PRO_4PM_V2: "Shelly Pro 4PM",
@@ -184,6 +187,10 @@
     MODEL_PRO_EM3: "Shelly Pro 3EM",
     MODEL_PRO_EM3_400: "Shelly Pro 3EM-400",
     MODEL_WALL_DISPLAY: "Shelly Wall Display",
+    # Gen3 RPC based models
+    MODEL_PLUS_1_MINI_G3: "Shelly Plus 1 Mini",
+    MODEL_PLUS_1PM_MINI_G3: "Shelly Plus 1PM Mini",
+    MODEL_PLUS_PM_MINI_G3: "Shelly Plus PM Mini",
 }
 
 # Timeout used for Device IO
@@ -198,10 +205,16 @@
 # Firmware 0.8.1 release date
 GEN2_MIN_FIRMWARE_DATE = 20210921
 
+# Firmware 1.0.99 release date
+GEN3_MIN_FIRMWARE_DATE = 20231102
+
 WS_HEARTBEAT = 55
 
 # Default Gen2 outbound websocket API URL
 WS_API_URL = "/api/shelly/ws"
 
 # Notification sent by RPC device in case of WebSocket close
 NOTIFY_WS_CLOSED = "NotifyWebSocketClosed"
+
+BLOCK_GENERATIONS = (1,)
+RPC_GENERATIONS = (2, 3)

aioshelly/rpc_device/device.py

@@ -373,8 +373,11 @@ def shelly(self) -> dict[str, Any]:
 
     @property
     def gen(self) -> int:
-        """Device generation: GEN2 - RPC."""
-        return 2
+        """Device generation: GEN2/3 - RPC."""
+        if self._shelly is None:
+            raise NotInitialized
+
+        return cast(int, self._shelly["gen"])
 
     @property
     def firmware_version(self) -> str:

example.py

@@ -18,7 +18,7 @@
 import aioshelly
 from aioshelly.block_device import BLOCK_VALUE_UNIT, COAP, BlockDevice, BlockUpdateType
 from aioshelly.common import ConnectionOptions
-from aioshelly.const import MODEL_NAMES, WS_API_URL
+from aioshelly.const import BLOCK_GENERATIONS, MODEL_NAMES, RPC_GENERATIONS, WS_API_URL
 from aioshelly.exceptions import (
     DeviceConnectionError,
     FirmwareUnsupported,
@@ -39,17 +39,17 @@ async def create_device(
     init: bool,
     gen: int | None,
 ) -> Any:
-    """Create a Gen1/Gen2 device."""
+    """Create a Gen1/Gen2/Gen3 device."""
     if gen is None:
         if info := await aioshelly.common.get_info(aiohttp_session, options.ip_address):
             gen = info.get("gen", 1)
         else:
             raise ShellyError("Unknown Gen")
 
-    if gen == 1:
+    if gen in BLOCK_GENERATIONS:
         return await BlockDevice.create(aiohttp_session, coap_context, options, init)
 
-    if gen == 2:
+    if gen in RPC_GENERATIONS:
         return await RpcDevice.create(aiohttp_session, ws_context, options, init)
 
     raise ShellyError("Unknown Gen")
@@ -164,9 +164,9 @@ def print_device(device: BlockDevice | RpcDevice) -> None:
     print(f"** {device.name} - {model_name}  @ {device.ip_address} **")
     print()
 
-    if device.gen == 1:
+    if device.gen in BLOCK_GENERATIONS:
         print_block_device(cast(BlockDevice, device))
-    elif device.gen == 2:
+    elif device.gen == RPC_GENERATIONS:
         print_rpc_device(cast(RpcDevice, device))
 
 
@@ -192,7 +192,7 @@ def print_block_device(device: BlockDevice) -> None:
 
 
 def print_rpc_device(device: RpcDevice) -> None:
-    """Print RPC (GEN2) device data."""
+    """Print RPC (GEN2/3) device data."""
     print(f"Status: {device.status}")
     print(f"Event: {device.event}")
     print(f"Connected: {device.connected}")
@@ -243,6 +243,9 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
     parser.add_argument(
         "--gen2", "-g2", action="store_true", help="Force Gen 2 (RPC) device"
     )
+    parser.add_argument(
+        "--gen3", "-g3", action="store_true", help="Force Gen 3 (RPC) device"
+    )
     parser.add_argument(
         "--debug", "-deb", action="store_true", help="Enable debug level for logging"
     )
@@ -254,7 +257,7 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
         "--update_ws",
         "-uw",
         type=str,
-        help="Update outbound WebSocket (Gen2) and exit",
+        help="Update outbound WebSocket (Gen2/3) and exit",
     )
 
     arguments = parser.parse_args()
@@ -265,7 +268,7 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
 async def update_outbound_ws(
     options: ConnectionOptions, init: bool, ws_url: str
 ) -> None:
-    """Update outbound WebSocket URL (Gen2)."""
+    """Update outbound WebSocket URL (Gen2/3)."""
     async with aiohttp.ClientSession() as aiohttp_session:
         device: RpcDevice = await create_device(aiohttp_session, options, init, 2)
         print(f"Updating outbound weboskcet URL to {ws_url}")
@@ -281,12 +284,18 @@ async def main() -> None:
 
     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 = None
     if args.gen1:
         gen = 1
     elif args.gen2:
         gen = 2
+    elif args.gen3:
+        gen = 3
 
     if args.debug:
         logging.basicConfig(level="DEBUG", force=True)