Use websockets to update data (firstof9#9)

* fix: url and form data

* fix tests

* formatting

* feature: add ota_update, vehicle, and state properties

* start adding HTTP override support
* more tests

* Delete tests/__pycache__ directory

* move exceptions to their own file

* Add websocket support (firstof9#7)

* Add websocket listener

* Add auth to websocket connection

* version bump

* clean up (firstof9#8)

* Add websocket listener

* Add auth to websocket connection

* clean up

* Update setup.py

* fix ws url parsing

* Update setup.py

* update function starts websocket listener

* update tests and adjust functions

* switch to aiohttp - stage 1

* Update setup.py

* remove unneeded .json()

* fix typo

* adjust tests

* Update setup.py

* add more debugging

* Attempt to place listener in it's own event loop

* move websocket listener start out side of session

* create event loop for websocket

* new function to start the listener

* websockets

* make sure not to keep opening listeners if one is running

* only start a loop if it's not running

* better loop handling

* move loop detection

* more debugging

* better loop check

* output ws data as json

* parse data into status variable

* properly update dictionary

* don't parse data

* add callback

* add coordinator

* Update setup.py

* async

* clean up

* typing

* Revert "Merge branch 'main' into dev"

This reverts commit 8a3c855, reversing
changes made to fd07d7b.

Author: firstof9

openevsehttp/__init__.py

@@ -1,13 +1,16 @@
 """Provide a package for python-openevse-http."""
 from __future__ import annotations
 
+import asyncio
 import datetime
 import logging
-from typing import Optional
+from typing import Any, Callable, Optional
 
+import aiohttp  # type: ignore
 import requests  # type: ignore
 
 from .const import MAX_AMPS, MIN_AMPS
+from .exceptions import AuthenticationError, ParseJSONError, HTTPError
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -27,17 +30,132 @@
     255: "disabled",
 }
 
+ERROR_AUTH_FAILURE = "Authorization failure"
+ERROR_TOO_MANY_RETRIES = "Too many retries"
+ERROR_UNKNOWN = "Unknown"
 
-class AuthenticationError(Exception):
-    """Exception for authentication errors."""
+MAX_FAILED_ATTEMPTS = 5
 
+SIGNAL_CONNECTION_STATE = "websocket_state"
+STATE_CONNECTED = "connected"
+STATE_DISCONNECTED = "disconnected"
+STATE_STARTING = "starting"
+STATE_STOPPED = "stopped"
 
-class ParseJSONError(Exception):
-    """Exception for JSON parsing errors."""
 
+class OpenEVSEWebsocket:
+    """Represent a websocket connection to a OpenEVSE charger."""
 
-class HTTPError(Exception):
-    """Exception for HTTP errors."""
+    def __init__(
+        self,
+        server,
+        callback,
+        user=None,
+        password=None,
+    ):
+        """Initialize a OpenEVSEWebsocket instance."""
+        self.session = aiohttp.ClientSession()
+        self.uri = self._get_uri(server)
+        self._user = user
+        self._password = password
+        self.callback = callback
+        self._state = None
+        self.failed_attempts = 0
+        self._error_reason = None
+
+    @property
+    def state(self):
+        """Return the current state."""
+        return self._state
+
+    @state.setter
+    def state(self, value):
+        """Set the state."""
+        self._state = value
+        _LOGGER.debug("Websocket %s", value)
+        self.callback(SIGNAL_CONNECTION_STATE, value, self._error_reason)
+        self._error_reason = None
+
+    @staticmethod
+    def _get_uri(server):
+        """Generate the websocket URI."""
+        return server[: server.rfind("/")].replace("http", "ws") + "/ws"
+
+    async def running(self):
+        """Open a persistent websocket connection and act on events."""
+        self.state = STATE_STARTING
+        auth = None
+
+        if self._user and self._password:
+            auth = aiohttp.BasicAuth(self._user, self._password)
+
+        try:
+            async with self.session.ws_connect(
+                self.uri,
+                heartbeat=15,
+                auth=auth,
+            ) as ws_client:
+                self.state = STATE_CONNECTED
+                self.failed_attempts = 0
+
+                async for message in ws_client:
+                    if self.state == STATE_STOPPED:
+                        break
+
+                    if message.type == aiohttp.WSMsgType.TEXT:
+                        msg = message.json()
+                        msgtype = "data"
+                        self.callback(msgtype, msg, None)
+
+                    elif message.type == aiohttp.WSMsgType.CLOSED:
+                        _LOGGER.warning("Websocket connection closed")
+                        break
+
+                    elif message.type == aiohttp.WSMsgType.ERROR:
+                        _LOGGER.error("Websocket error")
+                        break
+
+        except aiohttp.ClientResponseError as error:
+            if error.code == 401:
+                _LOGGER.error("Credentials rejected: %s", error)
+                self._error_reason = ERROR_AUTH_FAILURE
+            else:
+                _LOGGER.error("Unexpected response received: %s", error)
+                self._error_reason = ERROR_UNKNOWN
+            self.state = STATE_STOPPED
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as error:
+            if self.failed_attempts >= MAX_FAILED_ATTEMPTS:
+                self._error_reason = ERROR_TOO_MANY_RETRIES
+                self.state = STATE_STOPPED
+            elif self.state != STATE_STOPPED:
+                retry_delay = min(2 ** (self.failed_attempts - 1) * 30, 300)
+                self.failed_attempts += 1
+                _LOGGER.error(
+                    "Websocket connection failed, retrying in %ds: %s",
+                    retry_delay,
+                    error,
+                )
+                self.state = STATE_DISCONNECTED
+                await asyncio.sleep(retry_delay)
+        except Exception as error:  # pylint: disable=broad-except
+            if self.state != STATE_STOPPED:
+                _LOGGER.exception("Unexpected exception occurred: %s", error)
+                self._error_reason = ERROR_UNKNOWN
+                self.state = STATE_STOPPED
+        else:
+            if self.state != STATE_STOPPED:
+                self.state = STATE_DISCONNECTED
+                await asyncio.sleep(5)
+
+    async def listen(self):
+        """Start the listening websocket."""
+        self.failed_attempts = 0
+        while self.state != STATE_STOPPED:
+            await self.running()
+
+    def close(self):
+        """Close the listening websocket."""
+        self.state = STATE_STOPPED
 
 
 class OpenEVSE:
@@ -47,57 +165,133 @@ def __init__(self, host: str, user: str = None, pwd: str = None) -> None:
         """Connect to an OpenEVSE charger equipped with wifi or ethernet."""
         self._user = user
         self._pwd = pwd
-        self._url = f"http://{host}"
-        self._status = None
-        self._config = None
+        self.url = f"http://{host}/"
+        self._status: dict = {}
+        self._config: dict = {}
         self._override = None
+        self._ws_listening = False
+        self.websocket: Optional[OpenEVSEWebsocket] = None
+        self.callback: Optional[Callable] = None
+        self._loop = None
 
-    def send_command(self, command: str) -> tuple | None:
+    async def send_command(self, command: str) -> tuple | None:
         """Send a RAPI command to the charger and parses the response."""
-        url = f"{self._url}/r"
+        auth = None
+        url = f"{self.url}r"
         data = {"json": 1, "rapi": command}
 
-        _LOGGER.debug("Posting data: %s to %s", command, url)
-        if self._user is not None:
-            value = requests.post(url, data=data, auth=(self._user, self._pwd))
-        else:
-            value = requests.post(url, data=data)
-
-        if value.status_code == 400:
-            _LOGGER.debug("JSON error: %s", value.text)
-            raise ParseJSONError
-        if value.status_code == 401:
-            _LOGGER.debug("Authentication error: %s", value)
-            raise AuthenticationError
-
-        if "ret" not in value.json():
-            return False, ""
-        resp = value.json()
-        return resp["cmd"], resp["ret"]
+        if self._user and self._pwd:
+            auth = aiohttp.BasicAuth(self._user, self._pwd)
 
-    def update(self) -> None:
+        _LOGGER.debug("Posting data: %s to %s", command, url)
+        async with aiohttp.ClientSession() as session:
+            async with session.post(url, data=data, auth=auth) as resp:
+                if resp.status == 400:
+                    _LOGGER.debug("JSON error: %s", await resp.text())
+                    raise ParseJSONError
+                if resp.status == 401:
+                    _LOGGER.debug("Authentication error: %s", await resp)
+                    raise AuthenticationError
+
+                value = await resp.json()
+
+                if "ret" not in value:
+                    return False, ""
+                return value["cmd"], value["ret"]
+
+    async def update(self) -> None:
         """Update the values."""
-        urls = [f"{self._url}/status", f"{self._url}/config"]
-
-        for url in urls:
-            _LOGGER.debug("Updating data from %s", url)
-            if self._user is not None:
-                value = requests.get(url, auth=(self._user, self._pwd))
-            else:
-                value = requests.get(url)
-
-            if value.status_code == 401:
-                _LOGGER.debug("Authentication error: %s", value)
-                raise AuthenticationError
-
-            if "/status" in url:
-                self._status = value.json()
-            else:
-                self._config = value.json()
+        auth = None
+        urls = [f"{self.url}config"]
+
+        if self._user and self._pwd:
+            auth = aiohttp.BasicAuth(self._user, self._pwd)
+
+        if not self._ws_listening:
+            urls = [f"{self.url}status", f"{self.url}config"]
+
+        async with aiohttp.ClientSession() as session:
+            for url in urls:
+                _LOGGER.debug("Updating data from %s", url)
+                async with session.get(url, auth=auth) as resp:
+                    if resp.status == 401:
+                        _LOGGER.debug("Authentication error: %s", resp)
+                        raise AuthenticationError
+
+                    if "/status" in url:
+                        self._status = await resp.json()
+                        _LOGGER.debug("Status update: %s", self._status)
+                    else:
+                        self._config = await resp.json()
+                        _LOGGER.debug("Config update: %s", self._config)
+
+        if not self.websocket:
+            # Start Websocket listening
+            self.websocket = OpenEVSEWebsocket(
+                self.url, self._update_status, self._user, self._pwd
+            )
+            if not self._ws_listening:
+                self._start_listening()
+
+    def _start_listening(self):
+        """Start the websocket listener."""
+        try:
+            _LOGGER.debug("Attempting to find running loop...")
+            self._loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._loop = asyncio.get_event_loop()
+            _LOGGER.debug("Using new event loop...")
+
+        if not self._ws_listening:
+            self._loop.create_task(self.websocket.listen())
+            pending = asyncio.all_tasks()
+            self._ws_listening = True
+            self._loop.run_until_complete(asyncio.gather(*pending))
+
+    def _update_status(self, msgtype, data, error):
+        """Update data from websocket listener."""
+        if msgtype == SIGNAL_CONNECTION_STATE:
+            if data == STATE_CONNECTED:
+                _LOGGER.debug("Websocket to %s successful", self.url)
+                self._ws_listening = True
+            elif data == STATE_DISCONNECTED:
+                _LOGGER.debug(
+                    "Websocket to %s disconnected, retrying",
+                    self.url,
+                )
+                self._ws_listening = False
+            # Stopped websockets without errors are expected during shutdown
+            # and ignored
+            elif data == STATE_STOPPED and error:
+                _LOGGER.error(
+                    "Websocket to %s failed, aborting [Error: %s]",
+                    self.url,
+                    error,
+                )
+                self._ws_listening = False
+
+        elif msgtype == "data":
+            _LOGGER.debug("ws_data: %s", data)
+            self._status.update(data)
+
+            if self.callback is not None:
+                self.callback()
+
+    def ws_disconnect(self) -> None:
+        """Disconnect the websocket listener."""
+        assert self.websocket
+        self.websocket.close()
+        self._ws_listening = False
+
+    @property
+    def ws_state(self) -> Any:
+        """Return the status of the websocket listener."""
+        assert self.websocket
+        return self.websocket.state
 
     def get_override(self) -> None:
         """Get the manual override status."""
-        url = f"{self._url}/overrride"
+        url = f"{self.url}/overrride"
 
         _LOGGER.debug("Geting data from %s", url)
         if self._user is not None:
@@ -121,7 +315,7 @@ def set_override(
         auto_release: bool = True,
     ) -> str:
         """Set the manual override status."""
-        url = f"{self._url}/overrride"
+        url = f"{self.url}/overrride"
 
         if state not in ["active", "disabled"]:
             raise ValueError
@@ -149,7 +343,7 @@ def set_override(
 
     def toggle_override(self) -> None:
         """Toggle the manual override status."""
-        url = f"{self._url}/overrride"
+        url = f"{self.url}/overrride"
 
         _LOGGER.debug("Toggling manual override %s", url)
         if self._user is not None:
@@ -167,7 +361,7 @@ def toggle_override(self) -> None:
 
     def clear_override(self) -> None:
         """Clear the manual override status."""
-        url = f"{self._url}/overrride"
+        url = f"{self.url}/overrride"
 
         _LOGGER.debug("Clearing manual overrride %s", url)
         if self._user is not None:

openevsehttp/exceptions.py

@@ -0,0 +1,13 @@
+"""Exceptions."""
+
+
+class AuthenticationError(Exception):
+    """Exception for authentication errors."""
+
+
+class ParseJSONError(Exception):
+    """Exception for JSON parsing errors."""
+
+
+class HTTPError(Exception):
+    """Exception for HTTP errors."""

pylintrc

@@ -26,7 +26,8 @@ disable=
   too-many-public-methods,
   too-many-instance-attributes,
   too-many-branches,
-  no-self-use
+  no-self-use,
+  too-many-statements
 
 [REPORTS]
 score=no

requirements_test.txt

@@ -1,4 +1,7 @@
 pytest==6.2.4
 pytest-cov==2.12.1
 pytest-timeout==1.4.2
+pytest-aiohttp
 requests_mock
+aiohttp
+aioresponses