diff --git a/README.md b/README.md index 3e92c2d..0f20421 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,8 @@ The `Vehicle` class extends `dict` and stores vehicle data returned by the Owner | `sync_wake_up()` | No | wakes up and waits for the vehicle to come online | | `decode_option()` | No | lookup option code description (read from *option_codes.json*) | | `option_code_list()` 1 | No | lists known descriptions of the vehicle option codes | -| `get_vehicle_data()` | Yes | gets a rollup of all the data request endpoints plus vehicle config | +| `get_vehicle_data()` | Yes | get vehicle data for selected endpoints, defaults to all endpoints| +| `get_vehicle_location_data()` | Yes | gets the basic and location data for the vehicle| | `get_nearby_charging_sites()` | Yes | lists nearby Tesla-operated charging stations | | `get_service_scheduling_data()` | No | retrieves next service appointment for this vehicle | | `get_charge_history()` 2 | No | lists vehicle charging history data points | @@ -455,6 +456,8 @@ optional arguments: -r, --stream receive streaming vehicle data on-change -S, --service get service self scheduling eligibility -H, --history get charging history data + -B, --basic get basic vehicle data only + -G, --location get location (GPS) data, wake as needed -V, --verify disable verify SSL certificate -L, --logout clear token from cache and logout -u, --user get user account details diff --git a/cli.py b/cli.py index a07f097..3a70622 100755 --- a/cli.py +++ b/cli.py @@ -20,11 +20,13 @@ raw_input = vars(__builtins__).get('raw_input', input) # Py2/3 compatibility + def custom_auth(url): - # Use pywebview if no web browser specified + """ Use pywebview if no web browser specified """ if webview and not (webdriver and args.web is not None): result = [''] window = webview.create_window('Login', url) + def on_loaded(): result[0] = window.get_current_url() if 'void/callback' in result[0].split('?')[0]: @@ -46,6 +48,7 @@ def on_loaded(): WebDriverWait(browser, 300).until(EC.url_contains('void/callback')) return browser.current_url + def main(): default_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO, @@ -63,7 +66,7 @@ def main(): selected = [p for p in prod for v in p.values() if v == args.filter] logging.info('%d product(s), %d selected', len(prod), len(selected)) for i, product in enumerate(selected): - print('Product %d:' % i) + print(f'Product {i}:') # Show information or invoke API depending on arguments if args.list: print(product) @@ -74,6 +77,10 @@ def main(): product.sync_wake_up() if args.get: print(product.get_vehicle_data()) + if args.location: + print(product.get_vehicle_location_data()) + if args.basic: + print(product.get_vehicle_data(endpoints='')) if args.nearby: print(product.get_nearby_charging_sites()) if args.mobile: @@ -117,6 +124,7 @@ def main(): else: tesla.logout(not (webdriver and args.web is not None)) + if __name__ == "__main__": parser = argparse.ArgumentParser(description='Tesla Owner API CLI') parser.add_argument('-e', dest='email', help='login email', required=True) @@ -155,16 +163,20 @@ def main(): help='get service self scheduling eligibility') parser.add_argument('-H', '--history', action='store_true', help='get charging history data') + parser.add_argument('-B', '--basic', action='store_true', + help='get basic vhicle data only') + parser.add_argument('-G', '--location', action='store_true', + help='get location (GPS) data, wake as needed') parser.add_argument('-V', '--verify', action='store_false', help='disable verify SSL certificate') parser.add_argument('-L', '--logout', action='store_true', help='clear token from cache and logout') if webdriver: - h = 'use Chrome browser' if webview else 'use Chrome browser (default)' + H = 'use Chrome browser' if webview else 'use Chrome browser (default)' parser.add_argument('--chrome', action='store_const', dest='web', - help=h, const=0, default=None if webview else 0) + help=H, const=0, default=None if webview else 0) parser.add_argument('--opera', action='store_const', dest='web', - help='use Opera browser', const=1) + help='use Opera browser', const=1) if hasattr(webdriver.edge, 'options'): parser.add_argument('--edge', action='store_const', dest='web', help='use Edge browser', const=2) diff --git a/setup.cfg b/setup.cfg index 3b80f42..834451d 100644 --- a/setup.cfg +++ b/setup.cfg @@ -3,7 +3,7 @@ universal = 1 [metadata] name = TeslaPy -version = 2.8.0 +version = 2.9.0 author = Tim Dorssers author_email = tim.dorssers@xs4all.nl description = A Python module to use the Tesla Motors Owner API diff --git a/teslapy/__init__.py b/teslapy/__init__.py index 8b273ca..b3fb892 100644 --- a/teslapy/__init__.py +++ b/teslapy/__init__.py @@ -6,7 +6,7 @@ # Author: Tim Dorssers -__version__ = '2.8.0' +__version__ = '2.9.0' import os import ast @@ -192,7 +192,8 @@ def authorization_url(self, url='oauth2/v3/authorize', url = urljoin(self.sso_base_url, url) kwargs['code_challenge'] = code_challenge kwargs['code_challenge_method'] = 'S256' - without_hint, state = super(Tesla, self).authorization_url(url, **kwargs) + without_hint, state = super(Tesla, self).authorization_url(url, + **kwargs) # Detect account's registered region kwargs['login_hint'] = self.email kwargs['state'] = state @@ -291,9 +292,9 @@ def _authenticate(url): def _cache_load(self): """ Default cache loader method """ try: - with open(self.cache_file) as infile: + with open(self.cache_file, encoding='utf-8') as infile: cache = json.load(infile) - except (IOError, ValueError) as e: + except (IOError, ValueError): logger.warning('Cannot load cache: %s', self.cache_file, exc_info=True) cache = {} @@ -302,9 +303,11 @@ def _cache_load(self): def _cache_dump(self, cache): """ Default cache dumper method """ try: - with open(self.cache_file, 'w') as outfile: + with open(self.cache_file, 'w', encoding='utf-8') as outfile: json.dump(cache, outfile) - os.chmod(self.cache_file, (stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP)) + os.chmod(self.cache_file, + (stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP) + ) except IOError: logger.error('Cache not updated') else: @@ -353,8 +356,8 @@ def api(self, name, path_vars=None, **kwargs): # Lookup endpoint name try: endpoint = self.endpoints[name] - except KeyError: - raise ValueError('Unknown endpoint name ' + name) + except KeyError as e: + raise ValueError(f'Unknown endpoint name {name}') from e # Fetch token if not authorized and API requires authorization if endpoint['AUTH'] and not self.authorized: self.fetch_token() @@ -362,7 +365,7 @@ def api(self, name, path_vars=None, **kwargs): try: uri = endpoint['URI'].format(**path_vars) except KeyError as e: - raise ValueError('%s requires path variable %s' % (name, e)) + raise ValueError(f"{name} requires path variable {e}") from e # Perform request using given keyword arguments as parameters arg_name = 'params' if endpoint['TYPE'] == 'GET' else 'json' serialize = endpoint.get('CONTENT') != 'HTML' and name != 'STATUS' @@ -514,8 +517,9 @@ def sync_wake_up(self, timeout=60, interval=2, backoff=1.15): break # Raise exception when task has timed out if start_time + timeout - interval < time.time(): - raise VehicleError('%s not woken up within %s seconds' - % (self['display_name'], timeout)) + name = self['display_name'] + err = f"{name} not woken up within {timeout} seconds" + raise VehicleError(err) interval *= backoff logger.info('%s is %s', self['display_name'], self['state']) @@ -540,12 +544,31 @@ def option_code_list(self): return list(filter(None, [self.decode_option(code) for code in codes.split(',')])) - def get_vehicle_data(self): - """ A rollup of all the data request endpoints plus vehicle config. - Raises HTTPError when vehicle is not online. """ - self.update(self.api('VEHICLE_DATA', endpoints='location_data;' - 'charge_state;climate_state;vehicle_state;' - 'gui_settings;vehicle_config')['response']) + def get_vehicle_data(self, endpoints='location_data;charge_state;' + 'climate_state;vehicle_state;' + 'gui_settings;vehicle_config'): + """ Allow specifying individual endpoints to query. Defaults to all + endpoints. Raises HTTPError when vehicle is not online. + + endpoints: string containing each endpoint to query, separate with ;""" + self.update(self.api('VEHICLE_DATA', endpoints=endpoints)['response']) + self.timestamp = time.time() + return self + + def get_vehicle_location_data(self, max_age=300): + """ Get basic and location_data. Wakes vehicle if location data is not + already present, or older than max_age seconds. Raises HTTPError when + vehicle is not online. + + max_age: how long in seconds before refreshing location data. Defaults + to 300 (5 minutes). """ + last_update = self.get('drive_state', {}).get('gps_as_of') + # Check for cached data more recent than max_age + if last_update is None or last_update < (time.time() - max_age): + self.sync_wake_up() + self.update(self.api('VEHICLE_DATA', + endpoints='location_data')['response']) + self.timestamp = time.time() self.timestamp = time.time() return self @@ -568,7 +591,7 @@ def mobile_enabled(self): """ Checks if the Mobile Access setting is enabled in the car. Raises HTTPError when vehicle is in service or not online. """ # Construct URL and send request - uri = 'api/1/vehicles/%s/mobile_enabled' % self['id_s'] + uri = f"api/1/vehicles/{self['id_s']}/mobile_enabled" return self.tesla.get(uri)['response'] def compose_image(self, view='STUD_3QTR', size=640, options=None): @@ -583,7 +606,7 @@ def compose_image(self, view='STUD_3QTR', size=640, options=None): # Retrieve image from compositor url = 'https://static-assets.tesla.com/v1/compositor/' response = requests.get(url, params=params, verify=self.tesla.verify, - proxies=self.tesla.proxies) + proxies=self.tesla.proxies, timeout=30) response.raise_for_status() # Raise HTTPError, if one occurred return response.content @@ -636,37 +659,41 @@ def last_seen(self): def decode_vin(self): """ Returns decoded VIN as dict """ make = 'Tesla Model ' + self['vin'][3] - body = {'A': 'Hatch back 5 Dr / LHD', 'B': 'Hatch back 5 Dr / RHD', - 'C': 'Class E MPV / 5 Dr / LHD', 'E': 'Sedan 4 Dr / LHD', - 'D': 'Class E MPV / 5 Dr / RHD', 'F': 'Sedan 4 Dr / RHD', - 'G': 'Class D MPV / 5 Dr / LHD', 'H': 'Class D MPV / 5 Dr / RHD' - }.get(self['vin'][4], 'Unknown') - belt = {'1': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' - 'PODS, side inflatable restraints, knee airbags (FR)', - '3': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' - 'side inflatable restraints, knee airbags (FR)', - '4': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' - 'side inflatable restraints, knee airbags (FR)', - '5': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' - 'side inflatable restraints', - '6': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' - 'side inflatable restraints', - '7': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' - 'side inflatable restraints & active hood', - '8': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' - 'side inflatable restraints & active hood', - 'A': 'Type 2 manual seatbelts (FR, SR*3, TR*2) with front ' - 'airbags, PODS, side inflatable restraints, knee airbags (FR)', - 'B': 'Type 2 manual seatbelts (FR, SR*2, TR*2) with front ' - 'airbags, PODS, side inflatable restraints, knee airbags (FR)', - 'C': 'Type 2 manual seatbelts (FR, SR*2, TR*2) with front ' - 'airbags, PODS, side inflatable restraints, knee airbags (FR)', - 'D': 'Type 2 Manual Seatbelts (FR, SR*3) with front airbag, ' - 'PODS, side inflatable restraints, knee airbags (FR)' - }.get(self['vin'][5], 'Unknown') - batt = {'E': 'Electric (NMC)', 'F': 'Li-Phosphate (LFP)', - 'H': 'High Capacity (NMC)', 'S': 'Standard (NMC)', - 'V': 'Ultra Capacity (NMC)'}.get(self['vin'][6], 'Unknown') + body = { + 'A': 'Hatch back 5 Dr / LHD', 'B': 'Hatch back 5 Dr / RHD', + 'C': 'Class E MPV / 5 Dr / LHD', 'E': 'Sedan 4 Dr / LHD', + 'D': 'Class E MPV / 5 Dr / RHD', 'F': 'Sedan 4 Dr / RHD', + 'G': 'Class D MPV / 5 Dr / LHD', 'H': 'Class D MPV / 5 Dr / RHD' + }.get(self['vin'][4], 'Unknown') + belt = { + '1': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' + 'PODS, side inflatable restraints, knee airbags (FR)', + '3': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' + 'side inflatable restraints, knee airbags (FR)', + '4': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' + 'side inflatable restraints, knee airbags (FR)', + '5': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' + 'side inflatable restraints', + '6': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' + 'side inflatable restraints', + '7': 'Type 2 manual seatbelts (FR, SR*3) with front airbags, ' + 'side inflatable restraints & active hood', + '8': 'Type 2 manual seatbelts (FR, SR*2) with front airbags, ' + 'side inflatable restraints & active hood', + 'A': 'Type 2 manual seatbelts (FR, SR*3, TR*2) with front ' + 'airbags, PODS, side inflatable restraints, knee airbags (FR)', + 'B': 'Type 2 manual seatbelts (FR, SR*2, TR*2) with front ' + 'airbags, PODS, side inflatable restraints, knee airbags (FR)', + 'C': 'Type 2 manual seatbelts (FR, SR*2, TR*2) with front ' + 'airbags, PODS, side inflatable restraints, knee airbags (FR)', + 'D': 'Type 2 Manual Seatbelts (FR, SR*3) with front airbag, ' + 'PODS, side inflatable restraints, knee airbags (FR)' + }.get(self['vin'][5], 'Unknown') + batt = { + 'E': 'Electric (NMC)', 'F': 'Li-Phosphate (LFP)', + 'H': 'High Capacity (NMC)', 'S': 'Standard (NMC)', + 'V': 'Ultra Capacity (NMC)' + }.get(self['vin'][6], 'Unknown') drive = {'1': 'Single Motor - Standard', '2': 'Dual Motor - Standard', '3': 'Single Motor - Performance', '5': 'P2 Dual Motor', '4': 'Dual Motor - Performance', '6': 'P2 Tri Motor', @@ -778,8 +805,8 @@ class BatteryTariffPeriodCost( """ Represents the costs of a tariff period buy: A float containing the import price sell: A float containing the export price - name: The name for the period, must be 'ON_PEAK', 'PARTIAL_PEAK', 'OFF_PEAK', - or 'SUPER_OFF_PEAK' + name: The name for the period, must be 'ON_PEAK', 'PARTIAL_PEAK', + 'OFF_PEAK', or 'SUPER_OFF_PEAK' """ __slots__ = () @@ -860,8 +887,8 @@ def create_tariff(default_price, periods, provider, plan): if bg_period[0] <= period.start and period.end <= bg_period[1]: slot_found = True # If the period matches the start/end times, then we just - # need to adjust the existing background time slot. Otherwise - # we need to split it. + # need to adjust the existing background time slot. + # Otherwise we need to split it. if bg_period[0] == period.start: background_time[index][0] = period.end elif bg_period[1] == period.end: @@ -875,9 +902,10 @@ def create_tariff(default_price, periods, provider, plan): costs[period.cost].append(period) # The loop above can leave background time slots with zero duration. - # It's difficult to filter them out above as the list indexes can get - # out of sync as we end up modifying the array being iterated over. - # As a result it's easier to filter out invalid background slots now. + # It's difficult to filter them out above as the list indexes can + # get out of sync as we end up modifying the array being iterated + # over. As a result it's easier to filter out invalid background + # slots now. background_time = list(filter(lambda t: t[0] != t[1], background_time))