Separate getting full, selective and location data for vehicles (#153)

* Separate getting full, basic and location data
Addresses changes in how vehicle location data is provided after late
2023 vehicle software updates. Retains get_vehicle_data() as the method
to get all vehicle data to prevent breaking changes.

Updates version to 2.10.0 as it the initial fix for the location changes
didn't increment to 2.9.0

Adds two additional endpoints to teslapy/endpoints.json:
VEHICLE_BASIC_DATA: gets only basic vehicle data.

VEHICLE_LOCATION_DATA: basic vehicle_state and location_data.

Updates VEHICLE_DATA to get all endpoints.

Adds two methods to the Vehicle Class to use the new endpoints:
get_vehicle_basic_data(): Get only basic vehicle data.
get_vehicle_location_data(): Get basic vehicle data, and location data.

Adds two new flags to CLI:
-B --basic: Get only basic vehicle data
-G --location: Get current location (GPS) data

Updates readme.

* Fix missing version update in setup.cfg

* Revert endpoints, update __init__, set v=2.9.0
Implements feedback in original pull request:
#153 (comment)

Reverts endpoints.json to previous version for Android consistency:
remove VEHICLE_BASIC_DATA
remove VEHICLE_LOCATION_DATA

Update __init__ to include endpoints in Vehicle Class methods
-  get_vehicle_data(): all endpoints - location_data, charge_state;
    climate_state, vehicle_state, gui_settings, vehicle_config.
-  get_vehicle_basic_data(): no extra endpints.
-  get_vehicle_location_data(): only location_data and basic data.

Set version number to 2.9.0.

* Get recent location data, cleanup endpoint requests
Updates Vehicle Class:
Removes get_vehicle_basic_data()
Updates get_vehicle_data() to allow selection of individual endpoints,
    defaults to all endpoints.
Updates get_vehicle_location_data():
    - Adds optional location data max_age in seconds. Defaults to 300
      (five minutes).
    - Updates and cleans up logic to prevent issues with missing data if
    it ages out in persistent connections.
    - Fix recursion error when vehicle was online but doesn't come back
      online for subsequent requests.

Minor fixes:
    - Swaps most string formatting with fStrings, not panel ones since I
      can't test panel changes without a panel, and not the conversion
      ones since fStrings are less readable there.
    - Reflows lines over 80 chars, with same panel exception, most
      significantly in Vechicle.decode_vin()
    - Minor linting/spacing changes

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()` <sup>1</sup> | 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()` <sup>2</sup> | 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

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)

setup.cfg

@@ -3,7 +3,7 @@ universal = 1
 
 [metadata]
 name = TeslaPy
-version = 2.8.0
+version = 2.9.0
 author = Tim Dorssers
 author_email = [email protected]
 description = A Python module to use the Tesla Motors Owner API

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,16 +356,16 @@ 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()
         # Substitute path variables in URI
         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))