Added isort to format and typing (openviess#168)

* isort imports

* Add linter, addtional format and initial type hints

* fmt

PyViCare/PyViCare.py

@@ -1,9 +1,11 @@
+import logging
+
+from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
 from PyViCare.PyViCareBrowserOAuthManager import ViCareBrowserOAuthManager
+from PyViCare.PyViCareCachedService import ViCareCachedService
 from PyViCare.PyViCareDeviceConfig import PyViCareDeviceConfig
 from PyViCare.PyViCareOAuthManager import ViCareOAuthManager
 from PyViCare.PyViCareService import ViCareDeviceAccessor, ViCareService
-from PyViCare.PyViCareCachedService import ViCareCachedService
-import logging
 
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())
@@ -12,24 +14,22 @@
 
 
 class PyViCare:
-    def __init__(self):
+    def __init__(self) -> None:
         self.cacheDuration = 60
 
     def setCacheDuration(self, cache_duration):
         self.cacheDuration = cache_duration
 
-    def initWithCredentials(self, username, password, client_id, token_file):
-        self.oauth_manager = ViCareOAuthManager(
-            username, password, client_id, token_file)
-        self.__loadInstallations()
+    def initWithCredentials(self, username: str, password: str, client_id: str, token_file: str):
+        self.initWithExternalOAuth(ViCareOAuthManager(
+            username, password, client_id, token_file))
 
-    def initWithExternalOAuth(self, oauth_manager):
+    def initWithExternalOAuth(self, oauth_manager: AbstractViCareOAuthManager) -> None:
         self.oauth_manager = oauth_manager
         self.__loadInstallations()
 
-    def initWithBrowserOAuth(self, client_id, token_file):
-        self.oauth_manager = ViCareBrowserOAuthManager(client_id, token_file)
-        self.__loadInstallations()
+    def initWithBrowserOAuth(self, client_id: str, token_file: str) -> None:
+        self.initWithExternalOAuth(ViCareBrowserOAuthManager(client_id, token_file))
 
     def __buildService(self, accessor):
         if self.cacheDuration > 0:

PyViCare/PyViCareAbstractOAuthManager.py

@@ -1,8 +1,12 @@
-from PyViCare import Feature
-from PyViCare.PyViCareUtils import PyViCareRateLimitError, PyViCareCommandError
-from abc import abstractclassmethod
-from oauthlib.oauth2 import TokenExpiredError
 import logging
+from abc import abstractclassmethod
+from typing import Any
+
+from oauthlib.oauth2 import TokenExpiredError  # type: ignore
+from requests_oauthlib.oauth2_session import OAuth2Session
+
+from PyViCare import Feature
+from PyViCare.PyViCareUtils import PyViCareCommandError, PyViCareRateLimitError
 
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())
@@ -11,21 +15,21 @@
 
 
 class AbstractViCareOAuthManager:
-    def __init__(self, oauth_session):
+    def __init__(self, oauth_session: OAuth2Session) -> None:
         self.__oauth = oauth_session
 
     @property
-    def oauth_session(self):
+    def oauth_session(self) -> OAuth2Session:
         return self.__oauth
 
-    def replace_session(self, new_session):
+    def replace_session(self, new_session: OAuth2Session) -> None:
         self.__oauth = new_session
 
     @abstractclassmethod
     def renewToken(self):
         return
 
-    def get(self, url):
+    def get(self, url: str) -> Any:
         try:
             logger.debug(self.__oauth)
             response = self.__oauth.get(f"{API_BASE_URL}{url}").json()
@@ -69,7 +73,7 @@ def __handle_command_error(self, response):
         json representation of the answer
     """
 
-    def post(self, url, data):
+    def post(self, url, data) -> Any:
         headers = {"Content-Type": "application/json",
                    "Accept": "application/vnd.siren+json"}
         try:

PyViCare/PyViCareBrowserOAuthManager.py

@@ -1,14 +1,17 @@
-from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
-from PyViCare.PyViCareUtils import PyViCareBrowserOAuthTimeoutReachedError, PyViCareInvalidCredentialsError
-import requests
-import re
 import json
+import logging
 import os
-import pkce
+import re
+import webbrowser
 from http.server import BaseHTTPRequestHandler, HTTPServer
-import logging
+
+import pkce
+import requests
 from requests_oauthlib import OAuth2Session
-import webbrowser
+
+from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
+from PyViCare.PyViCareUtils import (PyViCareBrowserOAuthTimeoutReachedError,
+                                    PyViCareInvalidCredentialsError)
 
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())
@@ -34,7 +37,7 @@ def do_GET(self):
             self.end_headers()
             self.wfile.write(text.encode("utf-8"))
 
-    def __init__(self, client_id, token_file):
+    def __init__(self, client_id: str, token_file: str) -> None:
 
         self.token_file = token_file
         self.client_id = client_id
@@ -124,7 +127,7 @@ def __build_oauth_session(self, result, after_redirect):
         self.__storeToken(result)
         return OAuth2Session(client_id=self.client_id, token=result)
 
-    def renewToken(self):
+    def renewToken(self) -> None:
         token = self.oauth_session.token
         result = requests.post(url=TOKEN_URL, data={
             'grant_type': 'refresh_token',

PyViCare/PyViCareCachedService.py

@@ -1,24 +1,28 @@
-from datetime import datetime
 import threading
-from PyViCare.PyViCareService import ViCareService, readFeature
+from datetime import datetime
+from typing import Any
+
+from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
+from PyViCare.PyViCareService import (ViCareDeviceAccessor, ViCareService,
+                                      readFeature)
 
 
 class ViCareTimer:
     # class is used to replace logic in unittest
-    def now(self):
+    def now(self) -> datetime:
         return datetime.now()
 
 
 class ViCareCachedService(ViCareService):
 
-    def __init__(self, oauth_manager, accessor, cacheDuration):
+    def __init__(self, oauth_manager: AbstractViCareOAuthManager, accessor: ViCareDeviceAccessor, cacheDuration: int) -> None:
         ViCareService.__init__(self, oauth_manager, accessor)
         self.__cacheDuration = cacheDuration
         self.__cache = None
         self.__cacheTime = None
         self.__lock = threading.Lock()
 
-    def getProperty(self, property_name):
+    def getProperty(self, property_name: str) -> Any:
         data = self.__get_or_update_cache()
         entities = data["data"]
         return readFeature(entities, property_name)
@@ -35,7 +39,7 @@ def __get_or_update_cache(self):
                 self.__cacheTime = ViCareTimer().now()
             return self.__cache
 
-    def is_cache_invalid(self):
+    def is_cache_invalid(self) -> bool:
         return self.__cache is None or self.__cacheTime is None or (ViCareTimer().now() - self.__cacheTime).seconds > self.__cacheDuration
 
     def clear_cache(self):

PyViCare/PyViCareDevice.py

@@ -1,6 +1,10 @@
 import logging
-from datetime import datetime, time
-from PyViCare.PyViCareUtils import PyViCareNotSupportedFeatureError, handleNotSupported, handleAPICommandErrors
+from datetime import datetime
+from typing import Any, Callable, List, Optional
+
+from PyViCare.PyViCareService import ViCareService
+from PyViCare.PyViCareUtils import (PyViCareNotSupportedFeatureError,
+                                    handleAPICommandErrors, handleNotSupported)
 
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())
@@ -9,10 +13,10 @@
 VICARE_DHW_TEMP2 = "temp-2"
 
 
-def isSupported(method):
+def isSupported(method: Callable) -> bool:
     try:
         result = method()
-        return result != 'error'
+        return bool(result != 'error')
     except PyViCareNotSupportedFeatureError:
         return False
 
@@ -23,11 +27,11 @@ class Device:
     Note that currently, a new token is generate for each run.
     """
 
-    def __init__(self, service):
+    def __init__(self, service: ViCareService) -> None:
         self.service = service
 
     @property
-    def circuits(self):
+    def circuits(self) -> List[Any]:
         return list([self.getCircuit(x) for x in self.getAvailableCircuits()])
 
     def getCircuit(self, circuit):
@@ -59,8 +63,8 @@ def getDomesticHotWaterActiveMode(self):
 
         mode = None
         for s in schedule[current_day]:
-            startTime = time.fromisoformat(s["start"])
-            endTime = time.fromisoformat(s["end"])
+            startTime = datetime.strptime(s["start"], '%H:%M').time()
+            endTime = datetime.strptime(s["end"], '%H:%M').time()
             if startTime <= currentTime and currentTime <= endTime:
                 if s["mode"] == VICARE_DHW_TEMP2:  # temp-2 overrides all other modes
                     return s["mode"]
@@ -228,8 +232,8 @@ def getDomesticHotWaterCirculationMode(self):
             return None  # no schedule for day configured
 
         for s in schedule[current_day]:
-            startTime = time.fromisoformat(s["start"])
-            endTime = time.fromisoformat(s["end"])
+            startTime = datetime.strptime(s["start"], '%H:%M').time()
+            endTime = datetime.strptime(s["end"], '%H:%M').time()
             if startTime <= currentTime and currentTime <= endTime:
                 return s["mode"]
         return schedule['default_mode']
@@ -248,13 +252,13 @@ def getBoilerSerial(self):
 
 
 class DeviceWithCircuit:
-    def __init__(self, device, circuit):
+    def __init__(self, device: Device, circuit: str) -> None:
         self.service = device.service
         self.circuit = circuit
         self.device = device
 
     @property
-    def id(self):
+    def id(self) -> str:
         return self.circuit
 
     """ Set the active mode
@@ -414,7 +418,7 @@ def getHeatingSchedule(self):
 
     # Calculates target supply temperature based on data from Viessmann
     # See: https://www.viessmann-community.com/t5/Gas/Mathematische-Formel-fuer-Vorlauftemperatur-aus-den-vier/m-p/68890#M27556
-    def getTargetSupplyTemperature(self):
+    def getTargetSupplyTemperature(self) -> Optional[float]:
         if(not isSupported(self.getCurrentDesiredTemperature)
                 or not isSupported(self.device.getOutsideTemperature)
                 or not isSupported(self.getHeatingCurveShift)
@@ -429,4 +433,4 @@ def getTargetSupplyTemperature(self):
         targetSupply = (inside + shift - slope * delta_outside_inside
                         * (1.4347 + 0.021 * delta_outside_inside + 247.9
                             * pow(10, -6) * pow(delta_outside_inside, 2)))
-        return round(targetSupply, 1)
+        return float(round(targetSupply, 1))

PyViCare/PyViCareDeviceConfig.py

@@ -1,12 +1,14 @@
+import json
+import logging
+import re
+
 from PyViCare.PyViCareDevice import Device
-from PyViCare.PyViCareGazBoiler import GazBoiler
 from PyViCare.PyViCareFuelCell import FuelCell
+from PyViCare.PyViCareGazBoiler import GazBoiler
 from PyViCare.PyViCareHeatPump import HeatPump
 from PyViCare.PyViCareOilBoiler import OilBoiler
 from PyViCare.PyViCarePelletsBoiler import PelletsBoiler
-import re
-import logging
-import json
+
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())
 

PyViCare/PyViCareOAuthManager.py

@@ -1,14 +1,16 @@
-from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
-from PyViCare.PyViCareUtils import PyViCareInvalidCredentialsError
-import requests
-import re
-import pickle
+import logging
 import os
-import pkce
+import pickle
+import re
+from contextlib import suppress
 from pickle import UnpicklingError
+
+import pkce
+import requests
 from requests_oauthlib import OAuth2Session
-import logging
-from contextlib import suppress
+
+from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
+from PyViCare.PyViCareUtils import PyViCareInvalidCredentialsError
 
 logger = logging.getLogger('ViCare')
 logger.addHandler(logging.NullHandler())

PyViCare/PyViCareService.py

@@ -1,5 +1,8 @@
-import logging
 import json
+import logging
+from typing import Any
+
+from PyViCare.PyViCareAbstractOAuthManager import AbstractViCareOAuthManager
 from PyViCare.PyViCareUtils import PyViCareNotSupportedFeatureError
 
 logger = logging.getLogger('ViCare')
@@ -25,29 +28,29 @@ def buildGetPropertyUrl(accessor, property_name):
 
 
 class ViCareDeviceAccessor:
-    def __init__(self, id, serial, device_id):
+    def __init__(self, id: int, serial: str, device_id: str) -> None:
         self.id = id
         self.serial = serial
         self.device_id = device_id
 
 
 class ViCareService:
-    def __init__(self, oauth_manager, accessor):
+    def __init__(self, oauth_manager: AbstractViCareOAuthManager, accessor: ViCareDeviceAccessor) -> None:
         self.oauth_manager = oauth_manager
         self.accessor = accessor
 
-    def getProperty(self, property_name):
+    def getProperty(self, property_name: str) -> Any:
         url = buildGetPropertyUrl(
             self.accessor, property_name)
         return self.oauth_manager.get(url)
 
-    def setProperty(self, property_name, action, data):
+    def setProperty(self, property_name: str, action: str, data: Any) -> Any:
         url = buildSetPropertyUrl(
             self.accessor, property_name, action)
 
         post_data = data if isinstance(data, str) else json.dumps(data)
         return self.oauth_manager.post(url, post_data)
 
-    def fetch_all_features(self):
+    def fetch_all_features(self) -> Any:
         url = f'/equipment/installations/{self.accessor.id}/gateways/{self.accessor.serial}/devices/{self.accessor.device_id}/features/'
         return self.oauth_manager.get(url)

PyViCare/PyViCareUtils.py

@@ -1,14 +1,16 @@
-from PyViCare import Feature
 import datetime
 from functools import wraps
+from typing import Callable
+
+from PyViCare import Feature
 
 # This decorator handles access to underlying JSON properties.
 # If the property is not found (KeyError) or the index does not
 # exists (IndexError), the requested feature is not supported by
 # the device.
 
 
-def handleNotSupported(func):
+def handleNotSupported(func: Callable) -> Callable:
     @wraps(func)
     def wrapper(*args, **kwargs):
         try:
@@ -28,7 +30,7 @@ def feature_flag_wrapper(*args, **kwargs):
     return feature_flag_wrapper
 
 
-def handleAPICommandErrors(func):
+def handleAPICommandErrors(func: Callable) -> Callable:
     @wraps(func)
     def wrapper(*args, **kwargs):
         try:

format.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
+isort .
 autopep8 --in-place --recursive .
 npx prettier --write {**.md,**.yml,**.json}
-flake8