Merge pull request #2 from Moustachauve/automatic-charging-toggle

Automatic charging toggle

examples/get_info.py

@@ -0,0 +1,19 @@
+"""Asynchronous Python client for TechnoVE."""
+
+import asyncio
+
+from technove import TechnoVE
+
+
+async def main() -> None:
+    """Show example on getting infos from your TechnoVE station."""
+    async with TechnoVE("192.168.10.162") as technove:
+        device = await technove.update()
+        print(device.info.name)
+        print(device.info.version)
+
+        print(device.info)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/ruff.toml

@@ -0,0 +1,6 @@
+# This extend our general Ruff rules specifically for the examples
+extend = "../pyproject.toml"
+
+extend-ignore = [
+  "T201", # Allow the use of print() in examples
+]

examples/set_auto_charge.py

@@ -0,0 +1,40 @@
+"""Asynchronous Python client for TechnoVE."""
+
+import asyncio
+
+from technove import TechnoVE
+
+
+async def main() -> None:
+    """Show example on setting the automatic charging feature."""
+    async with TechnoVE("192.168.10.162") as technove:
+        print("Initial value:")
+        device = await technove.update()
+        initial_value = device.info.auto_charge
+        print(initial_value)
+
+        print("Activating auto_charge...")
+        await technove.set_auto_charge(enabled=True)
+        # Sleep is needed because the station takes a bit of time to fully
+        # enable the automatic charging feature.
+        await asyncio.sleep(2)
+        device = await technove.update()
+        print(device.info.auto_charge)
+
+        print("Disabling auto_charge...")
+        await technove.set_auto_charge(enabled=False)
+        await asyncio.sleep(2)
+        device = await technove.update()
+        print(device.info.auto_charge)
+
+        if device.info.auto_charge != initial_value:
+            # Sets the initial value back, just to be nice
+            print("Setting back to initial value...")
+            await technove.set_auto_charge(enabled=initial_value)
+            await asyncio.sleep(2)
+            device = await technove.update()
+            print(device.info.auto_charge)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/technove/models.py

@@ -11,6 +11,7 @@
 class Status(Enum):
     """Describes the status of a TechnoVE station."""
 
+    UNKNOWN = "unknown"
     UNPLUGGED = "unplugged"
     PLUGGED_WAITING = "plugged_waiting"
     PLUGGED_CHARGING = "plugged_charging"
@@ -19,6 +20,7 @@ class Status(Enum):
     def build(cls: type[Status], status: int) -> Status:
         """Parse the status code int to a Status object."""
         statuses = {
+            None: Status.UNKNOWN,
             65: Status.UNPLUGGED,
             66: Status.PLUGGED_WAITING,
             67: Status.PLUGGED_CHARGING,
@@ -125,7 +127,7 @@ def from_dict(data: dict[str, Any]) -> Info:
             network_ssid=data.get("network_ssid", "Unknown"),
             normal_period_active=data.get("normalPeriodActive", False),
             rssi=data.get("rssi", 0),
-            status=Status.build(data.get("status", 0)),
+            status=Status.build(data.get("status", None)),
             time=data.get("time", 0),
             version=data.get("version", "Unknown"),
             voltage_in=data.get("voltageIn", 0),

src/technove/technove.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from dataclasses import dataclass
 from typing import Any, Self
 
@@ -97,16 +98,25 @@ async def request(
                     headers=headers,
                 )
 
+            content_type = response.headers.get("Content-Type", "")
             if response.status // 100 in [4, 5]:
                 contents = await response.read()
                 response.close()
 
+                if content_type == "application/json":
+                    raise TechnoVEError(
+                        response.status,
+                        json.loads(contents.decode("utf8")),
+                    )
                 raise TechnoVEError(
                     response.status,
                     {"message": contents.decode("utf8")},
                 )
 
-            response_data = await response.json()
+            if "application/json" in content_type:
+                response_data = await response.json()
+            else:
+                response_data = await response.text()
 
         except asyncio.TimeoutError as exception:
             msg = (
@@ -133,9 +143,43 @@ async def update(self) -> Station:
         -------
             TechnoVE station data.
         """
-        self.station = Station(await self.request("/station/get/info"))
+        data = await self.request("/station/get/info")
+        if not data:
+            msg = "No data was returned by the station"
+            raise TechnoVEError(msg)
+        self.station = Station(data)
         return self.station
 
+    async def set_auto_charge(self, *, enabled: bool) -> None:
+        """Set whether the auto-charge feature is enabled or disabled.
+
+        Args:
+        ----
+            enabled: True to enable the auto-charge feature, otherwise false.
+        """
+        await self.request(
+            "/station/set/automatic", method="POST", data={"activated": enabled}
+        )
+
+    async def set_charging_enabled(self, *, enabled: bool) -> None:
+        """Set whether the charging station is allowed to provide power or not.
+
+        This can only be set if the auto_charge feature is not enabled.
+
+        Args:
+        ----
+            enabled: True to allow a plugged-in vehicle to charge, otherwise false.
+
+        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 close(self) -> None:
         """Close open client session."""
         if self.session and self._close_session:

tests/test_models.py

@@ -0,0 +1,17 @@
+"""Tests for `technove.TechnoVE`."""
+
+
+import pytest
+
+from technove import Status, TechnoVEError
+
+
+def test_status_build() -> None:
+    """Test status build with a known status code."""
+    assert Status.build(67) == Status.PLUGGED_CHARGING
+
+
+def test_status_build_unknown() -> None:
+    """Test status build with an unknown status code."""
+    with pytest.raises(TechnoVEError):
+        Status.build(42)

tests/test_technove.py

@@ -6,7 +6,7 @@
 import pytest
 from aresponses import Response, ResponsesMockServer
 
-from technove import TechnoVE
+from technove import Station, TechnoVE
 from technove.exceptions import TechnoVEConnectionError, TechnoVEError
 
 
@@ -30,7 +30,7 @@ async def test_json_request(aresponses: ResponsesMockServer) -> None:
 
 
 @pytest.mark.asyncio
-async def test_internal_session(aresponses: ResponsesMockServer) -> None:
+async def test_json_request_internal_session(aresponses: ResponsesMockServer) -> None:
     """Test JSON response is handled correctly."""
     aresponses.add(
         "example.com",
@@ -47,6 +47,25 @@ async def test_internal_session(aresponses: ResponsesMockServer) -> None:
         assert response["status"] == "ok"
 
 
+@pytest.mark.asyncio
+async def test_text_request(aresponses: ResponsesMockServer) -> None:
+    """Test plain text response is handled correctly."""
+    aresponses.add(
+        "example.com",
+        "/",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "text/plain"},
+            text="ok",
+        ),
+    )
+    async with aiohttp.ClientSession() as session:
+        technove = TechnoVE("example.com", session=session)
+        response = await technove.request("/")
+        assert response == "ok"
+
+
 @pytest.mark.asyncio
 async def test_post_request(aresponses: ResponsesMockServer) -> None:
     """Test POST requests are handled correctly."""
@@ -159,8 +178,8 @@ async def test_http_error500(aresponses: ResponsesMockServer) -> None:
 
 
 @pytest.mark.asyncio
-async def test_empty_full_responses(aresponses: ResponsesMockServer) -> None:
-    """Test failure handling of full data request TechnoVE device state."""
+async def test_update_empty_responses(aresponses: ResponsesMockServer) -> None:
+    """Test failure handling of data request TechnoVE device state."""
     aresponses.add(
         "example.com",
         "/station/get/info",
@@ -171,17 +190,85 @@ async def test_empty_full_responses(aresponses: ResponsesMockServer) -> None:
             text="{}",
         ),
     )
+    async with aiohttp.ClientSession() as session:
+        technove = TechnoVE("example.com", session=session)
+        with pytest.raises(TechnoVEError):
+            await technove.update()
+
+
+@pytest.mark.asyncio
+async def test_update_partial_responses(aresponses: ResponsesMockServer) -> None:
+    """Test handling of data request TechnoVE device state."""
     aresponses.add(
         "example.com",
         "/station/get/info",
         "GET",
         aresponses.Response(
             status=200,
             headers={"Content-Type": "application/json"},
-            text="{}",
+            text='{"name":"testing"}',
         ),
     )
     async with aiohttp.ClientSession() as session:
         technove = TechnoVE("example.com", session=session)
-        with pytest.raises(TechnoVEError):
-            await technove.update()
+        station = await technove.update()
+        assert station.info.name == "testing"
+
+
+@pytest.mark.asyncio
+async def test_set_auto_charge(aresponses: ResponsesMockServer) -> None:
+    """Test that enabling auto_charge calls the right API."""
+    aresponses.add(
+        "example.com",
+        "/station/set/automatic",
+        "POST",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "plain/text"},
+            text="ok",
+        ),
+    )
+    async with aiohttp.ClientSession() as session:
+        technove = TechnoVE("example.com", session=session)
+        await technove.set_auto_charge(enabled=True)
+        aresponses.assert_plan_strictly_followed()
+
+
+@pytest.mark.asyncio
+async def test_set_charging_enabled(aresponses: ResponsesMockServer) -> None:
+    """Test that changing charging_enabled calls the right API."""
+    aresponses.add(
+        "example.com",
+        "/station/control/start",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "plain/text"},
+            text="ok",
+        ),
+    )
+    aresponses.add(
+        "example.com",
+        "/station/control/stop",
+        "GET",
+        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({"auto_charge": False})
+        await technove.set_charging_enabled(enabled=True)
+        await technove.set_charging_enabled(enabled=False)
+        aresponses.assert_plan_strictly_followed()
+
+
+@pytest.mark.asyncio
+async def test_set_charging_enabled_auto_charge() -> None:
+    """Test failure when enabling charging manually and auto-charge is enabled."""
+    technove = TechnoVE("example.com")
+    technove.station = Station({"auto_charge": True})
+    with pytest.raises(TechnoVEError):
+        await technove.set_charging_enabled(enabled=True)