Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions pypowerwall/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,8 @@ class Powerwall(object):
def __init__(self, host="", password="", email="nobody@nowhere.com",
timezone="America/Los_Angeles", pwcacheexpire=5, timeout=5, poolmaxsize=10,
cloudmode=False, siteid=None, authpath="", authmode="cookie", cachefile=".powerwall",
fleetapi=False, auto_select=False, retry_modes=False, gw_pwd=None):
fleetapi=False, auto_select=False, retry_modes=False, gw_pwd=None,
tedapi_auth_mode="basic"):
"""
Represents a Tesla Energy Gateway Powerwall device.

Expand Down Expand Up @@ -176,6 +177,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
self.retry_modes = retry_modes
self.mode = "unknown"
self.gw_pwd = gw_pwd # TEG Gateway password for TEDAPI mode
self.tedapi_auth_mode = tedapi_auth_mode # basic or bearer for TEDAPI
self.tedapi = False
self.tedapi_mode = "off" # off, full, hybrid

Expand Down Expand Up @@ -250,7 +252,8 @@ def connect(self, retry=False) -> bool:
self.client = PyPowerwallTEDAPI(self.gw_pwd, pwcacheexpire=self.pwcacheexpire,
pwconfigexpire=self.pwcacheexpire,
timeout=self.timeout, host=self.host,
poolmaxsize=self.poolmaxsize)
poolmaxsize=self.poolmaxsize,
auth_mode=self.tedapi_auth_mode)
else:
self.tedapi_mode = "hybrid"
self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout,
Expand Down
1,008 changes: 852 additions & 156 deletions pypowerwall/tedapi/__init__.py

Large diffs are not rendered by default.

5 changes: 4 additions & 1 deletion pypowerwall/tedapi/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@ def set_debug(toggle=True, color=True):
parser = argparse.ArgumentParser(description='Tesla Powerwall Gateway TEDAPI Reader')
parser.add_argument('gw_pwd', nargs='?', help='Powerwall Gateway Password')
parser.add_argument('--gw_ip', default=GW_IP, help='Powerwall Gateway IP Address')
parser.add_argument('--auth-mode', default='basic', choices=['basic', 'bearer'],
help='Authentication mode: basic (default, requires static route to 192.168.91.1) '
'or bearer (installer login, works from home network)')
parser.add_argument('--debug', action='store_true', help='Enable Debug Output')
# Parse arguments
args = parser.parse_args()
Expand Down Expand Up @@ -90,7 +93,7 @@ def set_debug(toggle=True, color=True):
# Create TEDAPI Object and get Configuration and Status
print()
print(f"Connecting to Powerwall Gateway {GW_IP}")
ted = TEDAPI(gw_pwd, host=GW_IP)
ted = TEDAPI(gw_pwd, host=GW_IP, auth_mode=args.auth_mode)
if ted.din is None:
print("\nERROR: Unable to connect to Powerwall Gateway. Check your password and try again")
sys.exit(1)
Expand Down
33 changes: 23 additions & 10 deletions pypowerwall/tedapi/pypowerwall_tedapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,25 +80,28 @@ def compute_LL_voltage(v1n=0, v2n=0, v3n=0):
# noinspection PyMethodMayBeStatic
class PyPowerwallTEDAPI(PyPowerwallBase):
def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5,
pwconfigexpire: int = 5, host: str = GW_IP, poolmaxsize: int = 10) -> None:
pwconfigexpire: int = 5, host: str = GW_IP, poolmaxsize: int = 10,
auth_mode: str = "basic") -> None:
super().__init__("nobody@nowhere.com")
self.tedapi = None
self.tedapi: Optional[TEDAPI] = None
self.timeout = timeout
self.pwcacheexpire = pwcacheexpire
self.pwconfigexpire = pwconfigexpire
self.poolmaxsize = poolmaxsize
self.host = host
self.gw_pwd = gw_pwd
self.debug = debug
self.auth_mode = auth_mode
self.poll_api_map = self.init_poll_api_map()
self.post_api_map = self.init_post_api_map()
self.siteid = None
self.auth = {'AuthCookie': 'local', 'UserRecord': 'local'} # Bogus local auth record

# Initialize TEDAPI
self.tedapi = TEDAPI(gw_pwd=self.gw_pwd, debug=self.debug, host=self.host,
self.tedapi = TEDAPI(gw_pwd=self.gw_pwd, debug=self.debug, host=self.host,
timeout=self.timeout, pwcacheexpire=self.pwcacheexpire,
pwconfigexpire=self.pwconfigexpire, poolmaxsize=self.poolmaxsize)
pwconfigexpire=self.pwconfigexpire, poolmaxsize=self.poolmaxsize,
auth_mode=self.auth_mode)
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}")
Expand All @@ -123,7 +126,6 @@ def init_poll_api_map(self) -> dict:
"/api/system_status/grid_status": self.get_api_system_status_grid_status,
"/api/system_status/soe": self.get_api_system_status_soe,
"/vitals": self.vitals,
# Possible Actions
"/api/login/Basic": self.api_login_basic,
"/api/logout": self.api_logout,
# Mock Actions
Expand Down Expand Up @@ -642,15 +644,26 @@ 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]]:
"""Logout from the Tesla Gateway, invalidating the bearer token session."""
if self.auth_mode == "bearer":
self.tedapi._bearer_logout()
return {"status": "ok"}

# noinspection PyUnusedLocal
@not_implemented_mock_data
def api_login_basic(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]:
"""Login to the Tesla Gateway via bearer token auth.

In bearer mode, delegates to the TEDAPI bearer login which POSTs to
/api/login/Basic with stored credentials. In basic mode, auth is handled
via HTTP Basic Auth so this is a no-op.
"""
if self.auth_mode == "bearer":
try:
self.tedapi._bearer_login()
return {"status": "ok", "token": self.tedapi.token}
except Exception as e:
log.error(f"api_login_basic: {e}")
return {"status": "error", "message": str(e)}
return {"status": "ok"}

# noinspection PyUnusedLocal
Expand Down
Loading