Add set_max_current to control the max current of a station

examples/set_max_current.py

@@ -0,0 +1,46 @@
+"""Asynchronous Python client for TechnoVE."""
+
+import asyncio
+
+from technove import MIN_CURRENT, TechnoVE
+
+
+async def main() -> None:
+    """Show example on setting the max current of a station."""
+    async with TechnoVE("192.168.10.162") as technove:
+        print("Initial value:")
+        device = await technove.update()
+        initial_value = device.info.max_current
+        print(initial_value)
+
+        print("Setting max current to maximum...")
+        await technove.set_max_current(device.info.max_station_current)
+        # Sleep is needed because the station takes a bit of time to fully
+        # set the correct value.
+        await asyncio.sleep(10)
+        device = await technove.update()
+        print(device.info.max_current)
+
+        print("Setting max current to minimum...")
+        await technove.set_max_current(MIN_CURRENT)
+        await asyncio.sleep(10)
+        device = await technove.update()
+        print(device.info.max_current)
+
+        print("Setting max current to odd value...")
+        await technove.set_max_current(15)
+        await asyncio.sleep(10)
+        device = await technove.update()
+        print(device.info.max_current)
+
+        if device.info.max_current != initial_value:
+            # Sets the initial value back, just to be nice
+            print("Setting back to initial value...")
+            await technove.set_max_current(initial_value)
+            await asyncio.sleep(10)
+            device = await technove.update()
+            print(device.info.max_current)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/technove/__init__.py

@@ -6,13 +6,14 @@
     TechnoVEError,
 )
 from .models import Info, Station, Status
-from .technove import TechnoVE
+from .technove import MIN_CURRENT, TechnoVE
 
 __all__ = [
     "Station",
     "Status",
     "Info",
     "TechnoVE",
+    "MIN_CURRENT",
     "TechnoVEConnectionError",
     "TechnoVEConnectionTimeoutError",
     "TechnoVEError",

src/technove/exceptions.py

@@ -5,6 +5,10 @@ class TechnoVEError(Exception):
     """Generic TechnoVE exception."""
 
 
+class TechnoVEOutOfBoundError(TechnoVEError):
+    """Generic TechnoVE exception."""
+
+
 class TechnoVEConnectionError(Exception):
     """TechnoVE connection exception."""
 

src/technove/technove.py

@@ -1,10 +1,11 @@
 """Asynchronous Python client for TechnoVE."""
+
 from __future__ import annotations
 
 import asyncio
 import json
 from dataclasses import dataclass
-from typing import Any, Self
+from typing import Any, Final, Self
 
 import aiohttp
 import backoff
@@ -14,9 +15,14 @@
     TechnoVEConnectionError,
     TechnoVEConnectionTimeoutError,
     TechnoVEError,
+    TechnoVEOutOfBoundError,
 )
 from .models import Station
 
+# TechnoVE stations don't allow setting values lower than 8. Calling the API with
+# a smaller value will just be ignored.
+MIN_CURRENT: Final[int] = 8
+
 
 @dataclass
 class TechnoVE:
@@ -36,11 +42,18 @@ class TechnoVE:
     # /station/network/list
     # /station/get/schedule
     # /station/partage/get/config
+    # /station/partage/set/config
+    # - ex: {"partageUuid":"GUID-HERE","partageName":"Mon partage",
+    #       "configuredMaxCurrent":24,"stations":[{"uuid":"{mac-address}",
+    #       "name":"nameHere","ipAddress":"192.168.1.2","configuredMaxCurrent":24,
+    #       "maxCurrent":32}]}
     # /station/update (POST)
     # /station/schedule/high/activate (POST)
     # /station/set/automatic (POST)
     # /station/control/stop
     # /station/control/start
+    # /station/control/partage (POST)
+    #    - ex: {" stationNumber": 1, "current":48.0}
 
     # 65: not plugged in, waiting
     # 66: plugged in, waiting
@@ -78,6 +91,7 @@ async def request(
             TechnoVEConnectionTimeoutError: A timeout occurred while communicating
                 with the TechnoVE station.
             TechnoVEError: Received an unexpected response from the TechnoVE station.
+
         """
         url = URL.build(scheme="http", host=self.station_ip, port=80, path=uri)
 
@@ -142,6 +156,7 @@ async def update(self) -> Station:
         Returns
         -------
             TechnoVE station data.
+
         """
         data = await self.request("/station/get/info")
         if not data:
@@ -156,6 +171,7 @@ async def set_auto_charge(self, *, enabled: bool) -> None:
         Args:
         ----
             enabled: True to enable the auto-charge feature, otherwise false.
+
         """
         await self.request(
             "/station/set/automatic", method="POST", data={"activated": enabled}
@@ -173,13 +189,49 @@ async def set_charging_enabled(self, *, enabled: bool) -> None:
         Raises:
         ------
             TechnoVEError: If auto_charge is enabled.
+
         """
         if self.station and self.station.info.auto_charge:
             msg = "Cannot start or stop charging when auto-charge is enabled."
             raise TechnoVEError(msg)
         action = "start" if enabled else "stop"
         await self.request(f"/station/control/{action}")
 
+    async def set_max_current(self, max_current: int) -> None:
+        """Set the max current the station is allowed to provide.
+
+        This can only be set if the sharing mode feature is not enabled.
+
+        Args:
+        ----
+            max_current: The maximum current the station can provide to the vehicle.
+                This value must be between 8 and max_station_current.
+
+        Raises:
+        ------
+            TechnoVEError: If in_sharing_mode is enabled.
+            TechnoVEOutOfBoundError: If max_current is below 8 (See MIN_CURRENT).
+            TechnoVEOutOfBoundError: If max_current is above max_station_current.
+
+        """
+        if self.station and self.station.info.in_sharing_mode:
+            msg = "Cannot set the max current when sharing mode is enabled."
+            raise TechnoVEError(msg)
+        if max_current < MIN_CURRENT:
+            msg = f"Max current needs to be greater than {MIN_CURRENT}."
+            raise TechnoVEOutOfBoundError(msg)
+        if self.station and max_current > self.station.info.max_station_current:
+            msg = (
+                "Max current needs to be equal or lower than "
+                f"{self.station.info.max_station_current}."
+            )
+            raise TechnoVEOutOfBoundError(msg)
+        await self.request(
+            "/station/control/partage",
+            "POST",
+            {" stationNumber": 1, "current": max_current},
+        )
+
     async def close(self) -> None:
         """Close open client session."""
         if self.session and self._close_session:
@@ -191,6 +243,7 @@ async def __aenter__(self) -> Self:
         Returns
         -------
             The TechnoVE object.
+
         """
         return self
 
@@ -200,5 +253,6 @@ async def __aexit__(self, *_exc_info: object) -> None:
         Args:
         ----
             _exc_info: Exec type.
+
         """
         await self.close()

tests/test_technove.py

@@ -7,7 +7,11 @@
 from aresponses import Response, ResponsesMockServer
 
 from technove import Station, Status, TechnoVE
-from technove.exceptions import TechnoVEConnectionError, TechnoVEError
+from technove.exceptions import (
+    TechnoVEConnectionError,
+    TechnoVEError,
+    TechnoVEOutOfBoundError,
+)
 
 
 @pytest.mark.asyncio
@@ -292,3 +296,50 @@ async def test_set_charging_enabled_auto_charge() -> None:
     technove.station = Station({"auto_charge": True})
     with pytest.raises(TechnoVEError):
         await technove.set_charging_enabled(enabled=True)
+
+
+@pytest.mark.asyncio
+async def test_set_max_current(aresponses: ResponsesMockServer) -> None:
+    """Test that changing set_max_current calls the right API."""
+    aresponses.add(
+        "example.com",
+        "/station/control/partage",
+        "POST",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "plain/text"},
+            text="ok",
+        ),
+    )
+    async with aiohttp.ClientSession() as session:
+        technove = TechnoVE("example.com", session=session)
+        technove.station = Station({"maxStationCurrent": 32, "in_sharing_mode": False})
+        await technove.set_max_current(32)
+        aresponses.assert_plan_strictly_followed()
+
+
+@pytest.mark.asyncio
+async def test_set_max_current_sharing_mode() -> None:
+    """Test failure when setting the max current and in_sharing_mode is enabled."""
+    technove = TechnoVE("example.com")
+    technove.station = Station({"maxStationCurrent": 32, "in_sharing_mode": True})
+    with pytest.raises(TechnoVEError):
+        await technove.set_max_current(32)
+
+
+@pytest.mark.asyncio
+async def test_set_max_current_too_low() -> None:
+    """Test failure when setting the max current below 8."""
+    technove = TechnoVE("example.com")
+    technove.station = Station({"maxStationCurrent": 32, "in_sharing_mode": False})
+    with pytest.raises(TechnoVEOutOfBoundError):
+        await technove.set_max_current(2)
+
+
+@pytest.mark.asyncio
+async def test_set_max_current_too_high() -> None:
+    """Test failure when setting the max current below 0."""
+    technove = TechnoVE("example.com")
+    technove.station = Station({"maxStationCurrent": 32, "in_sharing_mode": False})
+    with pytest.raises(TechnoVEOutOfBoundError):
+        await technove.set_max_current(48)