feat: add energy limit functions (#300)

* feat: add energy limit functions

* linting

* fix a test

* more coverage

* formatting

openevsehttp/__main__.py

@@ -13,10 +13,23 @@
 from awesomeversion import AwesomeVersion
 from awesomeversion.exceptions import AwesomeVersionCompareException
 
-from .const import MAX_AMPS, MIN_AMPS, SOLAR, GRID, BAT_LVL, BAT_RANGE, TTF, VOLTAGE
+from .const import (
+    BAT_LVL,
+    BAT_RANGE,
+    GRID,
+    MAX_AMPS,
+    MIN_AMPS,
+    RELEASE,
+    SOLAR,
+    TTF,
+    TYPE,
+    VALUE,
+    VOLTAGE,
+)
 from .exceptions import (
     AlreadyListening,
     AuthenticationError,
+    InvalidType,
     MissingMethod,
     MissingSerial,
     ParseJSONError,
@@ -131,7 +144,7 @@ async def process_request(
                         _LOGGER.error("Authentication error: %s", message)
                         raise AuthenticationError
                     if resp.status in [404, 405, 500]:
-                        _LOGGER.error("%s", message)
+                        _LOGGER.warning("%s", message)
 
                     if method == "post" and "config_version" in message:
                         await self.update()
@@ -658,6 +671,64 @@ async def soc(
             response = await self.process_request(url=url, method="post", data=data)
             _LOGGER.debug("SOC response: %s", response)
 
+    # Limit endpoint
+    async def set_limit(
+        self, limit_type: str, value: int, release: bool | None = None
+    ) -> Any:
+        """Set charge limit."""
+        if not self._version_check("5.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}limit"
+        data: Dict[str, Any] = {}
+        valid_types = ["time", "energy", "soc", "range"]
+
+        if limit_type not in valid_types:
+            raise InvalidType
+
+        data[TYPE] = limit_type
+        data[VALUE] = value
+        if release is not None:
+            data[RELEASE] = release
+
+        _LOGGER.debug("Limit data: %s", data)
+        _LOGGER.debug("Setting limit config on %s", url)
+        response = await self.process_request(
+            url=url, method="post", data=data
+        )  # noqa: E501
+        return response
+
+    async def clear_limit(self) -> Any:
+        """Clear charge limit."""
+        if not self._version_check("5.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}limit"
+        data: Dict[str, Any] = {}
+
+        _LOGGER.debug("Clearing limit config on %s", url)
+        response = await self.process_request(
+            url=url, method="delete", data=data
+        )  # noqa: E501
+        return response
+
+    async def get_limit(self) -> Any:
+        """Get charge limit."""
+        if not self._version_check("5.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}limit"
+        data: Dict[str, Any] = {}
+
+        _LOGGER.debug("Getting limit config on %s", url)
+        response = await self.process_request(
+            url=url, method="get", data=data
+        )  # noqa: E501
+        return response
+
     @property
     def hostname(self) -> str:
         """Return charger hostname."""

openevsehttp/const.py

@@ -10,3 +10,6 @@
 TTF = "time_to_full_charge"
 VOLTAGE = "voltage"
 SHAPER_LIVE = "shaper_live_pwr"
+TYPE = "type"
+VALUE = "value"
+RELEASE = "release"

openevsehttp/exceptions.py

@@ -27,3 +27,7 @@ class MissingSerial(Exception):
 
 class UnsupportedFeature(Exception):
     """Exception for firmware that is too old."""
+
+
+class InvalidType(Exception):
+    """Exception for invalid types."""

tests/test_main.py

@@ -1,19 +1,23 @@
 """Library tests."""
 
-import aiohttp
-from aiohttp.client_reqrep import ConnectionKey
 import asyncio
 import json
 import logging
 from unittest import mock
 
-from awesomeversion.exceptions import AwesomeVersionCompareException
-
+import aiohttp
 import pytest
 from aiohttp.client_exceptions import ContentTypeError, ServerTimeoutError
+from aiohttp.client_reqrep import ConnectionKey
+from awesomeversion.exceptions import AwesomeVersionCompareException
 
 import openevsehttp.__main__ as main
-from openevsehttp.exceptions import MissingSerial, UnknownError, UnsupportedFeature
+from openevsehttp.exceptions import (
+    InvalidType,
+    MissingSerial,
+    UnknownError,
+    UnsupportedFeature,
+)
 from tests.common import load_fixture
 
 pytestmark = pytest.mark.asyncio
@@ -24,6 +28,7 @@
 TEST_URL_CONFIG = "http://openevse.test.tld/config"
 TEST_URL_DIVERT = "http://openevse.test.tld/divertmode"
 TEST_URL_RESTART = "http://openevse.test.tld/restart"
+TEST_URL_LIMIT = "http://openevse.test.tld/limit"
 TEST_URL_WS = "ws://openevse.test.tld/ws"
 TEST_URL_GITHUB_v4 = (
     "https://api.github.com/repos/OpenEVSE/ESP32_WiFi_V4.x/releases/latest"
@@ -1564,3 +1569,91 @@ async def test_soc(test_charger, test_charger_v2, mock_aioclient, caplog):
         with caplog.at_level(logging.DEBUG):
             await test_charger_v2.soc(50, 90, 3100)
             assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_set_limit(
+    test_charger_modified_ver, test_charger, mock_aioclient, caplog
+):
+    """Test set limit."""
+    await test_charger_modified_ver.update()
+    mock_aioclient.post(
+        TEST_URL_LIMIT,
+        status=200,
+        body='{"msg": "OK"}',
+        repeat=True,
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger_modified_ver.set_limit("energy", 15, True)
+        assert (
+            "Limit data: {'type': 'energy', 'value': 15, 'release': True}"
+            in caplog.text
+        )
+        assert "Setting limit config on http://openevse.test.tld/limit" in caplog.text
+
+    with pytest.raises(InvalidType):
+        await test_charger_modified_ver.set_limit("invalid", 15)
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger.set_limit("energy", 15)
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_get_limit(
+    test_charger_modified_ver, test_charger, mock_aioclient, caplog
+):
+    """Test get limit."""
+    await test_charger_modified_ver.update()
+    mock_aioclient.get(
+        TEST_URL_LIMIT,
+        status=200,
+        body='{"type": "energy", "value": 10}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        response = await test_charger_modified_ver.get_limit()
+        assert response == {"type": "energy", "value": 10}
+        assert "Getting limit config on http://openevse.test.tld/limit" in caplog.text
+
+    mock_aioclient.get(
+        TEST_URL_LIMIT,
+        status=404,
+        body='{"msg": "No limit"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        response = await test_charger_modified_ver.get_limit()
+        assert response == {"msg": "No limit"}
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger.get_limit()
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_clear_limit(
+    test_charger_modified_ver, test_charger, mock_aioclient, caplog
+):
+    """Test clear limit."""
+    await test_charger_modified_ver.update()
+    mock_aioclient.delete(
+        TEST_URL_LIMIT,
+        status=200,
+        body='{"msg": "Deleted"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        response = await test_charger_modified_ver.clear_limit()
+        assert response == {"msg": "Deleted"}
+        assert "Clearing limit config on http://openevse.test.tld/limit" in caplog.text
+
+    mock_aioclient.delete(
+        TEST_URL_LIMIT,
+        status=404,
+        body='{"msg": "No limit to clear"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        response = await test_charger_modified_ver.clear_limit()
+        assert response == {"msg": "No limit to clear"}
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger.clear_limit()
+            assert "Feature not supported for older firmware." in caplog.text