Add export control

franklinwh/__init__.py

@@ -5,6 +5,8 @@
 from .client import (
     AccessoryType,
     Client,
+    ExportMode,
+    ExportSettings,
     GridStatus,
     HttpClientFactory,
     Mode,
@@ -18,6 +20,8 @@
     "AccessoryType",
     "CachingThread",
     "Client",
+    "ExportMode",
+    "ExportSettings",
     "GridStatus",
     "HttpClientFactory",
     "Mode",

franklinwh/client.py

@@ -129,6 +129,52 @@ def from_offgridreason(value: int | None) -> GridStatus:
                 raise ValueError(f"Unknown offgridreason value: {value}")
 
 
+class ExportMode(Enum):
+    """Represents the grid export mode for the FranklinWH gateway.
+
+    Attributes:
+        SOLAR_ONLY (int): Solar can export to the grid; battery (aPower) cannot.
+        SOLAR_AND_APOWER (int): Both solar and battery can export to the grid.
+        NO_EXPORT (int): No grid export permitted.
+    """
+
+    SOLAR_ONLY = 1
+    SOLAR_AND_APOWER = 2
+    NO_EXPORT = 3
+
+    @staticmethod
+    def from_flag(value: int) -> ExportMode:
+        """Convert a gridFeedMaxFlag API value to an ExportMode.
+
+        Parameters
+        ----------
+        value : int
+            The gridFeedMaxFlag value from the API response.
+
+        Returns:
+        -------
+        ExportMode
+            The corresponding ExportMode.
+        """
+        try:
+            return ExportMode(value)
+        except ValueError:
+            return ExportMode.SOLAR_ONLY
+
+
+@dataclass
+class ExportSettings:
+    """Current grid export configuration for the FranklinWH gateway.
+
+    Attributes:
+        mode: The active export mode.
+        limit_kw: Export power cap in kW, or None if unlimited.
+    """
+
+    mode: ExportMode
+    limit_kw: float | None
+
+
 @dataclass
 class Current:
     """Current statistics for FranklinWH gateway."""
@@ -739,6 +785,73 @@ async def set_grid_status(self, status: GridStatus, soc: int = 5):
         }
         await self._post(url, json.dumps(payload))
 
+    async def get_export_settings(self) -> ExportSettings:
+        """Get the current grid export mode and power limit.
+
+        Returns:
+        -------
+        ExportSettings
+            The active export mode and optional kW cap.
+        """
+        url = self.url_base + "hes-gateway/terminal/tou/getPowerControlSetting"
+        result = (await self._get(url))["result"]
+        mode = ExportMode.from_flag(result["gridFeedMaxFlag"])
+        feed_max = result.get("gridFeedMax", -1.0)
+        limit_kw = None if feed_max < 0 else feed_max
+        return ExportSettings(mode=mode, limit_kw=limit_kw)
+
+    async def set_export_settings(
+        self, mode: ExportMode, limit_kw: float | None = None
+    ) -> None:
+        """Set the grid export mode and optional power limit.
+
+        Uses a read-modify-write pattern: the setPowerControlV2 endpoint
+        requires all existing settings to be echoed back alongside the
+        fields being changed.
+
+        Parameters
+        ----------
+        mode : ExportMode
+            The desired export mode.
+        limit_kw : float | None, optional
+            Export power cap in kW (0.1–10000.0). None means unlimited.
+            Ignored when mode is NO_EXPORT.
+        """
+        get_url = self.url_base + "hes-gateway/terminal/tou/getPowerControlSetting"
+        set_url = self.url_base + "hes-gateway/terminal/tou/setPowerControlV2"
+
+        # Read current settings — endpoint requires all fields posted back
+        current = (await self._get(get_url))["result"]
+
+        if mode == ExportMode.NO_EXPORT:
+            feed_max = 0.0
+            discharge_max = 0.0
+        elif mode == ExportMode.SOLAR_AND_APOWER:
+            feed_max = -1.0 if limit_kw is None else float(limit_kw)
+            discharge_max = -1.0
+        else:  # SOLAR_ONLY
+            feed_max = -1.0 if limit_kw is None else float(limit_kw)
+            discharge_max = 0.0
+
+        payload = {k: v for k, v in current.items() if v is not None}
+        payload.update({
+            "gatewayId": self.gateway,
+            "lang": "EN_US",
+            "gridFeedMaxFlag": mode.value,
+            "gridFeedMax": feed_max,
+            "globalGridDischargeMax": discharge_max,
+        })
+
+        res = await self.session.post(
+            set_url,
+            headers={"loginToken": self.token, "Content-Type": "application/json"},
+            data=json.dumps(payload),
+        )
+        res.raise_for_status()
+        body = res.json()
+        if body.get("code") != 200:
+            raise RuntimeError(f"set_export_settings failed: {body}")
+
     async def get_composite_info(self):
         """Get composite information about the FranklinWH gateway."""
         url = self.url_base + "hes-gateway/terminal/getDeviceCompositeInfo"