feat: add support for Outdoor Plug EU (#604)

* feat: add support for Outdoor Plug EU

* fix: keep numeric Outdoor Plug EU cloud model only

switchbot/__init__.py

@@ -69,6 +69,7 @@ from .devices.light_strip import (
 )
 from .devices.lock import SwitchbotLock, SwitchbotLockUltra
 from .devices.meter_pro import SwitchbotMeterPro, SwitchbotMeterProCO2
+from .devices.outdoor_plug import SwitchbotOutdoorPlug
 from .devices.plug import SwitchbotPlugMini
 from .devices.relay_switch import (
     SwitchbotGarageDoorOpener,
@@ -141,6 +142,7 @@ __all__ = [
     "SwitchbotMeterProCO2",
     "SwitchbotModel",
     "SwitchbotOperationError",
+    "SwitchbotOutdoorPlug",
     "SwitchbotPermanentOutdoorLight",
     "SwitchbotPlugMini",
     "SwitchbotRelaySwitch",

switchbot/adv_parser.py

@@ -49,7 +49,7 @@ from .adv_parsers.lock import (
 )
 from .adv_parsers.meter import process_wosensorth, process_wosensorth_c
 from .adv_parsers.motion import process_wopresence
-from .adv_parsers.plug import process_woplugmini
+from .adv_parsers.plug import process_outdoor_plug, process_woplugmini
 from .adv_parsers.presence_sensor import process_presence_sensor
 from .adv_parsers.relay_switch import (
     process_garage_door_opener,
@@ -960,6 +960,12 @@ SUPPORTED_TYPES: dict[str | bytes, SwitchbotSupportedType] = {
         "func": process_relay_switch_1pm,
         "manufacturer_id": 2409,
     },
+    b"\x00\x12\x18\xd0": {
+        "modelName": SwitchbotModel.OUTDOOR_PLUG_EU,
+        "modelFriendlyName": "Outdoor Plug (EU)",
+        "func": process_outdoor_plug,
+        "manufacturer_id": 2409,
+    },
     b"\x00\x11\x07\x60": {
         "modelName": SwitchbotModel.STANDING_FAN,
         "modelFriendlyName": "Standing Fan",

switchbot/adv_parsers/plug.py

@@ -17,3 +17,18 @@ def process_woplugmini(
         "wifi_rssi": -mfr_data[9],
         "power": parse_power_data(mfr_data, 10, 10.0, 0x7FFF),  # W
     }
+
+
+def process_outdoor_plug(
+    data: bytes | None, mfr_data: bytes | None
+) -> dict[str, bool | int | float]:
+    """Process Outdoor Plug EU consumption and production power."""
+    if mfr_data is None or len(mfr_data) < 14:
+        return {}
+    return {
+        "switchMode": True,
+        "sequence_number": mfr_data[6],
+        "isOn": bool(mfr_data[7] & 0x80),
+        "power": parse_power_data(mfr_data, 10),
+        "power_production": parse_power_data(mfr_data, 12),
+    }

switchbot/const/__init__.py

@@ -121,6 +121,7 @@ class SwitchbotModel(StrEnum):
     CANDLE_WARMER_LAMP = "Candle Warmer Lamp"
     PLUG_MINI_EU = "Plug Mini (EU)"
     PLUG_MINI_IT = "Plug Mini (IT)"
+    OUTDOOR_PLUG_EU = "Outdoor Plug (EU)"
     RGBICWW_STRIP_LIGHT = "RGBICWW Strip Light"
     RGBICWW_FLOOR_LAMP = "RGBICWW Floor Lamp"
     RGBICWW_LIGHT_BARS = "RGBICWW Light Bars"

switchbot/devices/device.py

@@ -121,6 +121,7 @@ API_MODEL_TO_ENUM: dict[str, SwitchbotModel] = {
     "W1104000": SwitchbotModel.PLUG_MINI_EU,
     "W1197000": SwitchbotModel.PLUG_MINI_IT,
     "W1213000": SwitchbotModel.HUB4,
+    "W1186000": SwitchbotModel.OUTDOOR_PLUG_EU,
     "W1128000": SwitchbotModel.SMART_THERMOSTAT_RADIATOR,
     "W1111000": SwitchbotModel.CLIMATE_PANEL,
     "W1130000": SwitchbotModel.ART_FRAME,

switchbot/devices/outdoor_plug.py

@@ -0,0 +1,43 @@
+from typing import Any
+
+from ..const import SwitchbotModel
+from ..models import SwitchBotAdvertisement
+from .relay_switch import COMMAND_GET_CHANNEL2_INFO, SwitchbotRelaySwitch
+
+
+class SwitchbotOutdoorPlug(SwitchbotRelaySwitch):
+    """An Outdoor Plug EU with bidirectional metering."""
+
+    _model = SwitchbotModel.OUTDOOR_PLUG_EU
+
+    def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
+        """Preserve polled metering values alongside advertised power."""
+        data = advertisement.data["data"]
+        for key in (
+            "voltage",
+            "current",
+            "energy",
+            "current_production",
+            "energy_production",
+        ):
+            data[key] = self._get_adv_value(key) or 0
+        super().update_from_advertisement(advertisement)
+
+    async def get_basic_info(self) -> dict[str, Any] | None:
+        """Read consumption and production for the single relay."""
+        if not (data := await super().get_basic_info()):
+            return None
+        now, day_start = self.get_current_time_and_start_time()
+        raw = await self._get_basic_info(
+            COMMAND_GET_CHANNEL2_INFO.format(now, day_start)
+        )
+        if not raw or len(raw) < 15:
+            return None
+        production = self._parse_user_data(raw)
+        if not data["isOn"]:
+            self._reset_power_data(production)
+        return data | {
+            "power_production": production["power"],
+            "current_production": production["current"],
+            "energy_production": production["energy"],
+        }

tests/test_outdoor_plug_eu.py

@@ -0,0 +1,130 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from switchbot import (
+    SwitchBotAdvertisement,
+    SwitchbotModel,
+    SwitchbotOutdoorPlug,
+    parse_advertisement_data,
+)
+from switchbot.adv_parsers.plug import process_outdoor_plug
+from switchbot.devices.device import API_MODEL_TO_ENUM
+
+from .test_adv_parser import generate_advertisement_data, generate_ble_device
+
+
+@pytest.mark.parametrize("network", [0, 0x15, 0x75])
+@pytest.mark.parametrize(("status", "is_on"), [(0x01, False), (0x87, True)])
+@pytest.mark.parametrize("encrypted", [False, True])
+def test_outdoor_plug_eu_advertisement(network, status, is_on, encrypted):
+    service = bytes([0x80 if encrypted else 0, 0, 0, 0, 0x12, 0x18, 0xD0, network])
+    manufacturer = (
+        bytes.fromhex("010203040506")
+        + bytes([99, status, 0, 0])
+        + bytes.fromhex("04e6007b0000")
+    )
+    adv = parse_advertisement_data(
+        generate_ble_device("01:02:03:04:05:06", "Outdoor Plug (EU)"),
+        generate_advertisement_data(
+            manufacturer_data={2409: manufacturer},
+            service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": service},
+        ),
+    )
+    assert adv is not None
+    assert adv.data["modelName"] == SwitchbotModel.OUTDOOR_PLUG_EU
+    assert adv.data["modelFriendlyName"] == "Outdoor Plug (EU)"
+    assert adv.data["isEncrypted"] is encrypted
+    assert adv.data["data"] == {
+        "switchMode": True,
+        "sequence_number": 99,
+        "isOn": is_on,
+        "power": 1254.0,
+        "power_production": 123.0,
+    }
+
+
+def create_device():
+    return SwitchbotOutdoorPlug(
+        generate_ble_device("01:02:03:04:05:06", "Outdoor Plug"),
+        "ff",
+        "ffffffffffffffffffffffffffffffff",
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("is_on", [False, True])
+async def test_bidirectional_metering(is_on):
+    device = create_device()
+    device.get_current_time_and_start_time = lambda: ("00000010", "00000000")
+    common = bytes([1, 99, 0x81 if is_on else 1]) + bytes(13) + bytes([3])
+    consumed = bytes.fromhex("0100ea60003a98003c08fc03e808fc")
+    produced = bytes.fromhex("0100bb80007530001e08fc07d011f8")
+    device._get_basic_info = AsyncMock(side_effect=[common, consumed, produced])
+    info = await device.get_basic_info()
+    assert info["isOn"] is is_on
+    assert info["energy"] == 1.0
+    assert info["energy_production"] == 0.8
+    assert info["power"] == (230.0 if is_on else 0)
+    assert info["power_production"] == (460.0 if is_on else 0)
+    assert info["current"] == (1.0 if is_on else 0)
+    assert info["current_production"] == (2.0 if is_on else 0)
+    assert info["voltage"] == (230 if is_on else 0)
+    assert device._get_basic_info.call_args_list[1].args == (
+        "570f7106000000001000000000",
+    )
+    assert device._get_basic_info.call_args_list[2].args == (
+        "570f7106010000001000000000",
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("reply", [None, b"", b"\x01", bytes(14)])
+async def test_short_production_response(reply):
+    device = create_device()
+    common = bytes([1, 99, 0x81]) + bytes(13) + bytes([3])
+    device._get_basic_info = AsyncMock(side_effect=[common, bytes(15), reply])
+    assert await device.get_basic_info() is None
+
+
+@pytest.mark.parametrize("manufacturer", [None, b"", bytes(13)])
+def test_short_advertisement(manufacturer):
+
+    assert process_outdoor_plug(None, manufacturer) == {}
+
+
+@pytest.mark.asyncio
+async def test_advertisement_preserves_energy_and_refreshes_power():
+    device = create_device()
+    device.update = AsyncMock()
+
+    def advertisement(sequence, power, production):
+        return SwitchBotAdvertisement(
+            address="01:02:03:04:05:06",
+            data={
+                "data": {
+                    "sequence_number": sequence,
+                    "isOn": True,
+                    "power": power,
+                    "power_production": production,
+                }
+            },
+            device=device._device,
+            rssi=-40,
+            active=True,
+        )
+
+    device.update_from_advertisement(advertisement(1, 20, 30))
+    device.data["data"].update(
+        {"energy": 1.2, "energy_production": 3.4, "current_production": 0.5}
+    )
+    device.update_from_advertisement(advertisement(2, 40, 50))
+    assert device.parsed_data["energy"] == 1.2
+    assert device.parsed_data["energy_production"] == 3.4
+    assert device.parsed_data["current_production"] == 0.5
+    assert device.parsed_data["power"] == 40
+    assert device.parsed_data["power_production"] == 50
+
+
+def test_cloud_model_mapping():
+    assert API_MODEL_TO_ENUM["W1186000"] is SwitchbotModel.OUTDOOR_PLUG_EU

tests/test_relay_switch.py

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 from bleak.backends.device import BLEDevice
 
-from switchbot import SwitchBotAdvertisement, SwitchbotModel
+from switchbot import SwitchBotAdvertisement, SwitchbotModel, SwitchbotOutdoorPlug
 from switchbot.devices import relay_switch
 from switchbot.devices.device import _merge_data as merge_data
 
@@ -15,6 +15,7 @@ common_params = [
     (b">\x00\x00\x00", SwitchbotModel.GARAGE_DOOR_OPENER),
     (b"?\x00\x00\x00", SwitchbotModel.PLUG_MINI_EU),
     (b"\x00\x00\x00\x00\x12\x43\xc8\x15", SwitchbotModel.PLUG_MINI_IT),
+    (b"\x00\x00\x00\x00\x12\x18\xd0\x15", SwitchbotModel.OUTDOOR_PLUG_EU),
 ]
 
 
@@ -36,6 +37,8 @@ def create_device_for_command_testing(
         device_class = relay_switch.SwitchbotGarageDoorOpener
     elif model == SwitchbotModel.RELAY_SWITCH_2PM:
         device_class = relay_switch.SwitchbotRelaySwitch2PM
+    elif model == SwitchbotModel.OUTDOOR_PLUG_EU:
+        device_class = SwitchbotOutdoorPlug
     else:
         device_class = relay_switch.SwitchbotRelaySwitch
     device = device_class(