diff --git a/examples/get_info.py b/examples/get_info.py new file mode 100644 index 0000000..97077c9 --- /dev/null +++ b/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()) diff --git a/examples/ruff.toml b/examples/ruff.toml new file mode 100644 index 0000000..2eb42d9 --- /dev/null +++ b/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 +] diff --git a/examples/set_auto_charge.py b/examples/set_auto_charge.py new file mode 100644 index 0000000..d4a5581 --- /dev/null +++ b/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()) diff --git a/src/technove/models.py b/src/technove/models.py index 348c452..375ea80 100644 --- a/src/technove/models.py +++ b/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), diff --git a/src/technove/technove.py b/src/technove/technove.py index 6f8893f..191bb9b 100644 --- a/src/technove/technove.py +++ b/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: diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..3f70c51 --- /dev/null +++ b/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) diff --git a/tests/test_technove.py b/tests/test_technove.py index f7176e8..e267169 100644 --- a/tests/test_technove.py +++ b/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,6 +190,15 @@ 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", @@ -178,10 +206,69 @@ async def test_empty_full_responses(aresponses: ResponsesMockServer) -> None: 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)