Add shaper toggle functionality

openevsehttp/commands.py

@@ -491,3 +491,35 @@ async def set_divert_mode(self, mode: str = "fast") -> None:
         if not success:
             _LOGGER.error("Problem issuing command: %s", response)
             raise UnknownError
+
+    async def set_shaper(self, enable: bool = True) -> None:
+        """Set shaper mode."""
+        if not self._version_check("4.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}shaper"
+        mode = 1 if enable else 0
+        data = {"mode": mode}
+
+        _LOGGER.debug("Setting shaper to %s", mode)
+        response = await self.process_request(url=url, method="post", data=data)
+        response = self._normalize_response(response)
+        msg = response.get("msg") if isinstance(response, Mapping) else None
+        if msg not in ["OK", "done", "no change"]:
+            _LOGGER.error("Problem issuing command: %s", response)
+            raise UnknownError
+
+    async def toggle_shaper(self) -> None:
+        """Toggle shaper mode."""
+        shaper_active = self._status.get("shaper")
+        if shaper_active is None:
+            await self.update()
+            shaper_active = self._status.get("shaper")
+
+        if shaper_active is None:
+            _LOGGER.error("Cannot toggle shaper: unknown shaper state.")
+            raise RuntimeError("Cannot toggle shaper: unknown shaper state.")
+
+        new_state = not bool(shaper_active)
+        await self.set_shaper(new_state)

setup.py

@@ -6,7 +6,7 @@
 
 PROJECT_DIR = Path(__file__).parent.resolve()
 README_FILE = PROJECT_DIR / "README.md"
-VERSION = "0.3.0"
+VERSION = "0.3.1"
 
 setup(
     name="python_openevse_http",

tests/test_shaper.py

@@ -0,0 +1,128 @@
+"""Tests for shaper command methods."""
+
+import logging
+
+import pytest
+
+from openevsehttp.exceptions import UnknownError, UnsupportedFeature
+
+pytestmark = pytest.mark.asyncio
+
+TEST_URL_SHAPER = "http://openevse.test.tld/shaper"
+
+
+async def test_set_shaper(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test set_shaper command."""
+    await test_charger.update()
+    mock_aioclient.post(
+        TEST_URL_SHAPER,
+        status=200,
+        body='{"msg": "OK"}',
+        repeat=True,
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.set_shaper(True)
+        assert "Setting shaper to 1" in caplog.text
+
+        await test_charger.set_shaper(False)
+        assert "Setting shaper to 0" in caplog.text
+
+    await test_charger_v2.update()
+    # Force version lower than 4.0.0
+    test_charger_v2._config["version"] = "3.3.1"
+    with pytest.raises(UnsupportedFeature):
+        await test_charger_v2.set_shaper(True)
+
+
+async def test_set_shaper_fail(test_charger, mock_aioclient, caplog):
+    """Test set_shaper failure."""
+    await test_charger.update()
+    mock_aioclient.post(
+        TEST_URL_SHAPER,
+        status=200,
+        body='{"msg": "failure!"}',
+    )
+    with pytest.raises(UnknownError):
+        await test_charger.set_shaper(True)
+
+
+async def test_toggle_shaper(test_charger, mock_aioclient, caplog):
+    """Test toggle_shaper command."""
+    await test_charger.update()
+    # Initial state from status fixture is likely True or False
+    # Let's force it to 0 (off)
+    test_charger._status["shaper"] = 0
+
+    mock_aioclient.post(
+        TEST_URL_SHAPER,
+        status=200,
+        body='{"msg": "OK"}',
+    )
+
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.toggle_shaper()
+        assert "Setting shaper to 1" in caplog.text
+
+    # Now it's on (1)
+    test_charger._status["shaper"] = 1
+    mock_aioclient.post(
+        TEST_URL_SHAPER,
+        status=200,
+        body='{"msg": "OK"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.toggle_shaper()
+        assert "Setting shaper to 0" in caplog.text
+
+
+async def test_toggle_shaper_missing_state(test_charger, mock_aioclient, caplog):
+    """Test toggle_shaper when state is missing."""
+    # Clear status to force update()
+    test_charger._status = {}
+
+    # Mock the /status call that update() will make
+    from tests.common import load_fixture
+
+    TEST_URL_STATUS = "http://openevse.test.tld/status"
+    mock_aioclient.get(
+        TEST_URL_STATUS,
+        status=200,
+        body=load_fixture("v4_json/status.json"),
+    )
+
+    mock_aioclient.post(
+        TEST_URL_SHAPER,
+        status=200,
+        body='{"msg": "OK"}',
+    )
+
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.toggle_shaper()
+        # status.json has shaper: 1, so it should toggle to 0
+        assert "Setting shaper to 0" in caplog.text
+
+
+async def test_toggle_shaper_failed_update(mock_aioclient, caplog):
+    """Test toggle_shaper when state is still missing after update()."""
+    from openevsehttp import OpenEVSE
+
+    charger = OpenEVSE("openevse.test.tld")
+
+    # Mock the /status call but return status without shaper
+    mock_aioclient.get(
+        "http://openevse.test.tld/status",
+        status=200,
+        body='{"mode": 1}',  # No shaper key
+    )
+    mock_aioclient.get(
+        "http://openevse.test.tld/config",
+        status=200,
+        body='{"firmware": "4.1.2"}',
+    )
+
+    with pytest.raises(
+        RuntimeError, match="Cannot toggle shaper: unknown shaper state."
+    ):
+        await charger.toggle_shaper()
+
+    assert "Cannot toggle shaper: unknown shaper state." in caplog.text