feat: Add shaper_live_pwr method

openevsehttp/__main__.py

@@ -1312,6 +1312,20 @@ def charging_power(self) -> float | None:
             return round(self._status["voltage"] * self._status["amp"], 2)
         return None
 
+    # Shaper HTTP Posting
+    async def set_shaper_live_pwr(self, power: int) -> None:
+        """Send pushed sensor data to shaper."""
+        if not self._version_check("4.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}status"
+        data = {"shaper_live_pwr": power}
+
+        _LOGGER.debug("Posting shaper data: %s", data)
+        response = await self.process_request(url=url, method="post", data=data)
+        _LOGGER.debug("Shaper response: %s", response)
+
     # Shaper values
     @property
     def shaper_active(self) -> bool | None:

tests/test_external_session.py

@@ -1,12 +1,11 @@
 """Test external session management."""
 
 import json
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
 
 import aiohttp
 import pytest
 
-import openevsehttp.__main__ as main
 from openevsehttp.__main__ import OpenEVSE
 from tests.common import load_fixture
 

tests/test_main.py

@@ -11,7 +11,6 @@
 import pytest
 from aiohttp.client_exceptions import ContentTypeError, ServerTimeoutError
 from aiohttp.client_reqrep import ConnectionKey
-from awesomeversion import AwesomeVersion
 from awesomeversion.exceptions import AwesomeVersionCompareException
 from freezegun import freeze_time
 
@@ -1271,6 +1270,38 @@ async def test_shaper_max_power(fixture, expected, request):
     await charger.ws_disconnect()
 
 
+async def test_set_shaper_live_power(
+    test_charger, test_charger_v2, mock_aioclient, caplog
+):
+    """Test setting shaper live power."""
+    await test_charger.update()
+    mock_aioclient.post(
+        TEST_URL_STATUS,
+        status=200,
+        body='{"shaper_live_pwr": 210}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.set_shaper_live_pwr(210)
+        assert "Posting shaper data: {'shaper_live_pwr': 210}" in caplog.text
+        assert "Shaper response: {'shaper_live_pwr': 210}" in caplog.text
+
+    mock_aioclient.post(
+        TEST_URL_STATUS,
+        status=200,
+        body='{"shaper_live_pwr": 0}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.set_shaper_live_pwr(0)
+        assert "Posting shaper data: {'shaper_live_pwr': 0}" in caplog.text
+
+    await test_charger_v2.update()
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.set_shaper_live_pwr(210)
+    assert "Feature not supported for older firmware." in caplog.text
+    await test_charger_v2.ws_disconnect()
+
+
 @pytest.mark.parametrize(
     "fixture, expected", [("test_charger", 75), ("test_charger_v2", None)]
 )