Feature: Take powerwall on/off grid (#42)

* Fix typo in GridStatus enum

* Add unit tests for expected behaviour of set_island_mode

* First attempt at passing through command to API

* Add a quick integration test, hoping race conditions won't cause in issues here 🤞

* IslandMode enum suggestion

* README updates

* Add missing "Transition to island" GridStatus value (observed during integration tests)

* Fix incorrect serialisation when POSTing to v2 endpoint

* Fix up integration test to test going offline and then online (or vice versa depending on start state)

* Update internal Post method to use json= and set default headers

* Switch to implied content-type from json= parameter

README.md

@@ -34,6 +34,7 @@ Powerwall Software versions from 1.47.0 to 1.50.1 as well as 20.40 to 22.9.2 are
     - [Powerwalls Serial Numbers](#powerwalls-serial-numbers)
     - [Gateway DIN](#gateway-din)
     - [VIN](#vin)
+    - [Off-grid status](#off-grid-status-set-island-mode)
 ## Installation
 
 Install the library via pip:
@@ -357,3 +358,19 @@ din = powerwall.get_gateway_din()
 ```python
 vin = powerwall.get_vin()
 ```
+
+### Off-grid status (Set Island mode)
+
+Take your powerwall on- and off-grid similar to the "Take off-grid" button in the Tesla app. 
+
+#### Set powerwall to off-grid (Islanded)
+
+```python
+powerwall.set_island_mode(IslandMode.OFFGRID)
+```
+
+#### Set powerwall to off-grid (Connected)
+
+```python
+powerwall.set_island_mode(IslandMode.ONGRID)
+```

tesla_powerwall/__init__.py

@@ -5,6 +5,7 @@
     DeviceType,
     GridState,
     GridStatus,
+    IslandMode,
     LineStatus,
     MeterType,
     OperationMode,

tesla_powerwall/api.py

@@ -131,7 +131,7 @@ def post(
         try:
             response = self._http_session.post(
                 url=self.url(path),
-                data=payload,
+                json=payload,
                 timeout=self._timeout,
                 headers=headers,
             )
@@ -245,3 +245,6 @@ def get_site_info_grid_codes(self) -> list:
 
     def post_site_info_site_name(self, body: dict) -> dict:
         return self.post("site_info/site_name", body)
+
+    def post_islanding_mode(self, body: dict) -> dict:
+        return self.post("v2/islanding/mode", body)

tesla_powerwall/const.py

@@ -20,10 +20,14 @@ class Roles(Enum):
 
 class GridStatus(Enum):
     CONNECTED = "SystemGridConnected"
-    ISLANEDED_READY = "SystemIslandedReady"
-    ISLANEDED = "SystemIslandedActive"
+    ISLANDED_READY = "SystemIslandedReady"
+    ISLANDED = "SystemIslandedActive"
     TRANSITION_TO_GRID = "SystemTransitionToGrid"  # Used in version 1.46.0
+    TRANSITION_TO_ISLAND = "SystemTransitionToIsland"
 
+class IslandMode(Enum):
+    OFFGRID = "intentional_reconnect_failsafe"
+    ONGRID = "backup"
 
 class GridState(Enum):
     COMPLIANT = "Grid_Compliant"

tesla_powerwall/powerwall.py

@@ -9,6 +9,7 @@
     DeviceType,
     GridState,
     GridStatus,
+    IslandMode,
     LineStatus,
     MeterType,
     OperationMode,
@@ -161,6 +162,9 @@ def get_solars(self) -> List[Solar]:
     def get_vin(self) -> str:
         return assert_attribute(self._api.get_config(), "vin", "config")
 
+    def set_island_mode(self, mode: IslandMode) -> IslandMode:
+        return IslandMode(assert_attribute(self._api.post_islanding_mode({"island_mode": mode.value}), "island_mode"))
+
     def get_version(self) -> str:
         version_str = assert_attribute(self._api.get_status(), "version", "status")
         return version_str.split(" ")[

tests/integration/test_powerwall.py

@@ -1,7 +1,9 @@
 import unittest
+from time import sleep
 
 from tesla_powerwall import (
     GridStatus,
+    IslandMode,
     Meter,
     MetersAggregates,
     MeterType,
@@ -96,3 +98,39 @@ def test_status(self) -> None:
         status.up_time_seconds
         status.start_time
         status.version
+
+    def test_islanding(self) -> None:
+        initial_grid_status = self.powerwall.get_grid_status()
+        self.assertIsInstance(initial_grid_status, GridStatus)
+
+        if(initial_grid_status == GridStatus.CONNECTED):
+            self.go_offline()
+            self.go_online()
+        elif(initial_grid_status == GridStatus.ISLANDED):
+            self.go_offline()
+            self.go_online()
+
+    def go_offline(self) -> None:
+        observedIslandMode = self.powerwall.set_island_mode(IslandMode.OFFGRID)
+        self.assertEqual(observedIslandMode, IslandMode.OFFGRID)
+        self.wait_until_grid_status(GridStatus.ISLANDED)
+        self.assertEqual(self.powerwall.get_grid_status(), GridStatus.ISLANDED)
+
+    def go_online(self) -> None:
+        observedIslandMode = self.powerwall.set_island_mode(IslandMode.ONGRID)
+        self.assertEqual(observedIslandMode, IslandMode.ONGRID)
+        self.wait_until_grid_status(GridStatus.CONNECTED)
+        self.assertEqual(self.powerwall.get_grid_status(), GridStatus.CONNECTED)
+
+    def wait_until_grid_status(self, expectedStatus: GridStatus, sleepTime: int = 1, maxCycles: int = 20) -> None:
+        cycles = 0
+        observedStatus: GridStatus
+
+        while cycles < maxCycles:
+            observedStatus = self.powerwall.get_grid_status()
+            if(observedStatus == expectedStatus):
+                break
+            sleep(sleepTime)
+            cycles = cycles + 1
+
+        self.assertEqual(observedStatus, expectedStatus)

tests/unit/fixtures/islanding_mode_offgrid.json

@@ -0,0 +1 @@
+{"island_mode": "intentional_reconnect_failsafe"}

tests/unit/fixtures/islanding_mode_ongrid.json

@@ -0,0 +1 @@
+{"island_mode": "backup"}

tests/unit/test_powerwall.py

@@ -8,6 +8,7 @@
     API,
     DeviceType,
     GridStatus,
+    IslandMode,
     Meter,
     MeterNotAvailableError,
     MetersAggregates,
@@ -29,6 +30,8 @@
     SITEMASTER_RESPONSE,
     STATUS_RESPONSE,
     SYSTEM_STATUS_RESPONSE,
+    ISLANDING_MODE_ONGRID_RESPONSE,
+    ISLANDING_MODE_OFFGRID_RESPONSE,
 )
 
 
@@ -230,6 +233,32 @@ def test_system_status(self):
         self.assertEqual(batteries[0].energy_discharged, 4659550)
         self.assertEqual(batteries[0].wobble_detected, False)
 
+    @responses.activate
+    def test_islanding_mode_offgrid(self):
+        add(
+            Response(
+                responses.POST,
+                url=f"{ENDPOINT}v2/islanding/mode",
+                json=ISLANDING_MODE_OFFGRID_RESPONSE,
+            )
+        )
+
+        mode = self.powerwall.set_island_mode(IslandMode.OFFGRID)
+        self.assertEqual(mode, IslandMode.OFFGRID)
+
+    @responses.activate
+    def test_islanding_mode_ongrid(self):
+        add(
+            Response(
+                responses.POST,
+                url=f"{ENDPOINT}v2/islanding/mode",
+                json=ISLANDING_MODE_ONGRID_RESPONSE,
+            )
+        )
+
+        mode = self.powerwall.set_island_mode(IslandMode.ONGRID)
+        self.assertEqual(mode, IslandMode.ONGRID)
+
     def test_helpers(self):
         resp = {"a": 1}
         with self.assertRaises(MissingAttributeError):