From 2aa876fad6cbe7abf07cee91cf97c943c99eec52 Mon Sep 17 00:00:00 2001 From: Jason Cox Date: Sat, 15 Jun 2024 16:22:34 -0700 Subject: [PATCH 1/3] Address pyLint cleanup and minor bug fixes --- .github/workflows/pylint.yml | 24 ++++++ .pylintrc | 5 ++ RELEASE.md | 5 ++ proxy/RELEASE.md | 4 + proxy/server.py | 27 +++---- proxy/transform.py | 4 +- pypowerwall/__init__.py | 42 +++++----- pypowerwall/__main__.py | 14 ++-- pypowerwall/cloud/pypowerwall_cloud.py | 40 ++++------ pypowerwall/fleetapi/__main__.py | 39 +++++----- pypowerwall/fleetapi/fleetapi.py | 65 ++++++++-------- pypowerwall/fleetapi/pypowerwall_fleetapi.py | 54 +++++-------- pypowerwall/pypowerwall_base.py | 1 + pypowerwall/scan.py | 19 ++++- pypowerwall/tedapi/__init__.py | 80 ++++++++++---------- pypowerwall/tedapi/__main__.py | 2 +- pypowerwall/tedapi/pypowerwall_tedapi.py | 53 ++++++++----- 17 files changed, 257 insertions(+), 221 deletions(-) create mode 100644 .github/workflows/pylint.yml create mode 100644 .pylintrc diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml new file mode 100644 index 0000000..763f22a --- /dev/null +++ b/.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 diff --git a/.pylintrc b/.pylintrc new file mode 100644 index 0000000..1707126 --- /dev/null +++ b/.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 diff --git a/RELEASE.md b/RELEASE.md index c358b1f..b76a1f7 100644 --- a/RELEASE.md +++ b/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. diff --git a/proxy/RELEASE.md b/proxy/RELEASE.md index 9142be8..f865bd3 100644 --- a/proxy/RELEASE.md +++ b/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. diff --git a/proxy/server.py b/proxy/server.py index df6bdb5..93fdf44 100755 --- a/proxy/server.py +++ b/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,6 +712,7 @@ 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, @@ -719,7 +720,7 @@ def do_GET(self): # noinspection PyBroadException try: - server.serve_forever() + server.serve_forever() except (Exception, KeyboardInterrupt, SystemExit): print(' CANCEL \n') diff --git a/proxy/transform.py b/proxy/transform.py index 1d7ace6..5fb9f85 100644 --- a/proxy/transform.py +++ b/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: diff --git a/pypowerwall/__init__.py b/pypowerwall/__init__.py index 67b71e3..92d31c9 100644 --- a/pypowerwall/__init__.py +++ b/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: diff --git a/pypowerwall/__main__.py b/pypowerwall/__main__.py index cb42ece..49fa871 100644 --- a/pypowerwall/__main__.py +++ b/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", diff --git a/pypowerwall/cloud/pypowerwall_cloud.py b/pypowerwall/cloud/pypowerwall_cloud.py index 6e93092..aef070b 100644 --- a/pypowerwall/cloud/pypowerwall_cloud.py +++ b/pypowerwall/cloud/pypowerwall_cloud.py @@ -7,10 +7,11 @@ from teslapy import Tesla, Battery, SolarPanel from pypowerwall.cloud.decorators import not_implemented_mock_data -from pypowerwall.cloud.exceptions import * -from pypowerwall.cloud.mock_data import * +from pypowerwall.cloud.exceptions import * # pylint: disable=unused-wildcard-import +from pypowerwall.cloud.mock_data import * # pylint: disable=unused-wildcard-import from pypowerwall.cloud.stubs import * from pypowerwall.pypowerwall_base import PyPowerwallBase +from pypowerwall import __version__ log = logging.getLogger(__name__) @@ -45,7 +46,7 @@ def lookup(data, keylist): return None return data - +# pylint: disable=too-many-public-methods # noinspection PyMethodMayBeStatic class PyPowerwallCloud(PyPowerwallBase): def __init__(self, email: Optional[str], pwcacheexpire: int = 5, timeout: int = 5, siteid: Optional[int] = None, @@ -292,7 +293,7 @@ def _site_api(self, name: str, ttl: int, force: bool, **kwargs): cached - True if cached data was returned """ if self.tesla is None: - log.debug(f" -- cloud: No connection to Tesla Cloud") + log.debug(" -- cloud: No connection to Tesla Cloud") return None, False # Check for lock and wait if api request already sent if name in self.apilock: @@ -322,7 +323,7 @@ def _site_api(self, name: str, ttl: int, force: bool, **kwargs): finally: # Release lock self.apilock[name] = False - return response, False + return response, False def get_battery(self, force: bool = False): """ @@ -598,6 +599,7 @@ def get_api_site_info(self, **kwargs) -> Optional[Union[dict, list, str, bytes]] } return data + # pylint: disable=unused-argument # noinspection PyUnusedLocal def get_api_devices_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: # Protobuf payload - not implemented - use /vitals instead @@ -990,7 +992,7 @@ def close_session(self): if self.tesla: self.tesla.logout() else: - log.error(f"Tesla cloud not connected") + log.error("Tesla cloud not connected") self.auth = {} def vitals(self) -> Optional[dict]: @@ -1046,8 +1048,11 @@ def post_api_operation(self, **kwargs): if __name__ == "__main__": - # Test code + import sys + # Command Line Debugging Mode + print(f"pyPowerwall - Powerwall Gateway Cloud Test [v{__version__}]") set_debug(quiet=False, debug=True, color=True) + tesla_user = None # Check for .pypowerwall.auth file AUTHPATH = os.environ.get('PW_AUTH_PATH', "") @@ -1078,7 +1083,7 @@ def post_api_operation(self, **kwargs): cloud.setup() if not cloud.connect(): log.critical("Failed to connect to Tesla Cloud") - exit(1) + sys.exit(1) log.info("Connected to Tesla Cloud") @@ -1086,25 +1091,6 @@ def post_api_operation(self, **kwargs): tsites = cloud.getsites() log.info(tsites) - # print("\Battery") - # r = cloud.get_battery() - # print(r) - - # print("\Site Power") - # r = cloud.get_site_power() - # print(r) - - # print("\Site Config") - # r = cloud.get_site_config() - # print(r) - - # Test Poll - # '/api/logout','/api/login/Basic','/vitals','/api/meters/site','/api/meters/solar', - # '/api/sitemaster','/api/powerwalls','/api/installer','/api/customer/registration', - # '/api/system/update/status','/api/site_info','/api/system_status/grid_faults', - # '/api/site_info/grid_codes','/api/solars','/api/solars/brands','/api/customer', - # '/api/meters','/api/installer','/api/networks','/api/system/networks', - # '/api/meters/readings','/api/synchrometer/ct_voltage_references'] items = ['/api/status', '/api/system_status/grid_status', '/api/site_info/site_name', '/api/devices/vitals', '/api/system_status/soe', '/api/meters/aggregates', '/api/operation', '/api/system_status', '/api/synchrometer/ct_voltage_references', diff --git a/pypowerwall/fleetapi/__main__.py b/pypowerwall/fleetapi/__main__.py index 6df5eca..9de40dd 100644 --- a/pypowerwall/fleetapi/__main__.py +++ b/pypowerwall/fleetapi/__main__.py @@ -16,8 +16,8 @@ # Import Libraries import sys import json -from .fleetapi import FleetAPI, CONFIGFILE import argparse +from .fleetapi import FleetAPI, CONFIGFILE # Display help if no arguments if len(sys.argv) == 1: @@ -37,7 +37,7 @@ print(" --config CONFIG Specify alternate config file (default: .fleetapi.config)") print(" --site SITE Specify site_id") print(" --json Output in JSON format") - exit(0) + sys.exit(0) # Parse command line arguments parser = argparse.ArgumentParser(description='Tesla FleetAPI - Command Line Interface') @@ -67,25 +67,25 @@ settings_site = args.site # Create FleetAPI object -fleet = FleetAPI(configfile=settings_file, debug=settings_debug, site_id=settings_site) +fleet = FleetAPI(configfile=settings_file, debug=settings_debug, site_id=settings_site) # Load Configuration if not fleet.load_config(): print(f" Configuration file not found: {settings_file}") if args.command != "setup": print(" Run setup to access Tesla FleetAPI.") - exit(1) + sys.exit(1) else: fleet.setup() if not fleet.load_config(): print(" Setup failed, exiting...") - exit(1) - exit(0) + sys.exit(1) + sys.exit(0) # Command: Run Setup if args.command == "setup": fleet.setup() - exit(0) + sys.exit(0) # Command: List Sites if args.command == "sites": @@ -95,7 +95,7 @@ else: for site in sites: print(f" {site['energy_site_id']} - {site['site_name']}") - exit(0) + sys.exit(0) # Command: Status if args.command == "status": @@ -105,7 +105,7 @@ else: for key in status: print(f" {key}: {status[key]}") - exit(0) + sys.exit(0) # Command: Site Info if args.command == "info": @@ -115,7 +115,7 @@ else: for key in info: print(f" {key}: {info[key]}") - exit(0) + sys.exit(0) # Command: Get Operating Mode if args.command == "getmode": @@ -124,7 +124,7 @@ print(json.dumps({"mode": mode}, indent=4)) else: print(f"{mode}") - exit(0) + sys.exit(0) # Command: Get Battery Reserve if args.command == "getreserve": @@ -133,7 +133,7 @@ print(json.dumps({"reserve": reserve}, indent=4)) else: print(f"{reserve}") - exit(0) + sys.exit(0) # Command: Set Operating Mode if args.command == "setmode": @@ -145,10 +145,10 @@ print(fleet.set_operating_mode("autonomous")) else: print("Invalid mode, must be 'self' or 'auto'") - exit(1) + sys.exit(1) else: print("No mode specified, exiting...") - exit(0) + sys.exit(0) # Command: Set Battery Reserve if args.command == "setreserve": @@ -157,19 +157,16 @@ val = int(args.argument) if val < 0 or val > 100: print(f"Invalid reserve level {val}, must be 0-100") - exit(1) + sys.exit(1) elif args.argument == "current": val = fleet.battery_level() else: print("Invalid reserve level, must be 0-100 or 'current' to set to current level.") - exit(1) + sys.exit(1) print(fleet.set_battery_reserve(int(val))) else: print("No reserve level specified, exiting...") - exit(0) + sys.exit(0) print("No command specified, exiting...") -exit(1) - - - +sys.exit(1) diff --git a/pypowerwall/fleetapi/fleetapi.py b/pypowerwall/fleetapi/fleetapi.py index ae3c5ed..0497838 100644 --- a/pypowerwall/fleetapi/fleetapi.py +++ b/pypowerwall/fleetapi/fleetapi.py @@ -51,16 +51,19 @@ # FleetAPI Class import os -import json -import requests +import json import logging import sys -import urllib.parse import time +import urllib.parse +import requests # Defaults CONFIGFILE = ".pypowerwall.fleetapi" SCOPE = "openid offline_access energy_device_data energy_cmds" +SETUP_TIMEOUT = 15 # Time in seconds to wait for setup related API response +REFRESH_TIMEOUT = 60 # Time in seconds to wait for refresh token response +API_TIMEOUT = 10 # Time in seconds to wait for FleetAPI response fleet_api_urls = { "North America, Asia-Pacific": "https://fleet-api.prd.na.vn.cloud.tesla.com", @@ -71,9 +74,10 @@ # Set up logging log = logging.getLogger(__name__) +# pylint: disable=too-many-public-methods class FleetAPI: - def __init__(self, configfile=CONFIGFILE, debug=False, site_id=None, - pwcacheexpire: int = 5, timeout: int = 5): + def __init__(self, configfile=CONFIGFILE, debug=False, site_id=None, + pwcacheexpire: int = 5, timeout: int = API_TIMEOUT): self.CLIENT_ID = "" self.CLIENT_SECRET = "" self.DOMAIN = "" @@ -84,7 +88,7 @@ def __init__(self, configfile=CONFIGFILE, debug=False, site_id=None, self.access_token = "" self.refresh_token = "" self.site_id = "" - self.debug = debug + self.debug = debug self.configfile = configfile self.pwcachetime = {} # holds the cached data timestamps for api self.pwcacheexpire = pwcacheexpire # seconds to expire cache @@ -111,7 +115,7 @@ def random_string(self, length): # Return key value from data or None def keyval(self, data, key): return data.get(key) if data and key else None - + # Load Configuration def load_config(self): if os.path.isfile(self.configfile): @@ -173,7 +177,7 @@ def new_token(self): 'Content-Type': 'application/x-www-form-urlencoded' } response = requests.post('https://auth.tesla.com/oauth2/v3/token', - data=data, headers=headers) + data=data, headers=headers, timeout=REFRESH_TIMEOUT) # Extract access_token and refresh_token from this response access = response.json().get('access_token') refresh = response.json().get('refresh_token') @@ -191,7 +195,7 @@ def new_token(self): # Update config self.save_config() self.refreshing = False - + # Poll FleetAPI def poll(self, api="api/1/products", action="GET", data=None, recursive=False, force=False): url = f"{self.AUDIENCE}/{api}" @@ -203,8 +207,8 @@ def poll(self, api="api/1/products", action="GET", data=None, recursive=False, f # Post to FleetAPI with json data payload log.debug(f"POST: {url} {json.dumps(data)}") # Check for timeout exception - try: - response = requests.post(url, headers=headers, + try: + response = requests.post(url, headers=headers, data=json.dumps(data), timeout=self.timeout) except requests.exceptions.Timeout: log.error(f"Timeout error posting to {url}") @@ -238,7 +242,7 @@ def poll(self, api="api/1/products", action="GET", data=None, recursive=False, f self.pwcachetime[api] = time.time() self.pwcache[api] = data return data - + def get_live_status(self, force=False): # Get the current power information for the site. """ @@ -261,7 +265,7 @@ def get_live_status(self, force=False): } } """ - payload = self.poll(f"api/1/energy_sites/{self.site_id}/live_status", force=force) + payload = self.poll(f"api/1/energy_sites/{self.site_id}/live_status", force=force) log.debug(f"get_live_status: {payload}") return self.keyval(payload, "response") @@ -372,7 +376,7 @@ def get_site_info(self, force=False): payload = self.poll(f"api/1/energy_sites/{self.site_id}/site_info", force=force) log.debug(f"get_site_info: {payload}") return self.keyval(payload, "response") - + def get_site_status(self, force=False): # Get site status """ @@ -407,7 +411,7 @@ def get_backup_time_remaining(self, force=False): payload = self.poll(f"api/1/energy_sites/{self.site_id}/backup_time_remaining", force=force) log.debug(f"get_backup_time_remaining: {payload}") return self.keyval(payload, "response") - + def get_products(self, force=False): # Get list of Tesla products assigned to user """ @@ -454,16 +458,16 @@ def get_products(self, force=False): "count": 2 } """ - payload = self.poll(f"api/1/products", force=force) + payload = self.poll("api/1/products", force=force) log.debug(f"get_products: {payload}") return self.keyval(payload, "response") - + def set_battery_reserve(self, reserve: int): if reserve < 0 or reserve > 100: log.debug(f"Invalid reserve level: {reserve}") return False data = {"backup_reserve_percent": reserve} - # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/backup' + # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/backup' payload = self.poll(f"api/1/energy_sites/{self.site_id}/backup", "POST", data) # Invalidate cache self.pwcachetime.pop(f"api/1/energy_sites/{self.site_id}/site_info", None) @@ -474,12 +478,12 @@ def set_operating_mode(self, mode: str): if mode not in ["self_consumption", "autonomous"]: log.debug(f"Invalid mode: {mode}") return False - # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/operation' + # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/operation' payload = self.poll(f"api/1/energy_sites/{self.site_id}/operation", "POST", data) # Invalidate cache self.pwcachetime.pop(f"api/1/energy_sites/{self.site_id}/site_info", None) return payload - + def get_operating_mode(self, force=False): return self.keyval(self.get_site_info(force=force), "default_real_mode") @@ -489,7 +493,7 @@ def get_battery_reserve(self, force=False): def getsites(self, force=False): payload = self.poll("api/1/products", force=force) return self.keyval(payload, "response") - + # Macros for common data def solar_power(self): return self.keyval(self.get_live_status(), "solar_power") @@ -533,6 +537,7 @@ def setup(self): print(" generate a user token, and get the site_id and live data for your Tesla Powerwall.") print() + current_audience = 0 # Display current configuration if we have it config = self.load_config() if config: @@ -595,10 +600,10 @@ def setup(self): # Verify that the PEM key file exists print(" Verifying PEM Key file...") verify_url = f"https://{self.DOMAIN}/.well-known/appspecific/com.tesla.3p.public-key.pem" - response = requests.get(verify_url) + response = requests.get(verify_url, timeout=SETUP_TIMEOUT) if response.status_code != 200: print(f"ERROR: Could not verify PEM key file at {verify_url}") - print(f" Make sure you have created the PEM key file and uploaded it to your website.") + print(" Make sure you have created the PEM key file and uploaded it to your website.") print() print("Run create_pem_key.py to create a PEM key file for your website.") return False @@ -621,8 +626,8 @@ def setup(self): 'Content-Type': 'application/x-www-form-urlencoded' } log.debug(f"POST: https://auth.tesla.com/oauth2/v3/token {json.dumps(data)}") - response = requests.post('https://auth.tesla.com/oauth2/v3/token', - data=data, headers=headers) + response = requests.post('https://auth.tesla.com/oauth2/v3/token', + data=data, headers=headers, timeout=SETUP_TIMEOUT) log.debug(f"Response Code: {response.status_code}") partner_token = response.json().get("access_token") self.partner_token = partner_token @@ -636,7 +641,7 @@ def setup(self): # Register Partner Account print("Step 3B - Registering your partner account...") if self.partner_account: - print(f" Already registered. Skipping...") + print(" Already registered. Skipping...") else: # If not registered, register url = f"{self.AUDIENCE}/api/1/partner_accounts" @@ -648,7 +653,7 @@ def setup(self): 'domain': self.DOMAIN, } log.debug(f"POST: {url} {json.dumps(data)}") - response = requests.post(url, headers=headers, data=json.dumps(data)) + response = requests.post(url, headers=headers, data=json.dumps(data), timeout=SETUP_TIMEOUT) log.debug(f" Response Code: {response.status_code}") self.partner_account = response.json() log.debug(f"Partner Account: {json.dumps(self.partner_account, indent=4)}\n") @@ -660,7 +665,7 @@ def setup(self): # Generate User Token print("Step 3C - Generating a one-time authentication token...") if self.access_token and self.refresh_token: - print(f" Replacing cached tokens...") + print(" Replacing cached tokens...") scope = urllib.parse.quote(SCOPE) state = self.random_string(64) url = f"https://auth.tesla.com/oauth2/v3/authorize?&client_id={self.CLIENT_ID}&locale=en-US&prompt=login&redirect_uri={self.REDIRECT_URI}&response_type=code&scope={scope}&state={state}" @@ -680,7 +685,7 @@ def setup(self): log.debug(f"Code: {code}") # Step 3D - Exchange the authorization code for a token - # The access_token will be used as the Bearer token + # The access_token will be used as the Bearer token # in the Authorization header when making API requests. print("Step 3D - Exchange the authorization code for a token") data = { @@ -697,7 +702,7 @@ def setup(self): } log.debug(f"POST: https://auth.tesla.com/oauth2/v3/token {json.dumps(data)}") response = requests.post('https://auth.tesla.com/oauth2/v3/token', - data=data, headers=headers) + data=data, headers=headers, timeout=SETUP_TIMEOUT) log.debug(f"Response Code: {response.status_code}") # Extract access_token and refresh_token from this response access_token = response.json().get('access_token') diff --git a/pypowerwall/fleetapi/pypowerwall_fleetapi.py b/pypowerwall/fleetapi/pypowerwall_fleetapi.py index d23dcfb..bef722a 100644 --- a/pypowerwall/fleetapi/pypowerwall_fleetapi.py +++ b/pypowerwall/fleetapi/pypowerwall_fleetapi.py @@ -2,14 +2,15 @@ import logging import os import time -from typing import Optional, Union, List +from typing import Optional, Union from pypowerwall.fleetapi.fleetapi import FleetAPI, CONFIGFILE from pypowerwall.fleetapi.decorators import not_implemented_mock_data -from pypowerwall.fleetapi.exceptions import * -from pypowerwall.fleetapi.mock_data import * +from pypowerwall.fleetapi.exceptions import * # pylint: disable=unused-wildcard-import +from pypowerwall.fleetapi.mock_data import * # pylint: disable=unused-wildcard-import from pypowerwall.fleetapi.stubs import * from pypowerwall.pypowerwall_base import PyPowerwallBase +from pypowerwall import __version__ log = logging.getLogger(__name__) @@ -52,6 +53,7 @@ def lookup(data, keylist): return data +# pylint: disable=too-many-public-methods # noinspection PyMethodMayBeStatic class PyPowerwallFleetAPI(PyPowerwallBase): def __init__(self, email: Optional[str], pwcacheexpire: int = 5, timeout: int = 5, siteid: Optional[int] = None, @@ -70,9 +72,9 @@ def __init__(self, email: Optional[str], pwcacheexpire: int = 5, timeout: int = self.auth = {'AuthCookie': 'local', 'UserRecord': 'local'} # Bogus local auth record # Initialize FleetAPI - self.fleet = FleetAPI(configfile=self.configfile, site_id=self.siteid, + self.fleet = FleetAPI(configfile=self.configfile, site_id=self.siteid, pwcacheexpire=pwcacheexpire, timeout=self.timeout) - + # Load Configuration if not os.path.isfile(self.fleet.configfile): log.debug(f" -- fleetapi: Configuration file not found: {self.configfile} - run setup") @@ -353,7 +355,7 @@ def get_site_info(self): } """ return self.fleet.get_site_info() - + def get_live_status(self): """ { @@ -374,12 +376,12 @@ def get_live_status(self): } """ return self.fleet.get_live_status() - + def get_time_remaining(self, force: bool = False) -> Optional[float]: """ Get backup time remaining from Tesla FleetAPI """ - response = self.fleet.get_backup_time_remaining() + response = self.fleet.get_backup_time_remaining(force=force) if response is None or not isinstance(response, dict): return None if response.get('time_remaining_hours'): @@ -422,7 +424,7 @@ def get_api_system_status_grid_status(self, **kwargs) -> Optional[Union[dict, li if power is None: data = None else: - if not not power.get("grid_status") or power.get("grid_status") in ["Active", "Unknown"]: + if not power.get("grid_status") or power.get("grid_status") in ["Active", "Unknown"]: grid_status = "SystemGridConnected" else: # off_grid or off_grid_unintentional grid_status = "SystemIslandedActive" @@ -482,6 +484,7 @@ def get_api_site_info(self, **kwargs) -> Optional[Union[dict, list, str, bytes]] } return data + # pylint: disable=unused-argument # noinspection PyUnusedLocal def get_api_devices_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: # Protobuf payload - not implemented - use /vitals instead @@ -551,7 +554,7 @@ def get_api_meters_aggregates(self, **kwargs) -> Optional[Union[dict, list, str, grid_power = power.get("grid_power") battery_count = config.get("battery_count") inverters = lookup(config, ("components", "inverters")) - + if inverters is not None: solar_inverters = len(inverters) elif lookup(config, ("components", "solar")): @@ -608,7 +611,7 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt total_pack_energy = self.fleet.total_pack_energy(force=force) energy_left = self.fleet.energy_left(force=force) nameplate_power = config.get("nameplate_power") - + if power.get("island_status") == "on_grid": grid_status = "SystemGridConnected" else: # off_grid or off_grid_unintentional @@ -632,6 +635,7 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt return data + # pylint: disable=unused-argument # noinspection PyUnusedLocal @not_implemented_mock_data def api_logout(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: @@ -757,7 +761,7 @@ def post_api_operation(self, **kwargs): if payload.get('backup_reserve_percent') is not None: backup_reserve_percent = payload['backup_reserve_percent'] - if backup_reserve_percent == False: + if backup_reserve_percent is False: backup_reserve_percent = 0 op_level = self.fleet.set_battery_reserve(backup_reserve_percent) resp['set_backup_reserve_percent'] = { @@ -775,16 +779,19 @@ def post_api_operation(self, **kwargs): if __name__ == "__main__": + import sys + # Command Line Debugging Mode + print(f"pyPowerwall - Powerwall Gateway FleetAPI Test [v{__version__}]") set_debug(quiet=False, debug=True, color=True) - fleet = PyPowerwallFleetAPI() + fleet = PyPowerwallFleetAPI(email="") if not fleet.connect(): log.info("Failed to connect to Tesla FleetAPI") fleet.setup() if not fleet.connect(): log.critical("Failed to connect to Tesla FleetAPI") - exit(1) + sys.exit(1) log.info("Connected to Tesla FleetAPI") @@ -792,25 +799,6 @@ def post_api_operation(self, **kwargs): tsites = fleet.getsites() log.info(tsites) - # print("\Battery") - # r = fleet.get_battery() - # print(r) - - # print("\Site Power") - # r = fleet.get_site_power() - # print(r) - - # print("\Site Config") - # r = fleet.get_site_config() - # print(r) - - # Test Poll - # '/api/logout','/api/login/Basic','/vitals','/api/meters/site','/api/meters/solar', - # '/api/sitemaster','/api/powerwalls','/api/installer','/api/customer/registration', - # '/api/system/update/status','/api/site_info','/api/system_status/grid_faults', - # '/api/site_info/grid_codes','/api/solars','/api/solars/brands','/api/customer', - # '/api/meters','/api/installer','/api/networks','/api/system/networks', - # '/api/meters/readings','/api/synchrometer/ct_voltage_references'] items = ['/api/status', '/api/system_status/grid_status', '/api/site_info/site_name', '/api/devices/vitals', '/api/system_status/soe', '/api/meters/aggregates', '/api/operation', '/api/system_status', '/api/synchrometer/ct_voltage_references', diff --git a/pypowerwall/pypowerwall_base.py b/pypowerwall/pypowerwall_base.py index 7e98e87..1ec4c13 100644 --- a/pypowerwall/pypowerwall_base.py +++ b/pypowerwall/pypowerwall_base.py @@ -59,6 +59,7 @@ def vitals(self) -> Optional[dict]: def get_time_remaining(self) -> Optional[float]: raise NotImplementedError + # pylint: disable=inconsistent-return-statements def fetchpower(self, sensor, verbose=False) -> Any: if verbose: payload: dict = self.poll('/api/meters/aggregates') diff --git a/pypowerwall/scan.py b/pypowerwall/scan.py index a40ca3d..7c531a4 100644 --- a/pypowerwall/scan.py +++ b/pypowerwall/scan.py @@ -18,6 +18,7 @@ import socket import threading import time +import sys import requests @@ -46,6 +47,15 @@ def getmy_ip(): def scan_ip(color, timeout, addr): + """ + Thread Worker: Scan IP Address for Powerwall Gateway + + Parameter: + color = True or False, print output in color [Default: True] + timeout = Seconds to wait per host [Default: 1.0] + addr = IP address to scan + """ + # pylint: disable=global-statement,global-variable-not-assigned global discovered, firmware global bold, subbold, normal, dim, alert, alertdim @@ -111,6 +121,7 @@ def scan(color=True, timeout=1.0, hosts=30, ip=None): and Powerwall. It tries to use your local IP address as a default. """ + # pylint: disable=global-statement,global-variable-not-assigned global discovered, firmware global bold, subbold, normal, dim, alert, alertdim @@ -151,10 +162,10 @@ def scan(color=True, timeout=1.0, hosts=30, ip=None): except Exception: # Assume user aborted print(alert + ' Cancel\n\n' + normal) - exit() + sys.exit() if response != '': - # Verify we have a valid network + # Verify we have a valid network # noinspection PyBroadException try: network = ipaddress.IPv4Network(u'' + response) @@ -190,7 +201,7 @@ def scan(color=True, timeout=1.0, hosts=30, ip=None): print('') print(normal + 'Discovered %d Powerwall Gateway' % len(discovered)) - for ip in discovered: - print(dim + ' %s [%s] Firmware %s' % (ip, discovered[ip], firmware[ip])) + for pw_ip in discovered: + print(dim + ' %s [%s] Firmware %s' % (pw_ip, discovered[pw_ip], firmware[pw_ip])) print(normal + ' ') diff --git a/pypowerwall/tedapi/__init__.py b/pypowerwall/tedapi/__init__.py index 7d468cc..105297d 100644 --- a/pypowerwall/tedapi/__init__.py +++ b/pypowerwall/tedapi/__init__.py @@ -38,17 +38,17 @@ """ # Imports -import requests + import logging -from . import tedapi_pb2 +import math +import sys +import time +import json import requests from requests.packages.urllib3.exceptions import InsecureRequestWarning -requests.packages.urllib3.disable_warnings(InsecureRequestWarning) -import json -import time from pypowerwall import __version__ -import math -import sys +from . import tedapi_pb2 +requests.packages.urllib3.disable_warnings(InsecureRequestWarning) # TEDAPI Fixed Gateway IP Address GW_IP = "192.168.91.1" @@ -77,9 +77,9 @@ def lookup(data, keylist): # TEDAPI Class class TEDAPI: - def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5, + def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5, pwconfigexpire: int = 300, host: str = GW_IP) -> None: - self.debug = debug + self.debug = debug self.pwcachetime = {} # holds the cached data timestamps for api self.pwcacheexpire = pwcacheexpire # seconds to expire status cache self.pwconfigexpire = pwconfigexpire # seconds to expire config cache @@ -98,10 +98,10 @@ def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, tim # Connect to Powerwall Gateway if not self.connect(): log.error("Failed to connect to Powerwall Gateway") - + # TEDAPI Functions - def set_debug(toggle=True, color=True): + def set_debug(self, toggle=True, color=True): """Enable verbose logging""" if toggle: if color: @@ -112,16 +112,16 @@ def set_debug(toggle=True, color=True): log.debug("%s [%s]\n" % (__name__, __version__)) else: log.setLevel(logging.NOTSET) - + def get_din(self, force=False): """ Get the DIN from the Powerwall Gateway """ # Check Cache if not force and "din" in self.pwcachetime: - if time.time() - self.pwcachetime["din"] < self.pwcacheexpire: - log.debug("Using Cached DIN") - return self.pwcache["din"] + if time.time() - self.pwcachetime["din"] < self.pwcacheexpire: + log.debug("Using Cached DIN") + return self.pwcache["din"] if not force and self.pwcooldown > time.perf_counter(): # Rate limited - return None log.debug('Rate limit cooldown period - Pausing API calls') @@ -129,7 +129,7 @@ def get_din(self, force=False): # Fetch DIN from Powerwall log.debug("Fetching DIN from Powerwall...") url = f'https://{self.gw_ip}/tedapi/din' - r = requests.get(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False) + r = requests.get(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False, timeout=self.timeout) if r.status_code in BUSY_CODES: # Rate limited - Switch to cooldown mode for 5 minutes self.pwcooldown = time.perf_counter() + 300 @@ -146,7 +146,7 @@ def get_din(self, force=False): self.pwcachetime["din"] = time.time() self.pwcache["din"] = din return din - + def get_config(self,force=False): """ Get the Powerwall Gateway Configuration @@ -184,7 +184,7 @@ def get_config(self,force=False): while self.apilock['config']: time.sleep(0.2) if time.perf_counter() >= locktime + self.timeout: - log.debug(f" -- tedapi: Timeout waiting for config (unable to acquire lock)") + log.debug(" -- tedapi: Timeout waiting for config (unable to acquire lock)") return None # Check Cache if not force and "config" in self.pwcachetime: @@ -192,9 +192,9 @@ def get_config(self,force=False): log.debug("Using Cached Payload") return self.pwcache["config"] if not force and self.pwcooldown > time.perf_counter(): - # Rate limited - return None - log.debug('Rate limit cooldown period - Pausing API calls') - return None + # Rate limited - return None + log.debug('Rate limit cooldown period - Pausing API calls') + return None # Check Connection if not self.din: if not self.connect(): @@ -216,7 +216,7 @@ def get_config(self,force=False): self.apilock['config'] = True r = requests.post(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False, headers={'Content-type': 'application/octet-string'}, - data=pb.SerializeToString()) + data=pb.SerializeToString(), timeout=self.timeout) log.debug(f"Response Code: {r.status_code}") if r.status_code in BUSY_CODES: # Rate limited - Switch to cooldown mode for 5 minutes @@ -295,7 +295,7 @@ def get_status(self, force=False): while self.apilock['status']: time.sleep(0.2) if time.perf_counter() >= locktime + self.timeout: - log.debug(f" -- tedapi: Timeout waiting for status (unable to acquire lock)") + log.debug(" -- tedapi: Timeout waiting for status (unable to acquire lock)") return None # Check Cache if not force and "status" in self.pwcachetime: @@ -303,9 +303,9 @@ def get_status(self, force=False): log.debug("Using Cached Payload") return self.pwcache["status"] if not force and self.pwcooldown > time.perf_counter(): - # Rate limited - return None - log.debug('Rate limit cooldown period - Pausing API calls') - return None + # Rate limited - return None + log.debug('Rate limit cooldown period - Pausing API calls') + return None # Check Connection if not self.din: if not self.connect(): @@ -330,7 +330,7 @@ def get_status(self, force=False): self.apilock['status'] = True r = requests.post(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False, headers={'Content-type': 'application/octet-string'}, - data=pb.SerializeToString()) + data=pb.SerializeToString(), timeout=self.timeout) log.debug(f"Response Code: {r.status_code}") if r.status_code in BUSY_CODES: # Rate limited - Switch to cooldown mode for 5 minutes @@ -376,7 +376,7 @@ def connect(self): # Connected but appears to be Powerwall 3 log.debug("Detected Powerwall 3 Gateway") self.pw3 = True - self.din = self.get_din() + self.din = self.get_din() except Exception as e: log.error(f"Unable to connect to Powerwall Gateway {self.gw_ip}") log.error("Please verify your your host has a route to the Gateway.") @@ -404,7 +404,7 @@ def current_power(self, location=None, force=False): for p in power: power[p.get('location')] = p.get('realPowerW') return power - + def backup_time_remaining(self, force=False): """ @@ -418,7 +418,7 @@ def backup_time_remaining(self, force=False): time_remaining = nominalEnergyRemainingWh / load return time_remaining - + def battery_level(self, force=False): """ Get the battery level as a percentage @@ -431,7 +431,7 @@ def battery_level(self, force=False): battery_level = nominalEnergyRemainingWh / nominalFullPackEnergyWh * 100 return battery_level - + # Vitals API Mapping Function def vitals(self, force=False): @@ -443,7 +443,7 @@ def calculate_ac_power(Vpeak, Ipeak): Irms = Ipeak / math.sqrt(2) power = Vrms * Irms return power - + def calculate_dc_power(V, I): power = V * I return power @@ -470,7 +470,7 @@ def calculate_dc_power(V, I): "connection": meter.get('connection'), "real_power_scale_factor": meter.get('real_power_scale_factor', 1) } - + # Create Header tesla = {} header = {} @@ -480,7 +480,7 @@ def calculate_dc_power(V, I): "gateway": self.gw_ip, "pyPowerwall": __version__, } - + # Create NEURIO block neurio = {} c = 1000 @@ -509,7 +509,7 @@ def calculate_dc_power(V, I): i = i + 1 meter_manufacturer = None if lookup(meter_config, [sn, 'type']) == "neurio_w2_tcp": - meter_manufacturer = "NEURIO" + meter_manufacturer = "NEURIO" rest = { "componentParentDin": lookup(config, ['vin']), "firmwareVersion": None, @@ -521,7 +521,7 @@ def calculate_dc_power(V, I): "serialNumber": sn } neurio[f"NEURIO--{sn}"] = {**cts, **rest} - + # Create PVAC, PVS, and TESLA blocks - Assume the are aligned pvac = {} pvs = {} @@ -658,7 +658,7 @@ def calculate_dc_power(V, I): "ecuType": 207 } } - + # Create TETHC, TEPINV and TEPOD blocks tethc = {} # parent tepinv = {} @@ -855,7 +855,7 @@ def calculate_dc_power(V, I): **tethc, } return vitals - + def get_blocks(self, force=False): """ @@ -866,7 +866,7 @@ def get_blocks(self, force=False): if not isinstance(status, dict) or not isinstance(config, dict): return None - block = {} + block = {} i = 0 # Loop through each THC device serial number for p in lookup(status, ['esCan', 'bus', 'THC']) or {}: @@ -919,5 +919,5 @@ def get_blocks(self, force=False): }) i = i + 1 return block - + # End of TEDAPI Class diff --git a/pypowerwall/tedapi/__main__.py b/pypowerwall/tedapi/__main__.py index ab571c8..048cd80 100644 --- a/pypowerwall/tedapi/__main__.py +++ b/pypowerwall/tedapi/__main__.py @@ -64,7 +64,7 @@ def set_debug(toggle=True, color=True): try: resp = requests.get(url, verify=False, timeout=5) log.debug(f"Connection to Powerwall Gateway successful, code {resp.status_code}.") - print(f" SUCCESS") + print(" SUCCESS") except Exception as e: print(" FAILED") print() diff --git a/pypowerwall/tedapi/pypowerwall_tedapi.py b/pypowerwall/tedapi/pypowerwall_tedapi.py index 3e6309c..9312a2b 100644 --- a/pypowerwall/tedapi/pypowerwall_tedapi.py +++ b/pypowerwall/tedapi/pypowerwall_tedapi.py @@ -1,11 +1,11 @@ import json import logging -from typing import Optional, Union, List +from typing import Optional, Union from pypowerwall.tedapi import TEDAPI, GW_IP, lookup from pypowerwall.tedapi.decorators import not_implemented_mock_data -from pypowerwall.tedapi.exceptions import * -from pypowerwall.tedapi.mock_data import * +from pypowerwall.tedapi.exceptions import * # pylint: disable=unused-wildcard-import +from pypowerwall.tedapi.mock_data import * # pylint: disable=unused-wildcard-import from pypowerwall.tedapi.stubs import * from pypowerwall.pypowerwall_base import PyPowerwallBase from pypowerwall import __version__ @@ -25,9 +25,10 @@ def set_debug(debug=False, quiet=False, color=True): log.setLevel(logging.NOTSET) +# pylint: disable=too-many-public-methods # noinspection PyMethodMayBeStatic class PyPowerwallTEDAPI(PyPowerwallBase): - def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5, + def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5, pwconfigexpire: int = 300, host: str = GW_IP) -> None: super().__init__("nobody@nowhere.com") self.tedapi = None @@ -47,12 +48,12 @@ def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, tim log.debug(f" -- tedapi: Attempting to connect to {self.host}...") if not self.tedapi.connect(): raise ConnectionError(f"Unable to connect to Tesla TEDAPI at {self.host}") - else: - log.debug(f" -- tedapi: Connected to {self.host}") + log.debug(f" -- tedapi: Connected to {self.host}") def init_post_api_map(self) -> dict: - log.debug("No support for TEDAPI POST APIs.") - return None + return { + "/api/operation": self.post_api_operation, + } def init_poll_api_map(self) -> dict: # API map for local to cloud call conversion @@ -152,6 +153,7 @@ def getsites(self): return None def change_site(self, siteid): + log.debug(f"TEDAPI does not support sites - ignoring siteid: {siteid}") return False # TEDAPI Functions @@ -160,13 +162,13 @@ def get_site_info(self): Get the site config from the TEDAPI """ return self.tedapi.get_config() - + def get_live_status(self): """ Get the live status from the TEDAPI """ return self.tedapi.get_status() - + def get_time_remaining(self, force: bool = False) -> Optional[float]: return self.tedapi.backup_time_remaining(force=force) @@ -277,14 +279,14 @@ def get_api_site_info(self, **kwargs) -> Optional[Union[dict, list, str, bytes]] return data # noinspection PyUnusedLocal - def get_api_devices_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + def get_api_devices_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: # pylint: disable=unused-argument # Protobuf payload - not implemented - use /vitals instead data = None log.warning("Protobuf payload - not implemented for /api/devices/vitals - use /vitals instead") return data - def get_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: - return self.tedapi.vitals() + def get_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: # pylint: disable=unused-argument + return self.tedapi.vitals(force=kwargs.get('force', False)) def get_api_meters_aggregates(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: force = kwargs.get('force', False) @@ -336,8 +338,7 @@ def get_api_operation(self, **kwargs) -> Optional[Union[dict, list, str, bytes]] return None else: default_real_mode = config.get("default_real_mode") - backup_reserve_percent = lookup(config, ["site_info", "backup_reserve_percent"]) or 0 - backup = (backup_reserve_percent + (5 / 0.95)) * 0.95 + backup = lookup(config, ["site_info", "backup_reserve_percent"]) or 0 data = { "real_mode": default_real_mode, "backup_reserve_percent": backup @@ -365,8 +366,8 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt data = API_SYSTEM_STATUS_STUB # TODO: see inside API_SYSTEM_STATUS_STUB definition blocks = self.tedapi.get_blocks(force=force) b = [] - for i in blocks: - b.append(blocks[i]) + for bk in blocks: + b.append(blocks[bk]) data.update({ "nominal_full_pack_energy": total_pack_energy, "nominal_energy_remaining": energy_left, @@ -382,6 +383,7 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt }) return data + # pylint: disable=unused-argument # noinspection PyUnusedLocal @not_implemented_mock_data def api_logout(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: @@ -494,18 +496,27 @@ def vitals(self) -> Optional[dict]: return self.tedapi.vitals() def post_api_operation(self, **kwargs): - log.debug("No support for TEDAPI POST APIs.") - return None + log.error("No support for TEDAPI POST APIs.") if __name__ == "__main__": + import sys + # Command Line Debugging Mode + print(f"pyPowerwall - Powerwall Gateway TEDAPI Test [v{__version__}]") set_debug(quiet=False, debug=True, color=True) - tedapi = PyPowerwallTEDAPI() + # Get the Gateway Password from the command line + if len(sys.argv) < 2: + log.error("Usage: python -m pypowerwall.tedapi.pypowerwall_tedapi ") + sys.exit(1) + password = sys.argv[1] + + # Create TEDAPI Object and get Configuration and Status + tedapi = PyPowerwallTEDAPI(password, debug=True) if not tedapi.connect(): log.info("Failed to connect to Tesla TEDAPI") - exit(1) + sys.exit(1) log.info("Connected to Tesla TEDAPI") From 03c928de3abe2340f35b8c1f214ac6b81fb35a2c Mon Sep 17 00:00:00 2001 From: Jason Cox Date: Sat, 15 Jun 2024 20:21:34 -0700 Subject: [PATCH 2/3] Add proxy to pylint test coverage --- .github/workflows/pylint.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml index 763f22a..5012332 100644 --- a/.github/workflows/pylint.yml +++ b/.github/workflows/pylint.yml @@ -22,3 +22,4 @@ jobs: - name: Analyzing the code with pylint run: | pylint -E pypowerwall/*.py + pylint -E proxy/*.py From 8b4cce2d877ba1a3d0ef19f6ee50e90b5128c404 Mon Sep 17 00:00:00 2001 From: Jason Cox Date: Sat, 15 Jun 2024 21:30:29 -0700 Subject: [PATCH 3/3] Command mode error checking --- proxy/RELEASE.md | 2 +- proxy/requirements.txt | 2 +- proxy/server.py | 10 ++++++++-- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/proxy/RELEASE.md b/proxy/RELEASE.md index f865bd3..e7d0d50 100644 --- a/proxy/RELEASE.md +++ b/proxy/RELEASE.md @@ -2,7 +2,7 @@ ### Proxy t63 (15 Jun 2024) -* pyLint Code Cleaning +* Address pyLint code cleanup and minor command mode fixes. ### Proxy t62 (13 Jun 2024) diff --git a/proxy/requirements.txt b/proxy/requirements.txt index eaab2cc..47d715a 100644 --- a/proxy/requirements.txt +++ b/proxy/requirements.txt @@ -1,2 +1,2 @@ -pypowerwall==0.10.5 +pypowerwall==0.10.6 bs4==0.0.2 diff --git a/proxy/server.py b/proxy/server.py index 93fdf44..9fefabc 100755 --- a/proxy/server.py +++ b/proxy/server.py @@ -588,10 +588,16 @@ def do_GET(self): message: str = pw.poll(self.path, jsonformat=True) elif self.path.startswith('/control/reserve'): # Current battery reserve level - message = '{"reserve": %s}' % pw_control.get_reserve() + if not pw_control: + message = '{"error": "Control Commands Disabled - Set PW_CONTROL_SECRET to enable"}' + else: + message = '{"reserve": %s}' % pw_control.get_reserve() elif self.path.startswith('/control/mode'): # Current operating mode - message = '{"mode": "%s"}' % pw_control.get_mode() + if not pw_control: + message = '{"error": "Control Commands Disabled - Set PW_CONTROL_SECRET to enable"}' + else: + message = '{"mode": "%s"}' % pw_control.get_mode() else: # Everything else - Set auth headers required for web application proxystats['gets'] = proxystats['gets'] + 1