Separate getting full, selective and location data for vehicles

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 = tim.dorssers@xs4all.nl
 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))