Address pyLint cleanup and minor bug fixes

.github/workflows/pylint.yml

@@ -0,0 +1,24 @@
+name: Pylint
+
+on: [push]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pylint
+        pip install -r requirements.txt
+    - name: Analyzing the code with pylint
+      run: |
+        pylint -E pypowerwall/*.py

.pylintrc

@@ -0,0 +1,5 @@
+[MESSAGES CONTROL]
+disable=consider-iterating-dictionary, consider-swap-variables, consider-using-enumerate, cyclic-import, consider-using-max-builtin, no-else-continue, consider-using-min-builtin, consider-using-in, super-with-arguments, protected-access, import-outside-toplevel, multiple-statements, unidiomatic-typecheck, no-else-break, import-error, invalid-name, missing-docstring, no-else-return, no-member, too-many-lines, line-too-long, too-many-ancestors, too-many-arguments, too-many-branches, too-many-instance-attributes, too-many-locals, too-many-nested-blocks, too-many-return-statements, too-many-statements, too-few-public-methods, ungrouped-imports, use-dict-literal, superfluous-parens, fixme, consider-using-f-string, bare-except, broad-except, unused-variable, unspecified-encoding, redefined-builtin, consider-using-dict-items, redundant-u-string-prefix, useless-object-inheritance, wrong-import-position, logging-not-lazy, logging-fstring-interpolation, wildcard-import, logging-format-interpolation
+
+[SIMILARITIES]
+min-similarity-lines=8

RELEASE.md

@@ -1,5 +1,10 @@
 # RELEASE NOTES
 
+## v0.10.6 - pyLint Cleanup
+
+* Minor Bug Fixes - TEDAPI get_reserve() fix to address unscaled results.
+* pyLint Cleanup of Code
+
 ## v0.10.5 - Minor Fixes
 
 * Fix for TEDAPI "full" (e.g. Powerwall 3) mode, including `grid_status` bug resulting in false reports of grid status, `level()` bug where data gap resulted in 0% state of charge and `alerts()` where data gap from tedapi resulted in a `null` alert.

proxy/RELEASE.md

@@ -1,5 +1,9 @@
 ## pyPowerwall Proxy Release Notes
 
+### Proxy t63 (15 Jun 2024)
+
+* pyLint Code Cleaning
+
 ### Proxy t62 (13 Jun 2024)
 
 * Add battery full_pack and remaining energy data to `/pod` API call for all cases.

proxy/server.py

@@ -48,14 +48,13 @@
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from socketserver import ThreadingMixIn
 from typing import Optional
+from urllib.parse import urlparse, parse_qs
 
+from transform import get_static, inject_js
 import pypowerwall
 from pypowerwall import parse_version
-from pypowerwall.fleetapi.fleetapi import CONFIGFILE
-from transform import get_static, inject_js
-from urllib.parse import urlparse, parse_qs
 
-BUILD = "t62"
+BUILD = "t63"
 ALLOWLIST = [
     '/api/status', '/api/site_info/site_name', '/api/meters/site',
     '/api/meters/solar', '/api/sitemaster', '/api/powerwalls',
@@ -71,7 +70,7 @@
 ]
 web_root = os.path.join(os.path.dirname(__file__), "web")
 
-# Configuration for Proxy - Check for environmental variables 
+# Configuration for Proxy - Check for environmental variables
 #    and always use those if available (required for Docker)
 bind_address = os.getenv("PW_BIND_ADDRESS", "")
 password = os.getenv("PW_PASSWORD", "")
@@ -172,7 +171,7 @@ def get_value(a, key):
     pw = pypowerwall.Powerwall(host, password, email, timezone, cache_expire,
                                timeout, pool_maxsize, siteid=siteid,
                                authpath=authpath, authmode=authmode,
-                               cachefile=cachefile, auto_select=True, 
+                               cachefile=cachefile, auto_select=True,
                                retry_modes=True, gw_pwd=gw_pwd)
 except Exception as e:
     log.error(e)
@@ -231,9 +230,9 @@ def get_value(a, key):
 
 class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
     daemon_threads = True
-    pass
 
 
+# pylint: disable=arguments-differ,global-variable-not-assigned
 # noinspection PyPep8Naming
 class Handler(BaseHTTPRequestHandler):
     def log_message(self, log_format, *args):
@@ -246,7 +245,7 @@ def address_string(self):
         # replace function to avoid lookup delays
         hostaddr, hostport = self.client_address[:2]
         return hostaddr
-    
+
     def do_POST(self):
         global proxystats
         contenttype = 'application/json'
@@ -264,9 +263,9 @@ def do_POST(self):
                     query_params = parse_qs(post_data.decode('utf-8'))
                     value = query_params.get('value', [''])[0]
                     token = query_params.get('token', [''])[0]
-                except Exception as e:
+                except Exception as er:
                     message = '{"error": "Control Command Error: Invalid Request"}'
-                    log.error(f"Control Command Error: {e}")
+                    log.error(f"Control Command Error: {er}")
                 if not message:
                     # Check if unable to connect to cloud
                     if pw_control.client is None:
@@ -368,7 +367,7 @@ def do_GET(self):
             proxystats['clear'] = int(time.time())
             message: str = json.dumps(proxystats)
         elif self.path == '/temps':
-            # Temps of Powerwalls 
+            # Temps of Powerwalls
             message: str = pw.temps(jsonformat=True) or json.dumps({})
         elif self.path == '/temps/pw':
             # Temps of Powerwalls with Simple Keys
@@ -605,6 +604,7 @@ def do_GET(self):
                 self.send_header("Set-Cookie", f"UserRecord={pw.client.auth['UserRecord']};{cookiesuffix}")
 
             # Serve static assets from web root first, if found.
+            # pylint: disable=attribute-defined-outside-init
             if self.path == "/" or self.path == "":
                 self.path = "/index.html"
                 fcontent, ftype = get_static(web_root, self.path)
@@ -654,7 +654,7 @@ def do_GET(self):
                         )
                     fcontent = r.content
                     ftype = r.headers['content-type']
-                except AttributeError as e:
+                except AttributeError:
                     # Display 404
                     log.debug("File not found: {}".format(self.path))
                     fcontent = bytes("Not Found", 'utf-8')
@@ -712,14 +712,15 @@ def do_GET(self):
     if https_mode == "yes":
         # Activate HTTPS
         log.debug("Activating HTTPS")
+        # pylint: disable=deprecated-method
         server.socket = ssl.wrap_socket(server.socket,
                                         certfile=os.path.join(os.path.dirname(__file__), 'localhost.pem'),
                                         server_side=True, ssl_version=ssl.PROTOCOL_TLSv1_2, ca_certs=None,
                                         do_handshake_on_connect=True)
 
     # noinspection PyBroadException
     try:
-        server.serve_forever()        
+        server.serve_forever()
     except (Exception, KeyboardInterrupt, SystemExit):
         print(' CANCEL \n')
 

proxy/transform.py

@@ -1,8 +1,8 @@
 import os
+import logging
 
 from bs4 import BeautifulSoup as Soup
 
-import logging
 logging.basicConfig(
     format='%(asctime)s [%(name)s] [%(levelname)s] %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
 logger = logging.getLogger(os.path.basename(__file__))
@@ -43,7 +43,7 @@ def get_static(web_root, fpath):
             ftype = "application/json"
         elif freq.lower().endswith(".xml"):
             ftype = "application/xml"
-        else:   
+        else:
             ftype = "text/plain"
 
         with open(freq, 'rb') as f:

pypowerwall/__init__.py

@@ -84,7 +84,7 @@
 from typing import Union, Optional
 import time
 
-version_tuple = (0, 10, 5)
+version_tuple = (0, 10, 6)
 version = __version__ = '%d.%d.%d' % version_tuple
 __author__ = 'jasonacox'
 
@@ -121,6 +121,7 @@ def set_debug(toggle=True, color=True):
         log.setLevel(logging.NOTSET)
 
 
+# pylint: disable=too-many-public-methods
 class Powerwall(object):
     def __init__(self, host="", password="", email="nobody@nowhere.com",
                  timezone="America/Los_Angeles", pwcacheexpire=5, timeout=5, poolmaxsize=10,
@@ -179,6 +180,8 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
             self.cloudmode = True
         if self.cloudmode and not self.fleetapi:
             self.mode = "cloud"
+        elif self.cloudmode and self.fleetapi:
+            self.mode = "fleetapi"
         elif not self.cloudmode and not self.fleetapi:
             self.mode = "local"
         else:
@@ -200,9 +203,9 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
                         auth = json.load(file)
                     self.email = list(auth.keys())[0]
                 self.cloudmode = True
-                self.fleetapi = False     
-                self.mode = "cloud"       
-                log.debug("Auto selecting Cloud Mode (email: %s)" % self.email)    
+                self.fleetapi = False
+                self.mode = "cloud"
+                log.debug("Auto selecting Cloud Mode (email: %s)" % self.email)
             else:
                 log.debug("Auto Select Failed: Unable to use local, cloud or fleetapi mode.")
 
@@ -212,8 +215,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
         # Connect to Powerwall
         if not self.connect(self.retry_modes):
             log.error("Unable to connect to Powerwall.")
-            return
-
+
     def connect(self, retry=False) -> bool:
         """
         Connect to Tesla Energy Gateway Powerwall
@@ -223,7 +225,7 @@ def connect(self, retry=False) -> bool:
         """
         if self.mode == "unknown":
             log.error("Unable to determine mode to connect.")
-            return
+            return False
         count = 0
         while count < 3:
             count += 1
@@ -237,10 +239,10 @@ def connect(self, retry=False) -> bool:
                     if not self.password and self.gw_pwd:  # Use full TEDAPI mode
                         log.debug("TEDAPI ** full **")
                         self.tedapi_mode = "full"
-                        self.client = PyPowerwallTEDAPI(self.gw_pwd, pwcacheexpire=self.pwcacheexpire, 
+                        self.client = PyPowerwallTEDAPI(self.gw_pwd, pwcacheexpire=self.pwcacheexpire,
                                                         timeout=self.timeout, host=self.host)
-                    else:   
-                        self.tedapi_mode = "hybrid"                           
+                    else:
+                        self.tedapi_mode = "hybrid"
                         self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout,
                                                self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile,
                                                self.gw_pwd)
@@ -296,6 +298,7 @@ def is_connected(self):
         except Exception:
             return False
 
+    # pylint: disable=inconsistent-return-statements
     def poll(self, api='/api/site_info/site_name', jsonformat=False, raw=False, recursive=False,
              force=False) -> Optional[Union[dict, list, str, bytes]]:
         """
@@ -339,7 +342,7 @@ def post(self, api: str, payload: Optional[dict], din: Optional[str] = None, jso
             return response
 
     def level(self, scale=False):
-        """ 
+        """
         Battery Level Percentage 
             Note: Tesla App reserves 5% of battery => ( (batterylevel / 0.95) - (5 / 0.95) )
         Args:
@@ -429,7 +432,7 @@ def strings(self, jsonformat=False, verbose=False):
                                 result[name] = {}
                             result[name][idxname] = v[device][e]
                         # if
-                    # for   
+                    # for
                     deviceidx += 1
                 # else
         # If no devices found pull from /api/solar_powerwall
@@ -485,7 +488,7 @@ def home(self, verbose=False):
         """ Home Power Usage """
         return self.load(verbose)
 
-    # Shortcut Functions 
+    # Shortcut Functions
     def site_name(self) -> Optional[str]:
         """ System Site Name """
         payload = self.poll('/api/site_info/site_name')
@@ -575,11 +578,6 @@ def alerts(self, jsonformat=False, alertsonly=True) -> Union[list, str]:
         """
         alerts = []
         devices: dict = self.vitals() or {}
-        """
-        The vitals API is not present in firmware versions > 23.44, this 
-        is a workaround to get alerts from the /api/solar_powerwall endpoint
-        for newer firmware versions
-        """
         if devices:
             for device in devices:
                 if 'alerts' in devices[device]:
@@ -590,6 +588,8 @@ def alerts(self, jsonformat=False, alertsonly=True) -> Union[list, str]:
                             item = {device: i}
                             alerts.append(item)
         elif not devices and alertsonly is True:
+            # Vitals API is not present in firmware versions > 23.44 for local mode.
+            # This is a workaround to get alerts from the /api/solar_powerwall endpoint
             data: dict = self.poll('/api/solar_powerwall') or {}
             pvac_alerts = data.get('pvac_alerts') or {}
             for alert, value in pvac_alerts.items():
@@ -680,7 +680,7 @@ def set_operation(self, level: Optional[float] = None, mode: Optional[str] = Non
             Dictionary with operation results, if jsonformat is False, else a JSON string
         """
         if level and (level < 0 or level > 100):
-            log.error(f"Level can be in range of 0 to 100 only.")
+            log.error("Level can be in range of 0 to 100 only.")
             return None
 
         if level is None:
@@ -715,7 +715,7 @@ def grid_status(self, type="string") -> Optional[Union[str, int]]:
         payload: dict = self.poll('/api/system_status/grid_status')
 
         if payload is None:
-            log.error(f"Failed to get /api/system_status/grid_status")
+            log.error("Failed to get /api/system_status/grid_status")
             return None
 
         if type == "json":
@@ -884,7 +884,7 @@ def _check_if_dir_is_writable(dirpath, name=""):
                 os.makedirs(dirpath, exist_ok=True)
             except Exception as exc:
                 raise PyPowerwallInvalidConfigurationParameter(f"Unable to create {name} directory at "
-                                                               f"'{dirpath}': {exc}")
+                                                               f"'{dirpath}': {exc}") from exc
         elif not os.path.isdir(dirpath):
             raise PyPowerwallInvalidConfigurationParameter(f"'{dirpath}' must be a directory ({name}).")
         else:

pypowerwall/__main__.py

@@ -18,8 +18,6 @@
 
 # Modules
 from pypowerwall import version, set_debug
-from pypowerwall.cloud.pypowerwall_cloud import AUTHFILE
-from pypowerwall.fleetapi.fleetapi import CONFIGFILE
 
 # Global Variables
 authpath = os.getenv("PW_AUTH_PATH", "")
@@ -97,7 +95,7 @@
         print(f"Setup Complete. Auth file {c.authfile} ready to use.")
     else:
         print("ERROR: Failed to setup Tesla Cloud Mode")
-        exit(1)
+        sys.exit(1)
 
 # FleetAPI Mode Setup
 elif command == 'fleetapi':
@@ -110,7 +108,7 @@
         print(f"Setup Complete. Config file {c.configfile} ready to use.")
     else:
         print("Setup Aborted.")
-        exit(1)
+        sys.exit(1)
 
 # TEDAPI Test
 elif command == 'tedapi':
@@ -133,19 +131,19 @@
     # If no arguments, print usage
     if not args.mode and not args.reserve and not args.current:
         print("usage: pypowerwall set [-h] [-mode MODE] [-reserve RESERVE] [-current]")
-        exit(1)
+        sys.exit(1)
     import pypowerwall
     # Determine which cloud mode to use
     pw = pypowerwall.Powerwall(auto_select=True, host="", authpath=authpath)
     print(f"pyPowerwall [{version}] - Set Powerwall Mode and Power Levels using {pw.mode} mode.\n")
     if not pw.is_connected():
         print("ERROR: FleetAPI and Cloud access are not configured. Run 'fleetapi' or 'setup'.")
-        exit(1)
+        sys.exit(1)
     if args.mode:
         mode = args.mode.lower()
         if mode not in ['self_consumption', 'backup', 'autonomous']:
             print("ERROR: Invalid Mode [%s] - must be one of self_consumption, backup, or autonomous" % mode)
-            exit(1)
+            sys.exit(1)
         print("Setting Powerwall Mode to %s" % mode)
         pw.set_mode(mode)
     if args.reserve:
@@ -167,7 +165,7 @@
         print(f"pyPowerwall [{version}] - Get Powerwall Mode and Power Levels using {pw.mode} mode.\n")
     if not pw.is_connected():
         print("ERROR: Unable to connect. Set -host and -password or configure FleetAPI or Cloud access.")
-        exit(1)
+        sys.exit(1)
     output = {
         'site': pw.site_name(),
         'site_id': pw.siteid or "N/A",