Add charge override deletion and fetching connection status

* Add getting connection status from API:
  * `Client.async_get_connection_status` - @mattrayner
* Add support for charge override deletion:
  * Add `Client.async_delete_charge_override` - @mattrayner

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Pod Point Client Changelog
 
+## v1.6.0
+
+* Add getting connection status from API:
+  * `Client.async_get_connection_status` - @mattrayner
+* Add support for charge override deletion:
+  * Add `Client.async_delete_charge_override` - @mattrayner
+
 ## v1.5.0
 
 * Add support for refreshing expired tokens, rather than grabbing new ones each time
@@ -61,15 +68,15 @@
 * Added additional testing dependencies - @mattrayner
 * Add CD pipeline, when a new tag/release is pushed, auto-publish to PyPi - @mattrayner
 
-## v0.3.0
+## v0.3.0
 
 * Add http_debug flag - @mattrayner
 * When enabled, complete response bodies will be sent to logger.debug
 * Restructured helpers and other classes so that they made more sense - @mattrayner
 * Completed a pylon pass to standardize the code base - @mattrayner
 * Improved test coverage - @mattrayner
 
-## v0.2.2
+## v0.2.2
 
 * Make timestamp=XXX optional, and off by default
 * Greatly improve test coverage

example.py

@@ -63,6 +63,27 @@ async def main(username: str, password: str, http_debug: bool = False, loop=None
     energy_used = charges[0].kwh_used
     print(f"  kW charged: {energy_used}")
 
+    # Set charge override
+    print(f"Setting 'Charge now' for pod {pod.ppid}")
+    override = await client.async_set_charge_override(pod=pod, hours=1)
+    print(f"  Override until: {override.ends_at}")
+
+    # Delete override
+    print(f"Deleting 'Charge now' for pod {pod.ppid}")
+    await client.async_delete_charge_override(pod=pod)
+    print("  Done")
+
+    # Get charge override
+    print(f"Attempting to get charge override for pod {pod.ppid}")
+    override = await client.async_get_charge_override(pod=pod)
+    print(f"  Override ends at: {override.ends_at}")
+
+    # Get connectivity status
+    print(f"Getting connectivity status for pod {pod.ppid}")
+    connectivity = await client.async_get_connectivity_status(pod=pod)
+    print(f"  Connectivity status: {connectivity.evses[0].connectivity_state.connectivity_status}")
+    print(f"  Last message at: {connectivity.evses[0].connectivity_state.last_message_at}")
+
     # Expire token and exchange a refresh
     print("Expiring token and refreshing...")
     client.auth.access_token_expiry = datetime.now() - timedelta(minutes=10)

podpointclient/client.py

@@ -5,15 +5,16 @@
 
 import aiohttp
 
-from .endpoints import API_BASE_URL, CHARGE_SCHEDULES, PODS, UNITS, USERS, CHARGES, FIRMWARE, AUTH, CHARGE_OVERRIDE
+from .endpoints import API_BASE_URL, CHARGE_SCHEDULES, PODS, UNITS, USERS, CHARGES, FIRMWARE, AUTH, CHARGE_OVERRIDE, CHARGERS, CONNECTIVITY_STATUS, MOBILE_API_BASE_URL
 from .helpers.auth import Auth
 from .helpers.functions import auth_headers
 from .helpers.api_wrapper import APIWrapper
-from .factories import PodFactory, ScheduleFactory, ChargeFactory, FirmwareFactory, UserFactory, ChargeOverrideFactory
+from .factories import PodFactory, ScheduleFactory, ChargeFactory, FirmwareFactory, UserFactory, ChargeOverrideFactory, ConnectivityStatusFactory
 from .pod import Pod, Firmware
 from .charge import Charge
 from .charge_mode import ChargeMode
 from .charge_override import ChargeOverride
+from .connectivity_status import ConnectivityStatus
 from .schedule import Schedule
 from .user import User
 from .errors import ChargeOverrideValidationError
@@ -251,6 +252,33 @@ async def async_get_charge_override(self, pod: Pod) -> Union[None, ChargeOverrid
 
         return ChargeOverrideFactory().build_charge_override(charge_override_response=json)
 
+    async def async_delete_charge_override(self, pod:Pod) -> bool:
+        await self.auth.async_update_access_token()
+
+        response = await self.api_wrapper.delete(
+            url=self._url_from_path(
+                path=f"{UNITS}/{pod.unit_id}{CHARGE_OVERRIDE}"),
+            params=self._generate_complete_params(params=None),
+            headers=auth_headers(access_token=self.auth.access_token)
+        )
+
+        return response.status == 204
+
+    async def async_get_connectivity_status(self, pod:Pod) -> ConnectivityStatus:
+        await self.auth.async_update_access_token()
+
+        response = await self.api_wrapper.get(
+            url=self._url_from_path(
+                path=f"{CHARGERS}/{pod.ppid}{CONNECTIVITY_STATUS}",
+                base=MOBILE_API_BASE_URL
+            ),
+            params=self._generate_complete_params(params=None),
+            headers=auth_headers(access_token=self.auth.access_token)
+        )
+
+        json = await self._handle_json_response(response=response)
+
+        return ConnectivityStatusFactory().build_connectivity_status(connectivity_status_response=json)
 
     async def async_set_charge_override(self, pod:Pod, hours:int=0, minutes:int=0, seconds:int=0) -> ChargeOverride:
         await self.auth.async_update_access_token()
@@ -348,9 +376,9 @@ def _schedule_data(self, enabled: bool) -> Dict[str, Any]:
 
         return {"data": d_list}
 
-    def _url_from_path(self, path: str) -> str:
+    def _url_from_path(self, path: str, base: str = API_BASE_URL) -> str:
         """Given a path, return a complete API URL"""
-        return f"{API_BASE_URL}{path}"
+        return f"{base}{path}"
 
     def _generate_complete_params(self, params: Union[None, Dict[str, Any]]) -> Dict[str, any]:
         """Given a params object, add optional params if required"""

podpointclient/connectivity_status.py

@@ -0,0 +1,166 @@
+"""Connectivity State class, represents a 'Connectivity State' from the podpoint apis"""
+
+from datetime import datetime, timedelta
+from typing import Dict, Any, List
+from dataclasses import dataclass, field
+from .helpers.functions import lazy_convert_to_datetime, lazy_iso_format_datetime
+import json
+
+# {
+# 	"ppid": "PSL-266056",
+# 	"evses": [{
+# 		"id": 1,
+# 		"connectivityState": {
+# 			"protocol": "POW",
+# 			"connectivityStatus": "ONLINE",
+# 			"signalStrength": -68,
+# 			"lastMessageAt": "2024-04-05T18:36:29Z",
+# 			"connectionStartedAt": "2024-04-05T18:26:26.819Z",
+# 			"connectionQuality": 3
+# 		},
+# 		"connectors": [{
+# 			"id": 1,
+# 			"door": "A",
+# 			"chargingState": "SUSPENDED_EV"
+# 		}],
+# 		"architecture": "arch3",
+# 		"energyOfferStatus": {
+# 			"isOfferingEnergy": true,
+# 			"reason": "CHARGE_SCHEDULE",
+# 			"until": null,
+# 			"randomDelay": null,
+# 			"doNotCache": false
+# 		}
+# 	}],
+# 	"connectedComponents": ["evses"]
+# }
+
+@dataclass
+class Evse:
+    """Represents a Location within a charge from pod point"""
+
+    def __init__(self, data: Dict[str, Any]):
+        self.id: int = data.get('id', None)
+        self.architecture: str = data.get('architecture', None)
+
+        connectivity_state_data = data.get('connectivityState', {})
+        self.connectivity_state = self.ConnectivityState(data=connectivity_state_data)
+
+        connectors_data = data.get('connectors', [])
+        self.connectors = []
+        for connector in connectors_data:
+            self.connectors.append(self.Connector(data=connector))
+
+        energy_offer_status_data = data.get('energyOfferStatus', {})
+        self.energy_offer_status = self.EnergyOfferStatus(data=energy_offer_status_data)
+
+    @property
+    def dict(self):
+        return {
+            "id": self.id,
+            "architecture": self.architecture,
+            "connectivityState": self.connectivity_state.dict,
+            "connectors": [connector.dict for connector in self.connectors],
+            "energyOfferStatus": self.energy_offer_status.dict
+        }
+
+    def to_json(self):
+        """JSON representation of a ConnectivityState object"""
+        return json.dumps(self.dict, ensure_ascii=False)
+
+    @dataclass
+    class ConnectivityState:
+        """Represents a Location within a charge from pod point"""
+
+        def __init__(self, data: Dict[str, Any]):
+            self.protocol: str = data.get('protocol', None)
+            self.connectivity_status: str = data.get('connectivityStatus', None)
+            self.signal_strength:int = data.get('signalStrength', None)
+            self.last_message_at: datetime = lazy_convert_to_datetime(data.get('lastMessageAt', None))
+            self.connection_started_at: datetime = lazy_convert_to_datetime(data.get('connectionStartedAt', None))
+            self.connection_quality: int = data.get('connectionQuality', None)
+
+        @property
+        def dict(self):
+            return {
+                "protocol": self.protocol,
+                "connectivityStatus": self.connectivity_status,
+                "signalStrength": self.signal_strength,
+                "lastMessageAt": lazy_iso_format_datetime(self.last_message_at),
+                "connectionStartedAt": lazy_iso_format_datetime(self.connection_started_at),
+                "connectionQuality": self.connection_quality
+            }
+
+        def to_json(self):
+            """JSON representation of a ConnectivityState object"""
+            return json.dumps(self.dict, ensure_ascii=False)
+
+    @dataclass
+    class Connector:
+        """Represents a Location within a charge from pod point"""
+
+        def __init__(self, data: Dict[str, Any]):
+            self.id: int = data.get('id', None)
+            self.door: str = data.get('door', None)
+            self.charging_state: str = data.get('chargingState', None)
+
+        @property
+        def dict(self):
+            return {
+                "id": self.id,
+                "door": self.door,
+                "chargingState": self.charging_state
+            }
+
+    def to_json(self):
+        """JSON representation of a ConnectivityState object"""
+        return json.dumps(self.dict, ensure_ascii=False)
+
+    @dataclass
+    class EnergyOfferStatus:
+        """Represents a Location within a charge from pod point"""
+
+        def __init__(self, data: Dict[str, Any]):
+            self.is_offering_energy: bool = data.get('isOfferingEnergy', None)
+            self.reason: str = data.get('reason', None)
+            self.until: datetime = lazy_convert_to_datetime(data.get('until', None))
+            self.random_delay = data.get('randomDelay', None)
+            self.do_not_cache: bool = data.get('doNotCache', None)
+
+        @property
+        def dict(self):
+            return {
+                "isOfferingEnergy": self.is_offering_energy,
+                "reason": self.reason,
+                "until": lazy_iso_format_datetime(self.until),
+                "randomDelay": self.random_delay,
+                "doNotCache": self.do_not_cache
+            }
+
+        def to_json(self):
+            """JSON representation of a ConnectivityState object"""
+            return json.dumps(self.dict, ensure_ascii=False)
+
+
+class ConnectivityStatus:
+    """Representation of a Connectivity State from pod point"""
+
+    def __init__(self, data: Dict[str, Any]):
+        self.ppid: int = data.get('ppid', None)
+        self.connected_components: List[str] = data.get('connectedComponents', [])
+
+        self.evses: List[Evse] = []
+        for evse in data.get('evses', []):
+            self.evses.append(Evse(data=evse))
+
+    @property
+    def dict(self) -> Dict[str, Any]:
+        return {
+            "ppid": self.ppid,
+            "connected_components": self.connected_components,
+            "evses": [evse.dict for evse in self.evses]
+        }
+
+    def to_json(self):
+        """JSON representation of a ConnectivityState object"""
+        return json.dumps(self.dict, ensure_ascii=False)

podpointclient/endpoints.py

@@ -8,12 +8,17 @@
 CHARGE_SCHEDULES = '/charge-schedules'
 CHARGES = '/charges'
 CHARGE_OVERRIDE = '/charge-override'
+CHARGERS = '/chargers'
+CONNECTIVITY_STATUS = '/connectivity-status'
 FIRMWARE = '/firmware'
 
-API_BASE = 'mobile-api.pod-point.com/api3/'
+MOBILE_API_BASE = 'mobile-api.pod-point.com'
+API_BASE = f"{MOBILE_API_BASE}/api3/"
 API_VERSION = 'v5'
 API_BASE_URL = f"https://{API_BASE}{API_VERSION}"
 
+MOBILE_API_BASE_URL = f"https://{MOBILE_API_BASE}"
+
 """Google endpoint, used for auth"""
 GOOGLE_KEY = '?key=AIzaSyCwhF8IOl_7qHXML0pOd5HmziYP46IZAGU'
 PASSWORD_VERIFY = f"/verifyPassword{GOOGLE_KEY}"

podpointclient/factories.py

@@ -5,6 +5,7 @@
 from .schedule import Schedule, ScheduleStatus
 from .charge import Charge
 from .charge_override import ChargeOverride
+from .connectivity_status import ConnectivityStatus
 
 class PodFactory:
     """Factory for creating Pod objects"""
@@ -96,3 +97,12 @@ def build_user(self, user_response: Dict[str, Any]) -> User:
             return None
 
         return User(data=user_data)
+
+class ConnectivityStatusFactory:
+    """Factory  for creating ConnectivityStatus objects"""
+    def build_connectivity_status(self, connectivity_status_response: Dict[str, Any]):
+        """Build a ConnectivityStatus object based off of a response from pod point"""
+        if connectivity_status_response is None:
+            return None
+
+        return ConnectivityStatus(data=connectivity_status_response)

podpointclient/version.py

@@ -1,3 +1,3 @@
 """Version for the podpointclient library"""
 
-__version__ = "1.5.0"
+__version__ = "1.6.0a1"

tests/fixtures/connectivity_status.json

@@ -0,0 +1,28 @@
+{
+  "ppid": "PSL-123456",
+  "evses": [{
+    "id": 1,
+    "connectivityState": {
+      "protocol": "POW",
+      "connectivityStatus": "ONLINE",
+      "signalStrength": -68,
+      "lastMessageAt": "2024-04-05T18:26:28Z",
+      "connectionStartedAt": "2024-04-05T18:26:26.819Z",
+      "connectionQuality": 3
+    },
+    "connectors": [{
+      "id": 1,
+      "door": "A",
+      "chargingState": "SUSPENDED_EV"
+    }],
+    "architecture": "arch3",
+    "energyOfferStatus": {
+      "isOfferingEnergy": true,
+      "reason": "CHARGE_SCHEDULE",
+      "until": null,
+      "randomDelay": null,
+      "doNotCache": false
+    }
+  }],
+  "connectedComponents": ["evses"]
+}

tests/helpers.py

@@ -81,6 +81,9 @@ def firmware_response(self):
     def user_response(self):
         return self.__json_load_fixture('complete_user')
 
+    def connectivity_status_response(self):
+        return self.__json_load_fixture('connectivity_status')
+
     def __json_load_fixture(self, fixture_name: str):
         file_location = os.path.dirname(__file__)
         path = f'fixtures/{fixture_name}.json'