diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json new file mode 100644 index 0000000..a6eb40d --- /dev/null +++ b/.devcontainer/devcontainer.json @@ -0,0 +1,26 @@ +// For format details, see https://aka.ms/devcontainer.json. For config options, see the +// README at: https://github.com/devcontainers/templates/tree/main/src/python +{ + "name": "Python 3", + // Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile + "image": "mcr.microsoft.com/devcontainers/python:1-3.12-bullseye", + + // Features to add to the dev container. More info: https://containers.dev/features. + // "features": {}, + + // Use 'forwardPorts' to make a list of ports inside the container available locally. + // "forwardPorts": [], + + // Use 'postCreateCommand' to run commands after the container is created. + "postCreateCommand": "pip install -e .", + + // Configure tool-specific properties. + "customizations": { + "vscode": { + "extensions": ["ms-python.python"] + } + } + + // Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root. + // "remoteUser": "root" +} diff --git a/README.md b/README.md new file mode 100644 index 0000000..4b7850d --- /dev/null +++ b/README.md @@ -0,0 +1,196 @@ +# Python API and command line tool for the Nest™ Thermostat + +**NOTE: This library support the new (post 2020) API provided by Google which replaced the original Nest Developers API.** + +For a write up on developing this library see: + +## Installation + +This library does not support Python2 + +```bash + [sudo] pip install python-google-nest +``` + +In addition to the Python library it also adds a CLI tool `nest` that is documented [below](#command-line) + +## Google Device Access Registration + +This is a fairly onerous process, so make sure to read the details before you begin. + +The biggest roadblock is that access to this API requires registering with Google for Device Access . This has a one time $5 fee. + +The documentation walks you through the rest of the process. + +At a high level it involves: + +1. Making sure your Nest devices are linked to your Google account +2. Set up GCP (Google Cloud Platform) account +3. Set up a new GCP project + 1. Create a Oauth landing page and add your email as a test user + 2. Enable the Smart device management API + 3. Create an Oauth credential with the settings called from web server and https://www.google.com as the authorized redirect URI. Note the client ID and secret from this step. +4. In https://console.nest.google.com/device-access create a new project and add oauth client ID from step 3.3 +5. Follow the series of queries in https://developers.google.com/nest/device-access/authorize to authorize devices. **Note** This step handled by this library. + +Be careful as you follow along the guide in , since you're dealing with so many similar accounts and keys it can be easy to mix something up and you won't get particularly useful errors. + +You should end up with the following pieces of information: +* project_id - ID of the project you created in https://console.nest.google.com/device-access +* client_id - value from setting up OAuth in https://console.cloud.google.com/ project +* client_secret - value from setting up OAuth in https://console.cloud.google.com/ project + +you will need those values to use this library. + +## Authentication + +This library uses those values to authenticate itself using refresh token grants. See + +![Auth flow](https://docs.wso2.com/download/attachments/60493896/OAuth%20grant%20types%20-%20refresh-token.png?version=2&modificationDate=1510629793000&api=v2) + +The first time you use this library you'll need to follow a URL the library generates to https://nestservices.google.com/partnerconnections and authenticate your devices with your Google account. When you finish this process your browser will have a URL that looks like `https://www.google.com/?state=SOME_STATE_VALUE&code=SOME_AUTHENTICATION_CODE&scope=https://www.googleapis.com/auth/sdm.service` that you need to copy and paste into the callback. + +This will be cached and for however long the token is valid the library will keep refreshing the token cache. Eventually you'll be prompted to reauthenticate. + +## Usage + +At a high level this library is used to get references to the devices included in the account. These references can be sent commands, and have a list of "traits". See for details on these traits and commands. + +See docstring comments in for details on the usage of this library. + +Example: + +```python +# reautherize_callback should be set to a function with the signature +# Callable[[str], str]] it will be called if the user needs to reautherize +# the OAuth tokens. It will be passed the URL to go to, and need to have +# the resulting URL after authentication returned. + +with nest.Nest(client_id, client_secret + ,project_id, + access_token_cache_file=access_token_cache_file, + reautherize_callback=reautherize_callback, + cache_period=cache_period) as napi: + + # Will trigger initial auth and fetch of data + devices = napi.get_devices(args.name, args.structure) + + # For a list of traits and commands see: + # https://developers.google.com/nest/device-access/traits + + if cmd == 'show_trait': + # will reuse the cached result unless cache_period has elapsed + devices = nest.Device.filter_for_trait(devices, args.trait_name) + # will reuse the cached result unless cache_period has elapsed + print(devices[args.index].traits[args.trait_name]) + elif cmd == 'cmd': + # will reuse the cached result unless cache_period has elapsed + devices = nest.Device.filter_for_cmd(devices, args.cmd_name) + # will trigger a request to POST the cmd + print(devices[args.index].send_cmd( + args.cmd_name, json.loads(args.cmd_params))) + elif cmd == 'show': + try: + while True: + for device in devices: + # will reuse the cached result and trigger a new request + # each time the cache_period elapses + print(device) + print('=========================================') + if not args.keep_alive: + break + time.sleep(2) + except KeyboardInterrupt: + return +``` + +### Command Line + +```bash +usage: nest [-h] [--conf FILE] [--token-cache TOKEN_CACHE_FILE] [-t TOKEN] [--client-id ID] [--client-secret SECRET] [--project-id PROJECT] [-k] [-n NAME] [-S STRUCTURE] [-i INDEX] [-v] {show_trait,cmd,show} ... + +Command line interface to Nest™ Thermostats + +positional arguments: + {show_trait,cmd,show} + command help + show_trait show a trait + cmd send a cmd + show show everything + +optional arguments: + -h, --help show this help message and exit + --conf FILE config file (default ~/.config/nest/config) + --token-cache TOKEN_CACHE_FILE + auth access token cache file + -t TOKEN, --token TOKEN + auth access token + --client-id ID product id on developer.nest.com + --client-secret SECRET + product secret for nest.com + --project-id PROJECT device access project id + -k, --keep-alive keep showing update received from stream API in show and camera-show commands + -n NAME, --name NAME optional, specify name of nest thermostat to talk to + -S STRUCTURE, --structure STRUCTURE + optional, specify structure name toscope device actions + -i INDEX, --index INDEX + optional, specify index number of nest to talk to + -v, --verbose showing verbose logging +``` + +examples: + +```bash +# Show all of your devices +$ nest --conf myconfig show +name: AVPHwEvCbK85AJxEDHLe91Uf73nesTCg9RyUKBq2r5G2bDnKd_6OoVek1n8JtM4WlGoqsJpCBQkl9ny4oPkTiLith-XSLQ where:Downstairs - THERMOSTAT(,,,,,,,,,) +name: AVPHwEteWa8QXa8PQ7MMzh2CtnzgDPcQCfggZquzPyF__9wUCU7gp0EhO4-_17JiB4WlNupsP3dL28TJmA9-GknM6voZPw where:Upstairs - THERMOSTAT(,,,,,,,,,) +name: AVPHwEsz8-DzdIJjNkb7iY5A5HPla6UEy7azMVyXlerdgrcuabbuLMyvlGjMLWdmqtydqtXHWfx7GHmHMaVKSDysceL4XA where:Downstairs - DOORBELL(,,,,,,) +========================================= +# add the --keep-alive to update the results every 2 seconds until killed with keyboard interrupt + +# Show all of your devices in the "Upstairs" structure +$ nest --conf myconfig -S Upstairs show +name: AVPHwEteWa8QXa8PQ7MMzh2CtnzgDPcQCfggZquzPyF__9wUCU7gp0EhO4-_17JiB4WlNupsP3dL28TJmA9-GknM6voZPw where:Upstairs - THERMOSTAT(,,,,,,,,,) +========================================= + +# Show the device with the matching name +$ nest --conf myconfig -n AVPHwEsz8-DzdIJjNkb7iY5A5HPla6UEy7azMVyXlerdgrcuabbuLMyvlGjMLWdmqtydqtXHWfx7GHmHMaVKSDysceL4XA show +name: AVPHwEsz8-DzdIJjNkb7iY5A5HPla6UEy7azMVyXlerdgrcuabbuLMyvlGjMLWdmqtydqtXHWfx7GHmHMaVKSDysceL4XA where:Downstairs - DOORBELL(,,,,,,) +========================================= + +# Show the CameraImage trait of a device +$ nest --conf myconfig show_trait CameraImage +{'maxImageResolution': {'width': 1920, 'height': 1200}} + +# Set the ThermostatMode to "HEAT" +$ nest --conf myconfig cmd ThermostatMode.SetMode '{"mode":"HEAT"}' +{} +``` + +A configuration file may be specified and used for the credentials to communicate with the NEST Thermostat. + +```ini + + [NEST] + client_id = your_client_id + client_secret = your_client_secret + project_id = your_project_id + token_cache = ~/.config/nest/token_cache +``` + +The `[NEST]` section may also be named `[nest]` for convenience. Do not use `[DEFAULT]` as it cannot be read + +## Unimplemented Features + +There are two main parts of this API that are not implemented. + +1. This library does not handle the Device Access event Pub/Sub system . Using these would avoid needing to poll the API. +2. This library does not currently handle getting video/images from the cameras. This should be possible to implement on top of this library, but would require setting up a RTSP client, or the logic to follow the links in the camera events. +3. Google provides libraries to discover the details of an API and generate code . I took a look at this process, and it didn't seem like it wouldn't make a good fit for a simple library like this. + +History +======= +This module is a fork of [python-nest](https://github.com/jkoelker/python-nest) +which was a fork of [nest_thermostat](https://github.com/FiloSottile/nest_thermostat) +which was a fork of [pynest](https://github.com/smbaker/pynest) diff --git a/README.rst b/README.rst deleted file mode 100644 index 90893da..0000000 --- a/README.rst +++ /dev/null @@ -1,308 +0,0 @@ -========================================================= -Python API and command line tool for the Nest™ Thermostat -========================================================= - -.. image:: https://travis-ci.org/jkoelker/python-nest.svg?branch=master - :target: https://travis-ci.org/jkoelker/python-nest - - -Installation -============ - -.. code-block:: bash - - [sudo] pip install python-nest - - -*NOTE* The ``4.x`` version uses the streaming endpoint. To use the older -polling/caching behavior pin your requirements to ``python-nest<4.0``. - -*NOTE* The ``3.x`` version uses the Nest official api. As such, some functionality -was removed as it is not available. To keep the old version and functionality, make sure to set -your requirements to ``python-nest<3.0``. - -Nest Developer Account -======================= - - -You will need a Nest developer account, and a Product on the Nest developer portal to use this module: - -1. Visit `Nest Developers `_, and sign in. Create an account if you don't have one already. - -2. Fill in the account details: - - - The "Company Information" can be anything. - -3. Submit changes. - -4. Click "`Products `_" at top of page. - -5. Click "`Create New Product `_" - -6. Fill in details: - - - Product name must be unique. - - - The description, users, urls can all be anything you want. - -7. For permissions, check every box and if it's an option select the read/write option. - - - The description requires a specific format to be accepted. - -8. Click "Create Product". - -9. Once the new product page opens the "Product ID" and "Product Secret" are located on the right side. These will be used as client_id and client_secret below. - - -Usage -===== - -Migrate to 4.x --------------- -The version 4.x uses `Nest Stream API `_, so that you can get nearly real time status update of your Nest devices. - -If you use python-nest as a command line tool: - You don't need to change, but there is a new command line option ``--keep-alive`` you can give a try. - -If you use python-nest in a poll loop, to query Nest device's property in certain period, there are several noticeable changes: - - The internal cache removed, the ``Structure`` and ``Device`` objects will always return their current state presented in Nest API. - - A persistence HTTP connection will keep open for each ``Nest`` object. Therefore, please avoid to create more than one Nest object in your program. - - Your poll query would not hit the API rate limit, you can increase your poll frequency. - -If you want to change to Push mode: - You need to listen ``Nest.update_event``. - Please note, any data change in all of your structures an devices will set the ``update_event``. You don't know which field got update. - -.. code-block:: python - - import nest - - napi = nest.Nest(client_id=client_id, client_secret=client_secret, access_token_cache_file=access_token_cache_file) - while napi.update_event.wait(): - napi.update_event.clear() - # assume you have one Nest Camera - print (napi.structures[0].cameras[0].motion_detected) - -If you use asyncio: - You have to wrap ``update_event.wait()`` in an ``ThreadPoolExecutor``, for example: - -.. code-block:: python - - import asyncio - import nest - - napi = nest.Nest(client_id=client_id, client_secret=client_secret, access_token_cache_file=access_token_cache_file) - event_loop = asyncio.get_event_loop() - try: - event_loop.run_until_complete(nest_update(event_loop, napi)) - finally: - event_loop.close() - - async def nest_update(loop, napi): - with ThreadPoolExecutor(max_workers=1) as executor: - while True: - await loop.run_in_executor(executor, nest.update_event.wait) - nest.update_event.clear() - # assume you have one Nest Camera - print (napi.structures[0].cameras[0].motion_detected) - - -Module ------- - -You can import the module as ``nest``. - -.. code-block:: python - - import nest - import sys - - client_id = 'XXXXXXXXXXXXXXX' - client_secret = 'XXXXXXXXXXXXXXX' - access_token_cache_file = 'nest.json' - - napi = nest.Nest(client_id=client_id, client_secret=client_secret, access_token_cache_file=access_token_cache_file) - - if napi.authorization_required: - print('Go to ' + napi.authorize_url + ' to authorize, then enter PIN below') - if sys.version_info[0] < 3: - pin = raw_input("PIN: ") - else: - pin = input("PIN: ") - napi.request_token(pin) - - for structure in napi.structures: - print ('Structure %s' % structure.name) - print (' Away: %s' % structure.away) - print (' Security State: %s' % structure.security_state) - print (' Devices:') - for device in structure.thermostats: - print (' Device: %s' % device.name) - print (' Temp: %0.1f' % device.temperature) - - # Access advanced structure properties: - for structure in napi.structures: - print ('Structure : %s' % structure.name) - print (' Postal Code : %s' % structure.postal_code) - print (' Country : %s' % structure.country_code) - print (' num_thermostats : %s' % structure.num_thermostats) - - # Access advanced device properties: - for device in structure.thermostats: - print (' Device: %s' % device.name) - print (' Where: %s' % device.where) - print (' Mode : %s' % device.mode) - print (' HVAC State : %s' % device.hvac_state) - print (' Fan : %s' % device.fan) - print (' Fan Timer : %i' % device.fan_timer) - print (' Temp : %0.1fC' % device.temperature) - print (' Humidity : %0.1f%%' % device.humidity) - print (' Target : %0.1fC' % device.target) - print (' Eco High : %0.1fC' % device.eco_temperature.high) - print (' Eco Low : %0.1fC' % device.eco_temperature.low) - print (' hvac_emer_heat_state : %s' % device.is_using_emergency_heat) - print (' online : %s' % device.online) - - # The Nest object can also be used as a context manager - # It is only for demo purpose, please do not create more than one Nest object in your program especially after 4.0 release - with nest.Nest(client_id=client_id, client_secret=client_secret, access_token_cache_file=access_token_cache_file) as napi: - for device in napi.thermostats: - device.temperature = 23 - - # Nest products can be updated to include other permissions. Before you - # can access them with the API, a user has to authorize again. To handle this - # and detect when re-authorization is required, pass in a product_version - client_id = 'XXXXXXXXXXXXXXX' - client_secret = 'XXXXXXXXXXXXXXX' - access_token_cache_file = 'nest.json' - product_version = 1337 - - # It is only for demo purpose, please do not create more than one Nest object in your program especially after 4.0 release - napi = nest.Nest(client_id=client_id, client_secret=client_secret, access_token_cache_file=access_token_cache_file, product_version=product_version) - - print("Never Authorized: %s" % napi.never_authorized) - print("Invalid Token: %s" % napi.invalid_access_token) - print("Client Version out of date: %s" % napi.client_version_out_of_date) - if napi.authorization_required is None: - print('Go to ' + napi.authorize_url + ' to authorize, then enter PIN below') - pin = input("PIN: ") - napi.request_token(pin) - - - # NOTE: By default all datetime objects are timezone unaware (UTC) - # By passing ``local_time=True`` to the ``Nest`` object datetime objects - # will be converted to the timezone reported by nest. If the ``pytz`` - # module is installed those timezone objects are used, else one is - # synthesized from the nest data - napi = nest.Nest(username, password, local_time=True) - print napi.structures[0].weather.current.datetime.tzinfo - - - - -In the API, all temperature values are reported and set in the temperature scale -the device is set to (as determined by the ``device.temperature_scale`` property). - -Helper functions for conversion are in the ``utils`` module: - -.. code-block:: python - - from nest import utils as nest_utils - temp = 23.5 - fahrenheit = nest_utils.c_to_f(temp) - temp == nest_utils.f_to_c(fahrenheit) - - -The utils function use ``decimal.Decimal`` to ensure precision. - - -Command line ------------- - -.. code-block:: bash - - usage: nest [-h] [--conf FILE] [--token-cache TOKEN_CACHE_FILE] [-t TOKEN] - [--client-id ID] [--client-secret SECRET] [-k] [-c] [-s SERIAL] - [-S STRUCTURE] [-i INDEX] [-v] - {temp,fan,mode,away,target,humid,target_hum,show,camera-show,camera-streaming,protect-show} - ... - - Command line interface to Nest™ Thermostats - - positional arguments: - {temp,fan,mode,away,target,humid,target_hum,show,camera-show,camera-streaming,protect-show} - command help - temp show/set temperature - fan set fan "on" or "auto" - mode show/set current mode - away show/set current away status - target show current temp target - humid show current humidity - target_hum show/set target humidty - show show everything - camera-show show everything (for cameras) - camera-streaming show/set camera streaming - protect-show show everything (for Nest Protect) - - optional arguments: - -h, --help show this help message and exit - --conf FILE config file (default ~/.config/nest/config) - --token-cache TOKEN_CACHE_FILE - auth access token cache file - -t TOKEN, --token TOKEN - auth access token - --client-id ID product id on developer.nest.com - --client-secret SECRET - product secret for nest.com - -k, --keep-alive keep showing update received from stream API in show - and camera-show commands - -c, --celsius use celsius instead of farenheit - -s SERIAL, --serial SERIAL - optional, specify serial number of nest thermostat to - talk to - -S STRUCTURE, --structure STRUCTURE - optional, specify structure name toscope device - actions - -i INDEX, --index INDEX - optional, specify index number of nest to talk to - -v, --verbose showing verbose logging - - examples: - # If your nest is not in range mode - nest --conf myconfig --client-id CLIENTID --client-secret SECRET temp 73 - # If your nest is in range mode - nest --conf myconfig --client-id CLIENTID --client-secret SECRET temp 66 73 - - nest --conf myconfig --client-id CLIENTID --client-secret SECRET fan --auto - nest --conf myconfig --client-id CLIENTID --client-secret SECRET target_hum 35 - - # nestcam examples - nest --conf myconfig --client-id CLIENTID --client-secret SECRET camera-show - nest --conf myconfig --client-id CLIENTID --client-secret SECRET camera-streaming --enable-camera-streaming - - # Stream API example - nest --conf myconfig --client-id CLIENTID --client-secret SECRET --keep-alive show - nest --conf myconfig --client-id CLIENTID --client-secret SECRET --keep-alive camera-show - - # Set ETA 5 minutes from now - nest --conf myconfig --client-id CLIENTID --client-secret SECRET away --away --eta 5 - -A configuration file must be specified and used for the credentials to communicate with the NEST Thermostat initially. Once completed and a token is generated, if you're using the default location for the token, the command line option will read from it automatically. - - -.. code-block:: ini - - [NEST] - client_id = your_client_id - client_secret = your_client_secret - token_cache = ~/.config/nest/token_cache - - -The ``[NEST]`` section may also be named ``[nest]`` for convenience. Do not use ``[DEFAULT]`` as it cannot be read - - -History -======= - -This module was originally a fork of `nest_thermostat `_ -which was a fork of `pynest `_ diff --git a/nest/__init__.py b/nest/__init__.py index db3bbe1..779e225 100644 --- a/nest/__init__.py +++ b/nest/__init__.py @@ -1,11 +1,8 @@ # -*- coding:utf-8 -*- import logging -from .nest import Nest - -from .utils import CELSIUS -from .utils import FAHRENHEIT +from .nest import Device, Nest, APIError, AuthorizationError logging.getLogger(__name__).addHandler(logging.NullHandler()) -__all__ = ['CELSIUS', 'FAHRENHEIT', 'Nest'] +__all__ = ['Device', 'Nest', 'APIError', 'AuthorizationError'] diff --git a/nest/command_line.py b/nest/command_line.py index cb40c80..394df17 100644 --- a/nest/command_line.py +++ b/nest/command_line.py @@ -1,4 +1,4 @@ -#! /usr/bin/python +#! /usr/bin/python3 # -*- coding:utf-8 -*- ''' @@ -11,16 +11,14 @@ import datetime import logging import os +import time import sys import errno +import json from . import nest -from . import utils from . import helpers -# use six for python2/python3 compatibility -from six.moves import input - def get_parser(): # Get Executable name @@ -57,16 +55,16 @@ def get_parser(): parser.add_argument('--client-secret', dest='client_secret', help='product secret for nest.com', metavar='SECRET') + parser.add_argument('--project-id', dest='project_id', + help='device access project id', metavar='PROJECT') + parser.add_argument('-k', '--keep-alive', dest='keep_alive', action='http://github.com/store_true', help='keep showing update received from stream API ' 'in show and camera-show commands') - parser.add_argument('-c', '--celsius', dest='celsius', action='http://github.com/store_true', - help='use celsius instead of farenheit') - - parser.add_argument('-s', '--serial', dest='serial', - help='optional, specify serial number of nest ' + parser.add_argument('-n', '--name', dest='name', + help='optional, specify name of nest ' 'thermostat to talk to') parser.add_argument('-S', '--structure', dest='structure', @@ -83,220 +81,25 @@ def get_parser(): subparsers = parser.add_subparsers(dest='command', help='command help') - temp = subparsers.add_parser('temp', help='show/set temperature') - - temp.add_argument('temperature', nargs='*', type=float, - help='target temperature to set device to') - - fan = subparsers.add_parser('fan', help='set fan "on" or "auto"') - fan_group = fan.add_mutually_exclusive_group() - fan_group.add_argument('--auto', action='http://github.com/store_true', default=False, - help='set fan to auto') - fan_group.add_argument('--on', action='http://github.com/store_true', default=False, - help='set fan to on') - - mode = subparsers.add_parser('mode', help='show/set current mode') - mode_group = mode.add_mutually_exclusive_group() - mode_group.add_argument('--cool', action='http://github.com/store_true', default=False, - help='set mode to cool') - mode_group.add_argument('--heat', action='http://github.com/store_true', default=False, - help='set mode to heat') - mode_group.add_argument('--eco', action='http://github.com/store_true', default=False, - help='set mode to eco') - mode_group.add_argument('--range', action='http://github.com/store_true', default=False, - help='set mode to range') - mode_group.add_argument('--off', action='http://github.com/store_true', default=False, - help='set mode to off') - - away = subparsers.add_parser('away', help='show/set current away status') - away_group = away.add_mutually_exclusive_group() - away_group.add_argument('--away', action='http://github.com/store_true', default=False, - help='set away status to "away"') - away_group.add_argument('--home', action='http://github.com/store_true', default=False, - help='set away status to "home"') - eta_group = away.add_argument_group() - eta_group.add_argument('--trip', dest='trip_id', help='trip information') - eta_group.add_argument('--eta', dest='eta', type=int, - help='estimated arrival time from now, in minutes') - - subparsers.add_parser('target', help='show current temp target') - subparsers.add_parser('humid', help='show current humidity') - - target_hum = subparsers.add_parser('target_hum', - help='show/set target humidty') - target_hum.add_argument('humidity', nargs='*', - help='specify target humidity value or auto ' - 'to auto-select a humidity based on outside ' - 'temp') + show_trait = subparsers.add_parser('show_trait', help='show a trait') + show_trait.add_argument('trait_name', + help='name of trait to show') - subparsers.add_parser('show', help='show everything') + cmd = subparsers.add_parser('cmd', help='send a cmd') + cmd.add_argument('cmd_name', + help='name of cmd to send') + cmd.add_argument('cmd_params', + help='json for cmd params') - # Camera parsers - subparsers.add_parser('camera-show', - help='show everything (for cameras)') - cam_streaming = subparsers.add_parser('camera-streaming', - help='show/set camera streaming') - camera_streaming_group = cam_streaming.add_mutually_exclusive_group() - camera_streaming_group.add_argument('--enable-camera-streaming', - action='http://github.com/store_true', default=False, - help='Enable camera streaming') - camera_streaming_group.add_argument('--disable-camera-streaming', - action='http://github.com/store_true', default=False, - help='Disable camera streaming') - - # Protect parsers - subparsers.add_parser('protect-show', - help='show everything (for Nest Protect)') + subparsers.add_parser('show', help='show everything') parser.set_defaults(**defaults) return parser -def get_structure(napi, args): - if args.structure: - struct = [s for s in napi.structures if s.name == args.structure] - if struct: - return struct[0] - return napi.structures[0] - - -def get_camera(napi, args, structure): - if args.serial: - return nest.Camera(args.serial, napi) - else: - return structure.cameras[args.index] - - -def get_smoke_co_alarm(napi, args, structure): - if args.serial: - return nest.SmokeCoAlarm(args.serial, napi) - else: - return structure.smoke_co_alarms[args.index] - - -def handle_camera_show(device, print_prompt, print_meta_data=True): - if print_meta_data: - print('Device : %s' % device.name) - # print('Model : %s' % device.model) # Doesn't work - print('Serial : %s' % device.serial) - print('Where : %s' % device.where) - print('Where ID : %s' % device.where_id) - print('Video History Enabled : %s' % device.is_video_history_enabled) - print('Audio Enabled : %s' % device.is_audio_enabled) - print('Public Share Enabled : %s' % device.is_public_share_enabled) - print('Snapshot URL : %s' % device.snapshot_url) - print('Nest Web App URL : %s' % device.web_url) - - print('Away : %s' % device.structure.away) - print('Sound Detected : %s' % device.sound_detected) - print('Motion Detected : %s' % device.motion_detected) - print('Person Detected : %s' % device.person_detected) - print('Streaming : %s' % device.is_streaming) - if device.structure.security_state is not None: - print('Security State : %s' % device.structure.security_state) - - if print_prompt: - print('Press Ctrl+C to EXIT') - - -def handle_camera_streaming(device, args): - if args.disable_camera_streaming: - device.is_streaming = False - elif args.enable_camera_streaming: - device.is_streaming = True - - print('Streaming : %s' % device.is_streaming) - - -def handle_camera_commands(napi, args): - structure = get_structure(napi, args) - device = get_camera(napi, args, structure) - if args.command == "camera-show": - handle_camera_show(device, args.keep_alive) - if args.keep_alive: - try: - napi.update_event.clear() - while napi.update_event.wait(): - napi.update_event.clear() - handle_camera_show(device, True, False) - except KeyboardInterrupt: - return - elif args.command == "camera-streaming": - handle_camera_streaming(device, args) - - -def handle_protect_show(device, print_prompt, print_meta_data=True): - if print_meta_data: - print('Device : %s' % device.name) - print('Serial : %s' % device.serial) - print('Where : %s' % device.where) - print('Where ID : %s' % device.where_id) - - print('CO Status : %s' % device.co_status) - print('Smoke Status : %s' % device.smoke_status) - print('Battery Health : %s' % device.battery_health) - print('Color Status : %s' % device.color_status) - - if print_prompt: - print('Press Ctrl+C to EXIT') - - -def handle_protect_show_commands(napi, args): - structure = get_structure(napi, args) - device = get_smoke_co_alarm(napi, args, structure) - handle_protect_show(device, args.keep_alive) - if args.keep_alive: - try: - napi.update_event.clear() - while napi.update_event.wait(): - napi.update_event.clear() - handle_protect_show(device, True, False) - except KeyboardInterrupt: - return - - -def handle_show_commands(napi, device, display_temp, print_prompt, - print_meta_data=True): - if print_meta_data: - # TODO should pad key? old code put : out 35 - print('Device: %s' % device.name) - print('Where: %s' % device.where) - print('Can Heat : %s' % device.can_heat) - print('Can Cool : %s' % device.can_cool) - print('Has Humidifier : %s' % device.has_humidifier) - print('Has Dehumidifier : %s' % device.has_humidifier) - print('Has Fan : %s' % device.has_fan) - print('Has Hot Water Control : %s' % device.has_hot_water_control) - - print('Away : %s' % device.structure.away) - print('Mode : %s' % device.mode) - print('State : %s' % device.hvac_state) - if device.has_fan: - print('Fan : %s' % device.fan) - print('Fan Timer : %s' % device.fan_timer) - if device.has_hot_water_control: - print('Hot Water Temp : %s' % device.fan) - print('Temp : %0.1f%s' % (device.temperature, - device.temperature_scale)) - helpers.print_if('Humidity : %0.1f%%', device.humidity) - if isinstance(device.target, tuple): - print('Target : %0.1f-%0.1f%s' % ( - display_temp(device.target[0]), - display_temp(device.target[1]), - device.temperature_scale)) - else: - print('Target : %0.1f%s' % - (display_temp(device.target), device.temperature_scale)) - - print('Away Heat : %0.1f%s' % - (display_temp(device.eco_temperature[0]), device.temperature_scale)) - print('Away Cool : %0.1f%s' % - (display_temp(device.eco_temperature[1]), device.temperature_scale)) - - print('Has Leaf : %s' % device.has_leaf) - - if print_prompt: - print('Press Ctrl+C to EXIT') +def reautherize_callback(authorization_url): + print('Please go to %s and authorize access.' % authorization_url) + return input('Enter the full callback URL: ') def main(): @@ -308,8 +111,8 @@ def main(): logger.setLevel(logging.DEBUG) console_handler = logging.StreamHandler() formatter = logging.Formatter( - "%(asctime)s %(levelname)s (%(threadName)s) " - "[%(name)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") + "%(asctime)s %(levelname)s (%(threadName)s) " + "[%(name)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") console_handler.setFormatter(formatter) console_handler.setLevel(logging.DEBUG) logger.addHandler(console_handler) @@ -340,155 +143,39 @@ def _identity(x): token_cache = os.path.expanduser(args.token_cache) - if not os.path.exists(token_cache): - if args.client_id is None or args.client_secret is None: - print("Missing client and secret. If using a configuration file," - " ensure that it is formatted properly, with a section " - "titled as per the documentation-otherwise, call with " - "--client-id and --client-secret.") - return + if args.client_id is None or args.client_secret is None or args.project_id is None: + print("Missing client_id, project_id or client_secret. If using a " + "configuration file, ensure that it is formatted properly, with " + "a section titled as per the documentation-otherwise, call with " + "--client-id and --client-secret.") + return - with nest.Nest(client_id=args.client_id, client_secret=args.client_secret, + with nest.Nest(project_id=args.project_id, client_id=args.client_id, + client_secret=args.client_secret, access_token=args.token, - access_token_cache_file=token_cache) as napi: - - if napi.authorization_required: - print('Go to ' + napi.authorize_url + - ' to authorize, then enter PIN below') - pin = input("PIN: ") - napi.request_token(pin) - - if cmd.startswith("camera"): - return handle_camera_commands(napi, args) - - elif cmd == 'protect-show': - return handle_protect_show_commands(napi, args) - - elif cmd == 'away': - structure = None - - if args.structure: - struct = [s for s in napi.structures - if s.name == args.structure] - if struct: - structure = struct[0] - - else: - if args.serial: - serial = args.serial - else: - serial = napi.thermostats[args.index]._serial - - struct = [s for s in napi.structures for d in s.thermostats - if d._serial == serial] - if struct: - structure = struct[0] - - if not structure: - structure = napi.structures[0] - - if args.away: - structure.away = True - if args.eta: - eta = datetime.datetime.utcnow() \ - + datetime.timedelta(minutes=args.eta) - if args.trip_id is None: - dt = datetime.datetime.utcnow() - ts = (dt - datetime.datetime(1970, 1, 1)) \ - / datetime.timedelta(seconds=1) - args.trip_id = "trip_{}".format(round(ts)) - print("Set ETA %s for trip %s" % (eta, args.trip_id)) - structure.set_eta(args.trip_id, eta) - - elif args.home: - structure.away = False - - print(structure.away) - return - - if args.serial: - device = nest.Thermostat(args.serial, napi) - - elif args.structure: - struct = [s for s in napi.structures if s.name == args.structure] - if struct: - device = struct[0].thermostats[args.index] - - else: - device = napi.structures[0].thermostats[args.index] - - else: - device = napi.thermostats[args.index] - - if args.celsius and device.temperature_scale is 'F': - display_temp = utils.f_to_c - elif not args.celsius and device.temperature_scale is 'C': - display_temp = utils.c_to_f - - if cmd == 'temp': - if args.temperature: - if len(args.temperature) > 1: - if device.mode != 'range': - device.mode = 'range' - - device.temperature = args.temperature - - else: - device.temperature = args.temperature[0] - - print('%0.1f' % display_temp(device.temperature)) - - elif cmd == 'fan': - if args.auto: - device.fan = False - - elif args.on: - device.fan = True - - print(device.fan) - - elif cmd == 'mode': - if args.cool: - device.mode = 'cool' - - elif args.heat: - device.mode = 'heat' - - elif args.eco: - device.mode = 'eco' - - elif args.range: - device.mode = 'range' - - elif args.off: - device.mode = 'off' - - print(device.mode) - - elif cmd == 'humid': - print(device.humidity) - - elif cmd == 'target': - target = device.target - - if isinstance(target, tuple): - print('Lower: %0.1f' % display_temp(target[0])) - print('Upper: %0.1f' % display_temp(target[1])) - - else: - print('%0.1f' % display_temp(target)) - + access_token_cache_file=token_cache, + reautherize_callback=reautherize_callback) as napi: + + devices = napi.get_devices(args.name, args.structure) + + if cmd == 'show_trait': + devices = nest.Device.filter_for_trait(devices, args.trait_name) + print(devices[args.index].traits[args.trait_name]) + elif cmd == 'cmd': + devices = nest.Device.filter_for_cmd(devices, args.cmd_name) + print(devices[args.index].send_cmd( + args.cmd_name, json.loads(args.cmd_params))) elif cmd == 'show': - handle_show_commands(napi, device, display_temp, args.keep_alive) - if args.keep_alive: - try: - napi.update_event.clear() - while napi.update_event.wait(): - napi.update_event.clear() - handle_show_commands(napi, device, display_temp, - True, False) - except KeyboardInterrupt: - return + try: + while True: + for device in devices: + print(device) + print('=========================================') + if not args.keep_alive: + break + time.sleep(2) + except KeyboardInterrupt: + return if __name__ == '__main__': diff --git a/nest/helpers.py b/nest/helpers.py index 36eabfc..acd89bd 100644 --- a/nest/helpers.py +++ b/nest/helpers.py @@ -2,15 +2,11 @@ # a module of helper functions # mostly for the configuration -from __future__ import print_function -import contextlib import os +import configparser from . import nest -# use six for python2/python3 compatibility -from six.moves import configparser - class MissingCredentialsError(ValueError): pass @@ -20,12 +16,13 @@ def get_config(config_path=None, prog='nest'): if not config_path: config_path = os.path.sep.join(('~', '.config', prog, 'config')) - defaults = {'celsius': False} config_file = os.path.expanduser(config_path) + defaults = {} + # Note, this cannot accept sections titled 'DEFAULT' if os.path.exists(config_file): - config = configparser.SafeConfigParser() + config = configparser.ConfigParser() config.read([config_file]) if config.has_section('nest'): defaults.update(dict(config.items('nest'))) @@ -36,44 +33,3 @@ def get_config(config_path=None, prog='nest'): exit() return defaults - - -def get_auth_credentials(config_path=None): - config = get_config(config_path) - username = config.get('user') - password = config.get('password') - return username, password - - -@contextlib.contextmanager -def nest_login(config_path=None, username=None, password=None, **kwargs): - """ - This a context manager for creating a Nest object using - authentication credentials either provided as keyword arguments - or read from the configuration file. - - :param config_path: Path to the config file. - The default is used if none is provided. - Optional if the the credentials are provided as arguments. - :param username: Optional if the config file contains the username. - :param password: Optional if the config file contains the password. - :param kwargs: Keyword arguments to pass onto the Nest initializer. - :return: Nest object - """ - - credentials_config = get_auth_credentials(config_path) - if not username: - username = credentials_config[0] - if not password: - password = credentials_config[1] - - if username and password: - yield nest.Nest(username, password, **kwargs) - else: - raise MissingCredentialsError( - 'The login credentials have not been provided.') - - -def print_if(str, *fmt_args): - if all(fmt_args): - print(str % fmt_args) diff --git a/nest/nest.py b/nest/nest.py index da80500..6160958 100644 --- a/nest/nest.py +++ b/nest/nest.py @@ -1,1877 +1,317 @@ # -*- coding:utf-8 -*- -import collections -import copy -import datetime -import hashlib import logging -import threading -import time -import os -import uuid -import weakref - -from dateutil.parser import parse as parse_time - -import requests -from requests import auth -from requests import adapters -from requests.compat import json - -import sseclient - -ACCESS_TOKEN_URL = 'https://api.home.nest.com/oauth2/access_token' -AUTHORIZE_URL = 'https://home.nest.com/login/oauth2?client_id={0}&state={1}' -API_URL = 'https://developer-api.nest.com/' -LOGIN_URL = 'https://home.nest.com/user/login' -SIMULATOR_SNAPSHOT_URL = \ - 'https://developer.nest.com/' \ - '/simulator/api/v1/nest/devices/camera/snapshot' -SIMULATOR_SNAPSHOT_PLACEHOLDER_URL = \ - 'https://media.giphy.com/media/WCwFvyeb6WJna/giphy.gif' - -AWAY_MAP = {'on': 'away', - 'away': 'away', - 'off': 'home', - 'home': 'home', - True: 'away', - False: 'home'} - -FAN_MAP = {'auto on': False, - 'on': True, - 'auto': False, - '1': True, - '0': False, - 1: True, - 0: False, - True: True, - False: False} - -LowHighTuple = collections.namedtuple('LowHighTuple', ('low', 'high')) - -DEVICES = 'devices' -METADATA = 'metadata' -STRUCTURES = 'structures' -THERMOSTATS = 'thermostats' -SMOKE_CO_ALARMS = 'smoke_co_alarms' -CAMERAS = 'cameras' - -# https://developers.nest.com/documentation/api-reference/overview#targettemperaturef -MINIMUM_TEMPERATURE_F = 50 -MAXIMUM_TEMPERATURE_F = 90 -# https://developers.nest.com/documentation/api-reference/overview#targettemperaturec -MINIMUM_TEMPERATURE_C = 9 -MAXIMUM_TEMPERATURE_C = 32 - -_LOGGER = logging.getLogger(__name__) - - -class APIError(Exception): - def __init__(self, response, msg=None): - if response is None: - response_content = b'' - else: - try: - response_content = response.content - except AttributeError: - response_content = response.data - - if response_content != b'': - if isinstance(response, requests.Response): - message = response.json()['error'] - else: - message = "API Error Occured" - - if msg is not None: - message = "API Error Occured: " + msg - - # Call the base class constructor with the parameters it needs - super(APIError, self).__init__(message) - - self.response = response - - -class AuthorizationError(Exception): - def __init__(self, response, msg=None): - if response is None: - response_content = b'' - else: - try: - response_content = response.content - except AttributeError: - response_content = response.data - - if response_content != b'': - if isinstance(response, requests.Response): - message = response.json().get( - 'error_description', - "Authorization Failed") - else: - message = "Authorization failed" - - if msg is not None: - message = "Authorization Failed: " + msg - - # Call the base class constructor with the parameters it needs - super(AuthorizationError, self).__init__(message) - - self.response = response - - -class NestAuth(auth.AuthBase): - def __init__(self, auth_callback=None, session=None, - client_id=None, client_secret=None, - access_token=None, access_token_cache_file=None): - self._res = {} - self.auth_callback = auth_callback - self.pin = None - self._access_token_cache_file = access_token_cache_file - self._client_id = client_id - self._client_secret = client_secret - self._access_token = access_token - - if (access_token_cache_file is not None and - access_token is None and - os.path.exists(access_token_cache_file)): - with open(access_token_cache_file, 'r') as f: - _LOGGER.debug("Load access token from %s", - access_token_cache_file) - self._res = json.load(f) - self._callback(self._res) - - if session is not None: - session = weakref.ref(session) - - self._session = session - self._adapter = adapters.HTTPAdapter() - - def _cache(self): - if self._access_token_cache_file is not None: - with os.fdopen(os.open(self._access_token_cache_file, - os.O_WRONLY | os.O_CREAT, 0o600), - 'w') as f: - _LOGGER.debug("Save access token to %s", - self._access_token_cache_file) - json.dump(self._res, f) - - def _callback(self, res): - if self.auth_callback is not None and isinstance(self.auth_callback, - collections.Callable): - self.auth_callback(res) - - def login(self, headers=None): - data = {'client_id': self._client_id, - 'client_secret': self._client_secret, - 'code': self.pin, - 'grant_type': 'authorization_code'} - - post = requests.post - - if self._session: - session = self._session() - post = session.post - - _LOGGER.debug(">> POST %s", ACCESS_TOKEN_URL) - response = post(ACCESS_TOKEN_URL, data=data, headers=headers) - _LOGGER.debug("<< %s", response.status_code) - if response.status_code != 200: - raise AuthorizationError(response) - self._res = response.json() - - self._cache() - self._callback(self._res) - - @property - def access_token(self): - return self._res.get('access_token', self._access_token) - - def __call__(self, r): - if self.access_token: - r.headers['Authorization'] = 'Bearer ' + self.access_token - - return r - - -class NestBase(object): - def __init__(self, serial, nest_api): - self._serial = serial - self._nest_api = nest_api - - def __str__(self): - return '<%s: %s>' % (self.__class__.__name__, self._repr_name) - - def _set(self, what, data): - path = '/%s/%s' % (what, self._serial) - - response = self._nest_api._put(path=path, data=data) - - return response - - @property - def _weather(self): - raise NotImplementedError("Deprecated Nest API") - # merge_code = self.postal_code + ',' + self.country_code - # return self._nest_api._weather[merge_code] - - @property - def weather(self): - raise NotImplementedError("Deprecated Nest API") - # return Weather(self._weather, self._local_time) - - @property - def serial(self): - return self._serial - - @property - def _repr_name(self): - return self.serial - - -class Device(NestBase): - @property - def _device(self): - raise NotImplementedError("Implemented by subclass") - - @property - def _devices(self): - return self._nest_api._devices - - @property - def _repr_name(self): - if self.name: - return self.name - - return self.where - - def __repr__(self): - return str(self._device) - - @property - def name(self): - return self._device.get('name') - - @name.setter - def name(self, value): - raise NotImplementedError("Needs updating with new API") - # self._set('shared', {'name': value}) - - @property - def name_long(self): - return self._device.get('name_long') - - @property - def device_id(self): - return self._device.get('device_id') - - @property - def online(self): - return self._device.get('is_online') - - @property - def software_version(self): - return self._device.get('software_version') - - @property - def structure(self): - if 'structure_id' in self._device: - return Structure(self._device['structure_id'], - self._nest_api) - else: - return None - - @property - def where(self): - if self.where_id is not None: - # This name isn't always present due to upstream bug in the API - # https://nestdevelopers.io/t/missing-where-name-from-some-devices/1202 - if self.where_id in self.structure.wheres: - return self.structure.wheres[self.where_id]['name'] - else: - return self.where_id - - @property - def where_id(self): - return self._device.get('where_id') - - @where.setter - def where(self, value): - value = value.lower() - ident = self.structure.wheres.get(value) - - if ident is None: - self.structure.add_where(value) - ident = self.structure.wheres[value] - - self._set('device', {'where_id': ident}) - - @property - def description(self): - return self._device.get('name_long') - - @property - def is_thermostat(self): - return False - - @property - def is_camera(self): - return False - - @property - def is_smoke_co_alarm(self): - return False - - -class Thermostat(Device): - @property - def is_thermostat(self): - return True - - @property - def _device(self): - return self._devices.get(THERMOSTATS, {}).get(self._serial, {}) - - @property - def _shared(self): - raise NotImplementedError("Deprecated Nest API") - # return self._nest_api._status['shared'][self._serial] - - @property - def _track(self): - raise NotImplementedError("Deprecated Nest API") - # return self._nest_api._status['track'][self._serial] - - @property - def fan(self): - # FIXME confirm this is the same as old havac_fan_state - return self._device.get('fan_timer_active') - - @fan.setter - def fan(self, value): - mapped_value = FAN_MAP.get(value, False) - if mapped_value is None: - raise ValueError("Only True and False supported") - - self._set('devices/thermostats', {'fan_timer_active': mapped_value}) - - @property - def fan_timer(self): - return self._device.get('fan_timer_duration') - - @fan_timer.setter - def fan_timer(self, value): - self._set('devices/thermostats', {'fan_timer_duration': value}) - - @property - def humidity(self): - return self._device.get('humidity') - - @property - def target_humidity(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['target_humidity'] - - @target_humidity.setter - def target_humidity(self, value): - raise NotImplementedError("No longer available in Nest API") - # if value == 'auto': - - # if self._weather['current']['temp_c'] >= 4.44: - # hum_value = 45 - # elif self._weather['current']['temp_c'] >= -1.11: - # hum_value = 40 - # elif self._weather['current']['temp_c'] >= -6.67: - # hum_value = 35 - # elif self._weather['current']['temp_c'] >= -12.22: - # hum_value = 30 - # elif self._weather['current']['temp_c'] >= -17.78: - # hum_value = 25 - # elif self._weather['current']['temp_c'] >= -23.33: - # hum_value = 20 - # elif self._weather['current']['temp_c'] >= -28.89: - # hum_value = 15 - # elif self._weather['current']['temp_c'] >= -34.44: - # hum_value = 10 - # else: - # hum_value = value - - # if float(hum_value) != self._device['target_humidity']: - # self._set('device', {'target_humidity': float(hum_value)}) - - @property - def mode(self): - # FIXME confirm same as target_temperature_type - return self._device.get('hvac_mode') - - @mode.setter - def mode(self, value): - self._set('devices/thermostats', {'hvac_mode': value.lower()}) - - @property - def has_leaf(self): - return self._device.get('has_leaf') - - @property - def hvac_ac_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_ac_state'] - - @property - def hvac_cool_x2_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_cool_x2_state'] - - @property - def hvac_heater_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_heater_state'] - - @property - def hvac_aux_heater_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_aux_heater_state'] - - @property - def hvac_heat_x2_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_heat_x2_state'] - - @property - def hvac_heat_x3_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_heat_x3_state'] - - @property - def hvac_alt_heat_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_alt_heat_state'] - - @property - def hvac_alt_heat_x2_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._shared['hvac_alt_heat_x2_state'] - - @property - def hvac_emer_heat_state(self): - raise NotImplementedError( - "No longer available in Nest API. See " - "is_using_emergency_heat instead") - # return self._shared['hvac_emer_heat_state'] - - @property - def is_using_emergency_heat(self): - return self._device.get('is_using_emergency_heat') - - @property - def label(self): - return self._device.get('label') - - @property - def local_ip(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['local_ip'] - - @property - def last_ip(self): - raise NotImplementedError("No longer available in Nest API") - # return self._track['last_ip'] - - @property - def last_connection(self): - # TODO confirm this does get set, or if the API documentation is wrong - return self._device.get('last_connection') - - @property - def error_code(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['error_code'] - - @property - def battery_level(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['battery_level'] - - @property - def battery_health(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['battery_health'] - - @property - def postal_code(self): - return self.structure.postal_code - # return self._device['postal_code'] - - def _temp_key(self, key): - return "%s_%s" % (key, self.temperature_scale.lower()) - - def _round_temp(self, temp): - if self.temperature_scale == 'C': - return round(temp * 2) / 2 - else: - # F goes to nearest degree - return int(round(temp)) - - @property - def temperature_scale(self): - return self._device.get('temperature_scale') - - @temperature_scale.setter - def temperature_scale(self, value): - self._set('devices/thermostats', {'temperature_scale': value.upper()}) - - @property - def is_locked(self): - return self._device.get('is_locked') - - @property - def locked_temperature(self): - low = self._device.get(self._temp_key('locked_temp_min')) - high = self._device.get(self._temp_key('locked_temp_max')) - return LowHighTuple(low, high) - - @property - def temperature(self): - return self._device.get(self._temp_key('ambient_temperature')) - - @property - def min_temperature(self): - if self.is_locked: - return self.locked_temperature[0] - else: - if self.temperature_scale == 'C': - return MINIMUM_TEMPERATURE_C - else: - return MINIMUM_TEMPERATURE_F - - @property - def max_temperature(self): - if self.is_locked: - return self.locked_temperature[1] - else: - if self.temperature_scale == 'C': - return MAXIMUM_TEMPERATURE_C - else: - return MAXIMUM_TEMPERATURE_F - - @temperature.setter - def temperature(self, value): - self.target = value - - @property - def target(self): - if self.mode == 'heat-cool': - low = self._device.get(self._temp_key('target_temperature_low')) - high = self._device.get(self._temp_key('target_temperature_high')) - return LowHighTuple(low, high) - - return self._device.get(self._temp_key('target_temperature')) - - @target.setter - def target(self, value): - data = {} - - if self.mode == 'heat-cool': - rounded_low = self._round_temp(value[0]) - rounded_high = self._round_temp(value[1]) - - data[self._temp_key('target_temperature_low')] = rounded_low - data[self._temp_key('target_temperature_high')] = rounded_high - else: - rounded_temp = self._round_temp(value) - data[self._temp_key('target_temperature')] = rounded_temp - - self._set('devices/thermostats', data) - - @property - def away_temperature(self): - # see https://nestdevelopers.io/t/new-things-for-fall/226 - raise NotImplementedError( - "Deprecated Nest API, use eco_temperature instead") - - @away_temperature.setter - def away_temperature(self, value): - # see https://nestdevelopers.io/t/new-things-for-fall/226 - raise NotImplementedError( - "Deprecated Nest API, use eco_temperature instead") - - @property - def eco_temperature(self): - # use get, since eco_temperature isn't always filled out - low = self._device.get(self._temp_key('eco_temperature_low')) - high = self._device.get(self._temp_key('eco_temperature_high')) - - return LowHighTuple(low, high) - - @eco_temperature.setter - def eco_temperature(self, value): - low, high = value - data = {} - - if low is not None: - data[self._temp_key('eco_temperature_low')] = low - - if high is not None: - data[self._temp_key('eco_temperature_high')] = high - - self._set('devices/thermostats', data) - - @property - def can_heat(self): - return self._device.get('can_heat') - - @property - def can_cool(self): - return self._device.get('can_cool') - - @property - def has_humidifier(self): - return self._device.get('has_humidifier') - - @property - def has_dehumidifier(self): - return self._device.get('has_dehumidifier') - - @property - def has_fan(self): - return self._device.get('has_fan') - - @property - def has_hot_water_control(self): - return self._device.get('has_hot_water_control') - - @property - def hot_water_temperature(self): - return self._device.get('hot_water_temperature') - - @property - def hvac_state(self): - return self._device.get('hvac_state') - - @property - def eco(self): - raise NotImplementedError("Deprecated Nest API") - # eco_mode = self._device['eco']['mode'] - # # eco modes can be auto-eco or manual-eco - # return eco_mode.endswith('eco') - - @eco.setter - def eco(self, value): - raise NotImplementedError("Deprecated Nest API") - # data = {'eco': self._device['eco']} - # if value: - # data['eco']['mode'] = 'manual-eco' - # else: - # data['eco']['mode'] = 'schedule' - # data['eco']['mode_update_timestamp'] = time.time() - # self._set('device', data) - - @property - def previous_mode(self): - return self._device.get('previous_hvac_mode') - - @property - def time_to_target(self): - return self._device.get('time_to_target') - - @property - def time_to_target_training(self): - return self._device.get('time_to_target_training') - - -class SmokeCoAlarm(Device): - @property - def is_smoke_co_alarm(self): - return True - - @property - def _device(self): - return self._devices.get(SMOKE_CO_ALARMS, {}).get(self._serial, {}) - - @property - def auto_away(self): - raise NotImplementedError("No longer available in Nest API.") - # return self._device['auto_away'] - - @property - def battery_health(self): - return self._device.get('battery_health') - - @property - def battery_health_state(self): - raise NotImplementedError("use battery_health instead") - - @property - def battery_level(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['battery_level'] - - @property - def capability_level(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['capability_level'] - - @property - def certification_body(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['certification_body'] - - @property - def co_blame_duration(self): - raise NotImplementedError("No longer available in Nest API") - # if 'co_blame_duration' in self._device: - # return self._device['co_blame_duration'] - - @property - def co_blame_threshold(self): - raise NotImplementedError("No longer available in Nest API") - # if 'co_blame_threshold' in self._device: - # return self._device['co_blame_threshold'] - - @property - def co_previous_peak(self): - raise NotImplementedError("No longer available in Nest API") - # if 'co_previous_peak' in self._device: - # return self._device['co_previous_peak'] - - @property - def co_sequence_number(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['co_sequence_number'] - - @property - def co_status(self): - # TODO deprecate for new name - return self._device.get('co_alarm_state') - - @property - def color_status(self): - return self._device.get('ui_color_state') - - @property - def component_als_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_als_test_passed'] - - @property - def component_co_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_co_test_passed'] - - @property - def component_heat_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_heat_test_passed'] - - @property - def component_hum_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_hum_test_passed'] - - @property - def component_led_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_led_test_passed'] - - @property - def component_pir_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_pir_test_passed'] - - @property - def component_smoke_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_smoke_test_passed'] - - @property - def component_temp_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_temp_test_passed'] - - @property - def component_us_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_us_test_passed'] - - @property - def component_wifi_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_wifi_test_passed'] - - @property - def creation_time(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['creation_time'] - - @property - def device_external_color(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['device_external_color'] - - @property - def device_locale(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['device_locale'] - - @property - def fabric_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['fabric_id'] - - @property - def factory_loaded_languages(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['factory_loaded_languages'] - - @property - def gesture_hush_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['gesture_hush_enable'] - - @property - def heads_up_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['heads_up_enable'] - - @property - def home_alarm_link_capable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['home_alarm_link_capable'] - - @property - def home_alarm_link_connected(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['home_alarm_link_connected'] - - @property - def home_alarm_link_type(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['home_alarm_link_type'] - - @property - def hushed_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['hushed_state'] - - @property - def installed_locale(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['installed_locale'] - - @property - def kl_software_version(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['kl_software_version'] - - @property - def latest_manual_test_cancelled(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['latest_manual_test_cancelled'] - - @property - def latest_manual_test_end_utc_secs(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['latest_manual_test_end_utc_secs'] - - @property - def latest_manual_test_start_utc_secs(self): - # TODO confirm units, deprecate for new method name - return self._device.get('last_manual_test_time') - - @property - def last_manual_test_time(self): - # TODO parse time, check that it's in the dict - return self._device.get('last_manual_test_time') - - @property - def line_power_present(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['line_power_present'] - - @property - def night_light_continuous(self): - raise NotImplementedError("No longer available in Nest API") - # if 'night_light_continuous' in self._device: - # return self._device['night_light_continuous'] - - @property - def night_light_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['night_light_enable'] - - @property - def ntp_green_led_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['ntp_green_led_enable'] - - @property - def product_id(self): - return self._device.get('product_id') - - @property - def replace_by_date_utc_secs(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['replace_by_date_utc_secs'] - - @property - def resource_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['resource_id'] - - @property - def smoke_sequence_number(self): - return self._device.get('smoke_sequence_number') - - @property - def smoke_status(self): - return self._device.get('smoke_alarm_state') - - @property - def spoken_where_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['spoken_where_id'] - - @property - def steam_detection_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['steam_detection_enable'] - - @property - def thread_mac_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['thread_mac_address'] - - @property - def wifi_ip_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_ip_address'] - - @property - def wifi_mac_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_mac_address'] - - @property - def wifi_regulatory_domain(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_regulatory_domain'] - - @property - def wired_led_enable(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wired_led_enable'] - - @property - def wired_or_battery(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wired_or_battery'] - - -class ActivityZone(NestBase): - def __init__(self, camera, zone_id): - self.camera = camera - NestBase.__init__(self, camera.serial, camera._nest_api) - # camera's activity_zone dict has int, but an event's list of - # activity_zone ids is strings `\/0_0\/` - self._zone_id = int(zone_id) - - @property - def _camera(self): - return self.camera._device - - @property - def _repr_name(self): - return self.name - - @property - def _activity_zone(self): - return next( - z for z in self._camera.get('activity_zones') - if z['id'] == self.zone_id) - - @property - def zone_id(self): - return self._zone_id - - @property - def name(self): - return self._activity_zone.get('name') - - -class CameraEvent(NestBase): - def __init__(self, camera): - NestBase.__init__(self, camera.serial, camera._nest_api) - self.camera = camera - - @property - def _camera(self): - return self.camera._device - - @property - def _event(self): - return self._camera.get('last_event') - - def __str__(self): - return '<%s>' % (self.__class__.__name__) - - def __repr__(self): - return str(self._event) - - def activity_in_zone(self, zone_id): - if 'activity_zone_ids' in self._event: - return str(zone_id) in self._event['activity_zone_ids'] - return False - - @property - def activity_zones(self): - if 'activity_zone_ids' in self._event: - return [ActivityZone(self, z) - for z in self._event['activity_zone_ids']] - - @property - def animated_image_url(http://github.com/self): - return self._event.get('animated_image_url') - - @property - def app_url(http://github.com/self): - return self._event.get('app_url') - - @property - def has_motion(self): - return self._event.get('has_motion') - - @property - def has_person(self): - return self._event.get('has_person') - - @property - def has_sound(self): - return self._event.get('has_sound') - - @property - def image_url(http://github.com/self): - return self._event.get('image_url') - - @property - def start_time(self): - if 'start_time' in self._event: - return parse_time(self._event['start_time']) - - @property - def end_time(self): - if 'end_time' in self._event: - end_time = parse_time(self._event['end_time']) - if end_time: - return end_time + datetime.timedelta(seconds=30) - - @property - def urls_expire_time(self): - if 'urls_expire_time' in self._event: - return parse_time(self._event['urls_expire_time']) - - @property - def web_url(http://github.com/self): - return self._event.get('web_url') - - @property - def is_ongoing(self): - if self.end_time is not None: - # sometimes, existing event is updated with a new start time - # that's before the end_time which implies something new - if self.start_time > self.end_time: - return True - - now = datetime.datetime.now(self.end_time.tzinfo) - # end time should be in the past - return self.end_time > now - # no end_time implies it's ongoing - return True - - def has_ongoing_motion_in_zone(self, zone_id): - if self.is_ongoing and self.has_motion: - return self.activity_in_zone(zone_id) - - def has_ongoing_sound(self): - if self.is_ongoing: - return self.has_sound - - def has_ongoing_motion(self): - if self.is_ongoing: - return self.has_motion - - def has_ongoing_person(self): - if self.is_ongoing: - return self.has_person - - -class Camera(Device): - @property - def is_camera(self): - return True - - @property - def _device(self): - return self._devices.get(CAMERAS, {}).get(self._serial, {}) - - @property - def ongoing_event(self): - if self.last_event is not None and self.last_event.is_ongoing: - return self.last_event - - def has_ongoing_motion_in_zone(self, zone_id): - if self.ongoing_event is not None: - return self.last_event.has_ongoing_motion_in_zone(zone_id) - return False - - @property - def sound_detected(self): - if self.ongoing_event is not None: - return self.last_event.has_ongoing_sound() - return False - - @property - def motion_detected(self): - if self.ongoing_event is not None: - return self.last_event.has_ongoing_motion() - return False - - @property - def person_detected(self): - if self.ongoing_event is not None: - return self.last_event.has_ongoing_person() - return False - - @property - def activity_zones(self): - return [ActivityZone(self, z['id']) - for z in self._device.get('activity_zones', [])] - - @property - def last_event(self): - if 'last_event' in self._device: - return CameraEvent(self) - - @property - def is_streaming(self): - return self._device.get('is_streaming') - - @is_streaming.setter - def is_streaming(self, value): - self._set('devices/cameras', {'is_streaming': value}) - - @property - def is_video_history_enabled(self): - return self._device.get('is_video_history_enabled') - - @property - def is_audio_enabled(self): - return self._device.get('is_audio_input_enabled') - - @property - def is_public_share_enabled(self): - return self._device.get('is_public_share_enabled') - - @property - def capabilities(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['capabilities'] - - @property - def cvr(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['cvr_enrolled'] - - @property - def nexustalk_host(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['direct_nexustalk_host'] - - @property - def download_host(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['download_host'] - - @property - def last_connected(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['last_connected_time'] - - @property - def last_cuepoint(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['last_cuepoint'] - - @property - def live_stream(self): - # return self._device['live_stream_host'] - raise NotImplementedError("No longer available in Nest API") - - @property - def mac_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['mac_address'] - - @property - def model(self): - return self._device.get('model') - - @property - def nexus_api_http_server_url(http://github.com/self): - # return self._device['nexus_api_http_server_url'] - raise NotImplementedError("No longer available in Nest API") - - @property - def streaming_state(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['streaming_state'] - - @property - def component_hum_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_hum_test_passed'] - - @property - def component_led_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_led_test_passed'] - - @property - def component_pir_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_pir_test_passed'] - - @property - def component_smoke_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_smoke_test_passed'] - - @property - def component_temp_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_temp_test_passed'] - - @property - def component_us_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_us_test_passed'] - - @property - def component_wifi_test_passed(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['component_wifi_test_passed'] - - @property - def creation_time(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['creation_time'] - - @property - def device_external_color(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['device_external_color'] - - @property - def device_locale(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['device_locale'] - - @property - def fabric_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['fabric_id'] - - @property - def factory_loaded_languages(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['factory_loaded_languages'] - - @property - def installed_locale(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['installed_locale'] - - @property - def kl_software_version(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['kl_software_version'] - - @property - def product_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['product_id'] - - @property - def resource_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['resource_id'] - - @property - def spoken_where_id(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['spoken_where_id'] - - @property - def thread_mac_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['thread_mac_address'] - - @property - def where_id(self): - return self._device.get('where_id') - - @property - def wifi_ip_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_ip_address'] - - @property - def wifi_mac_address(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_mac_address'] - - @property - def wifi_regulatory_domain(self): - raise NotImplementedError("No longer available in Nest API") - # return self._device['wifi_regulatory_domain'] - - @property - def snapshot_url(http://github.com/self): - if ('snapshot_url' in self._device and - self._device['snapshot_url'] != SIMULATOR_SNAPSHOT_URL): - return self._device['snapshot_url'] - else: - return SIMULATOR_SNAPSHOT_PLACEHOLDER_URL - - @property - def web_url(http://github.com/self): - return self._device.get('web_url') - - -class Structure(NestBase): - @property - def _structure(self): - return self._nest_api._status.get( - STRUCTURES, {}).get(self._serial, {}) - - def __repr__(self): - return str(self._structure) - - def _set_away(self, value, auto_away=False): - self._set('structures', {'away': AWAY_MAP[value]}) - - @property - def away(self): - return self._structure.get('away') - - @away.setter - def away(self, value): - self._set_away(value) - - @property - def country_code(self): - return self._structure.get('country_code') - - @property - def devices(self): - raise NotImplementedError("Use thermostats instead") - - @property - def thermostats(self): - if THERMOSTATS in self._structure: - return [Thermostat(devid, self._nest_api) - for devid in self._structure[THERMOSTATS]] - else: - return [] - - @property - def protectdevices(self): - raise NotImplementedError("Use smoke_co_alarms instead") - - @property - def smoke_co_alarms(self): - if SMOKE_CO_ALARMS in self._structure: - return [SmokeCoAlarm(devid, self._nest_api) - for devid in self._structure[SMOKE_CO_ALARMS]] - else: - return [] - - @property - def cameradevices(self): - raise NotImplementedError("Use cameras instead") - - @property - def cameras(self): - if CAMERAS in self._structure: - return [Camera(devid, self._nest_api) - for devid in self._structure[CAMERAS]] - else: - return [] +import time +import os +import threading +from typing import Dict, Any, List, Callable, Optional - @property - def dr_reminder_enabled(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['dr_reminder_enabled'] +import requests +from requests.compat import json - @property - def emergency_contact_description(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['emergency_contact_description'] +from requests_oauthlib import OAuth2Session +from oauthlib.oauth2 import TokenExpiredError - @property - def emergency_contact_type(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['emergency_contact_type'] +# Interface URLs +ACCESS_TOKEN_URL = 'https://www.googleapis.com/oauth2/v4/token' +AUTHORIZE_URL = 'https://nestservices.google.com/partnerconnections/%7Bproject_id%7D/auth' +API_URL = 'https://smartdevicemanagement.googleapis.com/v1/enterprises/%7Bproject_id%7D/devices' +REDIRECT_URI = 'https://www.google.com/' +SCOPE = ['https://www.googleapis.com/auth/sdm.service'] - @property - def emergency_contact_phone(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['emergency_contact_phone'] +_LOGGER = logging.getLogger(__name__) - @property - def enhanced_auto_away_enabled(self): - # FIXME there is probably an equivilant thing for this - raise NotImplementedError("Deprecated Nest API") - # return self._structure['topaz_enhanced_auto_away_enabled'] - @property - def eta_preconditioning_active(self): - # FIXME there is probably an equivilant thing for this - # or something that can be recommended - raise NotImplementedError("Deprecated Nest API") - # return self._structure['eta_preconditioning_active'] +class Device(): + """This is the class used to access the traits of a Nest device - @property - def house_type(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['house_type'] + You can access a list of traits and send commands. + The class is linked back to a Nest instance which will keep it updated + based on the Nest objects cache_period. - @property - def hvac_safety_shutoff_enabled(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['hvac_safety_shutoff_enabled'] + Since any access can trigger a network request, they can result in exceptions + """ - @property - def name(self): - return self._structure['name'] + def __init__(self, nest_api: Optional['Nest'] = None, + name: Optional[str] = None, + device_data: Optional[Dict[str, Any]] = None): + """Meant for internal use, get instances of Device from the Nest api - @name.setter - def name(self, value): - self._set('structures', {'name': value}) + Devices returned have the nest_api and name set. - @property - def location(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure.get('location') + Parameters + ---------- + nest_api : Nest + The Nest instance providing updates + name : str + The unique name of this device + device_data : Dict + Instead of specifying the previous two, intialize directly from the + dict for the device returned from the API call. Used internally. + """ - @property - def address(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure.get('street_address') + self._name = name + self._nest_api = nest_api + self._device_data = device_data - @property - def num_thermostats(self): - if THERMOSTATS in self._structure: - return len(self._structure[THERMOSTATS]) - else: - return 0 + def __str__(self): + trait_str = ','.join([f'<{k}: {v}>' for k, v in self.traits.items()]) + return f'name: {self.name} where:{self.where} - {self.type}({trait_str})' @property - def num_cameras(self): - if CAMERAS in self._structure: - return len(self._structure[CAMERAS]) + def name(self) -> str: + """str representing the unique name of the device""" + if self._device_data is not None: + full_name = self._device_data['name'] else: - return 0 + full_name = self._name + return full_name.split('/')[-1] @property - def num_smokecoalarms(self): - if SMOKE_CO_ALARMS in self._structure: - return len(self._structure[SMOKE_CO_ALARMS]) + def _device(self): + if self._device_data is not None: + return self._device_data else: - return 0 - - @property - def measurement_scale(self): - raise NotImplementedError( - "Deprecated Nest API, see temperature_scale on " - "thermostats instead") - # return self._structure['measurement_scale'] - - @property - def postal_code(self): - # TODO check permissions if this is empty? - return self._structure.get('postal_code') - - @property - def renovation_date(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['renovation_date'] + return next(device for device in self._devices if self.name in device['name']) @property - def structure_area(self): - raise NotImplementedError("Deprecated Nest API") - # return self._structure['structure_area'] - - @property - def time_zone(self): - return self._structure.get('time_zone') + def _devices(self): + if self._device_data is not None: + raise RuntimeError("Invalid use of singular device") + return self._nest_api._devices @property - def peak_period_start_time(self): - if 'peak_period_start_time' in self._structure: - return parse_time(self._structure['peak_period_start_time']) + def where(self) -> str: + """str representing the parent structure of the device""" + return self._device['parentRelations'][0]['displayName'] @property - def peak_period_end_time(self): - if 'peak_period_end_time' in self._structure: - return parse_time(self._structure['peak_period_end_time']) + def type(self) -> str: + """str representing the type of device""" + return self._device['type'].split('.')[-1] @property - def eta_begin(self): - if 'eta_begin' in self._structure: - return parse_time(self._structure['eta_begin']) + def traits(self) -> Dict[str, Any]: + """list of traits see https://developers.google.com/nest/device-access/traits""" + return {k.split('.')[-1]: v for k, v in self._device['traits'].items()} - def _set_eta(self, trip_id, eta_begin, eta_end): - if self.num_thermostats == 0: - raise ValueError("ETA can only be set or cancelled when a" - " thermostat is in the structure.") - if trip_id is None: - raise ValueError("trip_id must be not None") + def send_cmd(self, cmd: str, params: Dict[str, Any]) -> Dict[str, Any]: + """Send a command to this device - data = {'trip_id': trip_id, - 'estimated_arrival_window_begin': eta_begin, - 'estimated_arrival_window_end': eta_end} + commands are listed in https://developers.google.com/nest/device-access/traits - self._set('structures', {'eta': data}) + Parameters + ---------- + cmd : str + The string for the command can include the full command ie: + "sdm.devices.commands.ThermostatTemperatureSetpoint.SetCool" + or just the last two parts ie: + "ThermostatTemperatureSetpoint.SetCool" + params : dict + The content of the params to send with the command - def set_eta(self, trip_id, eta_begin, eta_end=None): - """ - Set estimated arrival winow, use same trip_id to update estimation. - Nest may choose to ignore inaccurate estimation. - See: https://developers.nest.com/documentation/cloud/away-guide - #make_an_eta_write_call - """ - if eta_begin is None: - raise ValueError("eta_begin must be not None") - if eta_end is None: - eta_end = eta_begin + datetime.timedelta(minutes=1) + Exceptions + ---------- + Will return APIError if the command or params is invalid - self._set_eta(trip_id, eta_begin.isoformat(), eta_end.isoformat()) - - def cancel_eta(self, trip_id): - """ - Cancel estimated arrival winow. + Returns + ------- + Dict + The body of the response """ - eta_end = datetime.datetime.utcnow() - self._set_eta(trip_id, int(0), eta_end.isoformat()) - - @property - def wheres(self): - return self._structure.get('wheres') - - @wheres.setter - def wheres(self, value): - self._set('where', {'wheres': value}) - - def add_where(self, name, ident=None): - name = name.lower() - - if name in self.wheres: - return self.wheres[name] - - name = ' '.join([n.capitalize() for n in name.split()]) - wheres = copy.copy(self.wheres) - - if ident is None: - ident = str(uuid.uuid4()) - - wheres.append({'name': name, 'where_id': ident}) - self.wheres = wheres - - return self.add_where(name) - - def remove_where(self, name): - name = name.lower() - - if name not in self.wheres: - return None - - ident = self.wheres[name] + cmd = '.'.join(cmd.split('.')[-2:]) + path = f'/{self.name}:executeCommand' + data = { + "command": "sdm.devices.commands." + cmd, + 'params': params + } + response = self._nest_api._put(path=path, data=data) + return response - wheres = [w for w in copy.copy(self.wheres) - if w['name'] != name and w['where_id'] != ident] + @staticmethod + def filter_for_trait(devices: List['Device'], trait: str) -> List['Device']: + """Filter a list of Devices for ones with a the specified trait""" + trait = trait.split('.')[-1] + return [device for device in devices if trait in device.traits] - self.wheres = wheres - return ident + @staticmethod + def filter_for_cmd(devices: List['Device'], cmd: str) -> List['Device']: + """Filter a list of Devices for ones with a trait associated with a cmd - @property - def security_state(self): - """ - Return 'ok' or 'deter'. Need sercurity state ready permission. - Note: this is NOT for Net Secruity alarm system. - See https://developers.nest.com/documentation/cloud/security-guide + ie. "ThermostatTemperatureSetpoint.SetCool" will filter for devices + with the "ThermostatTemperatureSetpoint" trait """ - return self._structure.get('wwn_security_state') + trait = cmd.split('.')[-2] + return Device.filter_for_trait(devices, trait) class Nest(object): - def __init__(self, username=None, password=None, - user_agent=None, - access_token=None, access_token_cache_file=None, - local_time=False, - client_id=None, client_secret=None, - product_version=None): - self._urls = {} - self._limits = {} - self._user = None - self._userid = None - self._weave = None - self._staff = False - self._superuser = False - self._email = None - self._queue = collections.deque(maxlen=2) - self._event_thread = None - self._update_event = threading.Event() - self._queue_lock = threading.Lock() - - if local_time: - raise ValueError("local_time no longer supported") - - if user_agent: - raise ValueError("user_agent no longer supported") - - self._access_token = access_token + """This is the class used to manage the connection to Google Smart Devices + + It handles the authentication flow and returns a list of the devices + associated with the account. These devices will call back to this class + to keep their values up to date. + """ + + def __init__(self, + client_id: str, client_secret: str, + project_id: str, + access_token: Optional[Dict[str, Any]] = None, + access_token_cache_file: Optional[str] = None, + reautherize_callback: Optional[Callable[[str], str]] = None, + cache_period: float = 10): + """ + Parameters + ---------- + client_id : str + OAuth client_id + client_secret : str + OAuth secret + project_id : str + The project_id from https://console.nest.google.com/device-access/project-list + access_token : Optional[Dict[str, Any]] + Directly specify the OAuth access token ie.: + {"access_token": "", "expires_in": 3599, + "scope": ["https://www.googleapis.com/auth/sdm.service"], + "token_type": "Bearer", "expires_at": 1617334543.9341743, + "refresh_token": ""} + access_token_cache_file : Optional[str] + A path to store and load tokens to avoid needing to reauthentic + every time. + reautherize_callback : Optional[Callable[[str], str]] + If the token is expired or invalid, this callback will be called + with the URL the user needs to go to to revalidate. If not set + an AuthorizationError exception will trigger. + cache_period : float + When requesting the device set, how long should the previous + results be reused before making a new request. + """ self._client_id = client_id self._client_secret = client_secret - self._product_version = product_version - - self._session = requests.Session() - auth = NestAuth(client_id=self._client_id, - client_secret=self._client_secret, - session=self._session, access_token=access_token, - access_token_cache_file=access_token_cache_file) - self._session.auth = auth - - @property - def update_event(self): - return self._update_event - - @property - def authorization_required(self): - return self.never_authorized or \ - self.invalid_access_token or \ - self.client_version_out_of_date - - @property - def never_authorized(self): - return self.access_token is None - - @property - def invalid_access_token(self): - try: - self._get("/") - return False - except AuthorizationError: - return True + self._project_id = project_id + self._cache_period = cache_period + self._access_token_cache_file = access_token_cache_file + self._reautherize_callback = reautherize_callback + self._lock = threading.Lock() + self._last_update = 0 + self._client = None + self._devices_value = {} - @property - def client_version_out_of_date(self): - if self._product_version is not None: + if not access_token and self._access_token_cache_file: try: - return self.client_version < self._product_version - # an error means they need to authorize anyways - except AuthorizationError: - return True - return False - - @property - def authorize_url(http://github.com/self): - state = hashlib.md5(os.urandom(32)).hexdigest() - return AUTHORIZE_URL.format(self._client_id, state) - - def request_token(self, pin): - self._session.auth.pin = pin - self._session.auth.login() - - @property - def access_token(self): - return self._access_token or self._session.auth.access_token - - def _handle_ratelimit(self, res, verb, url, data, - max_retries=10, default_wait=5, - stream=False, headers=None): - response = res - retries = 0 - while response.status_code == 429 and retries <= max_retries: - retries += 1 - retry_after = response.headers.get('Retry-After') - _LOGGER.info("Reach rate limit, retry (%d), after %s", - retries, retry_after) - # Default Retry Time - wait = default_wait + with open(self._access_token_cache_file, 'r') as fd: + access_token = json.load(fd) + _LOGGER.debug("Loaded access token from %s", + self._access_token_cache_file) + except: + _LOGGER.warn("Token load failed from %s", + self._access_token_cache_file) + if access_token: + self._client = OAuth2Session(self._client_id, token=access_token) + + def __save_token(self, token): + if self._access_token_cache_file: + with open(self._access_token_cache_file, 'w') as fd: + json.dump(token, fd) + _LOGGER.debug("Save access token to %s", + self._access_token_cache_file) - if retry_after is not None: + def __reauthorize(self): + if self._reautherize_callback is None: + raise AuthorizationError(None, 'No callback to handle OAuth URL') + self._client = OAuth2Session( + self._client_id, redirect_uri=REDIRECT_URI, scope=SCOPE) + + authorization_url, state = self._client.authorization_url( + AUTHORIZE_URL.format(project_id=self._project_id), + # access_type and prompt are Google specific extra + # parameters. + access_type="offline", prompt="consent") + authorization_response = self._reautherize_callback(authorization_url) + _LOGGER.debug(">> fetch_token") + token = self._client.fetch_token( + ACCESS_TOKEN_URL, + authorization_response=authorization_response, + # Google specific extra parameter used for client + # authentication + client_secret=self._client_secret) + self.__save_token(token) + + def _request(self, verb, path, data=None): + url = self._api_url + path + if data is not None: + data = json.dumps(data) + attempt = 0 + while True: + attempt += 1 + if self._client: try: - # Checks if retry_after is a number - wait = float(retry_after) - except ValueError: - # If not: - try: - # Checks if retry_after is a HTTP date - now = datetime.datetime.now() - wait = (now - parse_time(retry_after)).total_seconds() - except ValueError: - # Use default - pass - - _LOGGER.debug("Wait %d seconds.", wait) - time.sleep(wait) - _LOGGER.debug(">> %s %s", 'STREAM' if stream else verb, url) - response = self._session.request(verb, url, + _LOGGER.debug(">> %s %s", verb, url) + r = self._client.request(verb, url, allow_redirects=False, - stream=stream, - headers=headers, data=data) - _LOGGER.debug("<< %s", response.status_code) - return response - - def _open_data_stream(self, path="/"): - url = "%s%s" % (API_URL, path) - _LOGGER.debug(">> STREAM %s", url) - - # Opens the data stream - headers = {'Accept': 'text/event-stream'} - # Set Connection Timeout to 30 seconds - # Set Read Timeout to 5 mintues, Nest Stream API will send - # keep alive event every 30 seconds, 5 mintues is long enough - # for us to belive network issue occurred - response = self._session.get(url, stream=True, headers=headers, - allow_redirects=False, - timeout=(30, 300)) - - _LOGGER.debug("<< %s", response.status_code) - if response.status_code == 401: - raise AuthorizationError(response) - - if response.status_code == 429: - response = self._handle_ratelimit(response, 'GET', url, None, - max_retries=10, - default_wait=5, - stream=True, - headers=headers) - - if response.status_code == 307: - redirect_url = response.headers['Location'] - _LOGGER.debug(">> STREAM %s", redirect_url) - # For stream API, we have to deal redirect manually - response = self._session.get(redirect_url, - allow_redirects=False, - headers=headers, - stream=True, - timeout=(30, 300)) - _LOGGER.debug("<< %s", response.status_code) - if response.status_code == 429: - response = self._handle_ratelimit(response, 'GET', url, None, - max_retries=10, - default_wait=5, - stream=True, - headers=headers) - - ready_event = threading.Event() - self._event_thread = threading.Thread(target=self._start_event_loop, - args=(response, - self._queue, - ready_event, - self._update_event)) - self._event_thread.setDaemon(True) - self._event_thread.start() - ready_event.wait(timeout=10) - _LOGGER.info("Event loop started.") - - def _start_event_loop(self, response, queue, ready_event, update_event): - _LOGGER.debug("Starting event loop.") - try: - client = sseclient.SSEClient(response.iter_content()) - for event in client.events(): - event_type = event.event - _LOGGER.debug("<<< %s event", event_type) - if event_type == 'open' or event_type == 'keep-alive': - pass - elif event_type == 'put': - queue.appendleft(json.loads(event.data)) - update_event.set() - elif event_type == 'auth_revoked': - raise AuthorizationError(None, - msg='Auth token has been revoked') - elif event_type == 'error': - raise APIError(None, msg=event.data) - - if not ready_event.is_set(): - ready_event.set() - except requests.exceptions.ConnectionError: - _LOGGER.warning("Haven't received data from Nest in 5 mintues") - finally: - _LOGGER.debug("Stopping event loop.") - queue.clear() - update_event.set() - try: - response.close() - except Exception: - pass - _LOGGER.info("Event loop stopped.") - - def _request(self, verb, path="/", data=None): - url = "%s%s" % (API_URL, path) - _LOGGER.debug(">> %s %s", verb, url) - - if data is not None: - data = json.dumps(data) - - response = self._session.request(verb, url, - allow_redirects=False, - data=data) - _LOGGER.debug("<< %s", response.status_code) - if response.status_code == 200: - return response.json() - - if response.status_code == 401: - raise AuthorizationError(response) - - # Rate Limit Exceeded Catch - if response.status_code == 429: - response = self._handle_ratelimit(response, verb, url, data, - max_retries=10, - default_wait=5) - - # Prevent this from catching as APIError - if response.status_code == 200: - return response.json() - - # This will handle the error if max_retries is exceeded - if response.status_code != 307: - raise APIError(response) - - redirect_url = response.headers['Location'] - _LOGGER.debug(">> %s %s", verb, redirect_url) - response = self._session.request(verb, redirect_url, - allow_redirects=False, - data=data) + _LOGGER.debug(f"<< {r.status_code}") + if r.status_code == 200: + return r.json() + if r.status_code != 401: + raise APIError(r) + except TokenExpiredError as e: + # most providers will ask you for extra credentials to be passed along + # when refreshing tokens, usually for authentication purposes. + extra = { + 'client_id': self._client_id, + 'client_secret': self._client_secret, + } + _LOGGER.debug(">> refreshing token") + token = self._client.refresh_token( + ACCESS_TOKEN_URL, **extra) + self.__save_token(token) + if attempt > 1: + raise AuthorizationError( + None, 'Repeated TokenExpiredError') + continue + self.__reauthorize() + + def _put(self, path, data=None): + pieces = path.split('/') + path = '/' + pieces[-1] + return self._request('POST', path, data=data) + + @property + def _api_url(http://github.com/self): + return API_URL.format(project_id=self._project_id) - _LOGGER.debug("<< %s", response.status_code) - # Rate Limit Exceeded Catch - if response.status_code == 429: - response = self._handle_ratelimit(response, verb, redirect_url, - data, max_retries=10, - default_wait=5) - - # This will handle the error if max_retries is exceeded - if 400 <= response.status_code < 600: - raise APIError(response) - - return response.json() - - def _get(self, path="/"): - return self._request('GET', path) - - def _put(self, path="/", data=None): - return self._request('PUT', path, data=data) + @property + def _devices(self): + if time.time() > self._last_update + self._cache_period: + with self._lock: + self._devices_value = self._request('GET', '')['devices'] + self._last_update = time.time() + return self._devices_value + + def get_devices(self, names: Optional[List[str]] = None, + wheres: Optional[List[str]] = None, + types: Optional[List[str]] = None) -> List[Device]: + """Return the list of devices on this account that match the specified criteria + + Parameters + ---------- + names : Optional[List[str]] + return devices that have names that appear in this list if not None + wheres : Optional[List[str]] + return devices that have where values that appear in this list if not None + types : Optional[List[str]] + return devices that have types that appear in this list if not None + """ + ret = [] + for device in self._devices: + obj = Device(device_data=device) + name_match = (names is None or obj.name in names) + where_match = (wheres is None or obj.where in wheres) + type_match = (types is None or obj.type in types) + if name_match and where_match and type_match: + ret.append(Device(nest_api=self, name=obj.name)) + return ret def __enter__(self): return self @@ -1879,75 +319,57 @@ def __enter__(self): def __exit__(self, exc_type, exc_value, traceback): return False - @property - def _status(self): - self._queue_lock.acquire() - if len(self._queue) == 0 or not self._queue[0]: - try: - _LOGGER.info("Open data stream") - self._open_data_stream("/") - except AuthorizationError as authorization_error: - self._queue_lock.release() - raise authorization_error - except Exception as error: - # other error still set update_event to trigger retry - _LOGGER.debug("Exception occurred in processing stream:" - " %s", error) - self._queue.clear() - self._update_event.set() - self._queue_lock.release() - - value = self._queue[0]['data'] if len(self._queue) > 0 else {} - return value - - @property - def _metadata(self): - return self._status.get(METADATA, {}) - @property - def client_version(self): - return self._metadata.get('client_version') +class APIError(Exception): + def __init__(self, response, msg=None): + if response is None: + response_content = b'' + else: + try: + response_content = response.content + except AttributeError: + response_content = response.data - @property - def _devices(self): - return self._status.get(DEVICES, {}) + if response_content != b'': + if isinstance(response, requests.Response): + try: + message = response.json()['error'] + except: + message = response_content + else: + message = "API Error Occured" - @property - def devices(self): - raise NotImplementedError("Use thermostats instead") + if msg is not None: + message = "API Error Occured: " + msg - @property - def thermostats(self): - return [Thermostat(devid, self) - for devid in self._devices.get(THERMOSTATS, [])] + # Call the base class constructor with the parameters it needs + super(APIError, self).__init__(message) - @property - def protectdevices(self): - raise NotImplementedError("Use smoke_co_alarms instead") + self.response = response - @property - def smoke_co_alarms(self): - return [SmokeCoAlarm(devid, self) - for devid in self._devices.get(SMOKE_CO_ALARMS, [])] - @property - def cameradevices(self): - raise NotImplementedError("Use cameras instead") +class AuthorizationError(Exception): + def __init__(self, response, msg=None): + if response is None: + response_content = b'' + else: + try: + response_content = response.content + except AttributeError: + response_content = response.data - @property - def cameras(self): - return [Camera(devid, self) - for devid in self._devices.get(CAMERAS, [])] + if response_content != b'': + if isinstance(response, requests.Response): + message = response.json().get( + 'error_description', + "Authorization Failed") + else: + message = "Authorization failed" - @property - def structures(self): - return [Structure(stid, self) - for stid in self._status.get(STRUCTURES, [])] + if msg is not None: + message = "Authorization Failed: " + msg - @property - def urls(self): - raise NotImplementedError("Deprecated Nest API") + # Call the base class constructor with the parameters it needs + super(AuthorizationError, self).__init__(message) - @property - def user(self): - raise NotImplementedError("Deprecated Nest API") + self.response = response diff --git a/nest/utils.py b/nest/utils.py deleted file mode 100644 index 08b0972..0000000 --- a/nest/utils.py +++ /dev/null @@ -1,29 +0,0 @@ -# -*- coding:utf-8 -*- - -import decimal - -CELSIUS = 'C' -FAHRENHEIT = 'F' -_THIRTYTWO = decimal.Decimal(32) -_ONEPOINTEIGHT = decimal.Decimal(18) / decimal.Decimal(10) -_TENPOINTSEVENSIXFOUR = decimal.Decimal(10764) / decimal.Decimal(1000) - - -def f_to_c(temp): - temp = decimal.Decimal(temp) - return float((temp - _THIRTYTWO) / _ONEPOINTEIGHT) - - -def c_to_f(temp): - temp = decimal.Decimal(temp) - return float(temp * _ONEPOINTEIGHT + _THIRTYTWO) - - -def ft2_to_m2(area): - area = decimal.Decimal(area) - return float(area / _TENPOINTSEVENSIXFOUR) - - -def m2_to_ft2(area): - area = decimal.Decimal(area) - return float(area * _TENPOINTSEVENSIXFOUR) diff --git a/setup.py b/setup.py index 7771327..08f7373 100755 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding:utf-8 -*- import io @@ -6,28 +6,33 @@ from setuptools import setup -# NOTE(jkoelker) Subjective guidelines for Major.Minor.Micro ;) # Bumping Major means an API contract change. # Bumping Minor means API bugfix or new functionality. # Bumping Micro means CLI change of any kind unless it is # significant enough to warrant a minor/major bump. -version = '4.1.0' +version = '5.2.1' -setup(name='python-nest', +setup(name='python-google-nest', version=version, description='Python API and command line tool for talking to the ' - 'Nest™ Thermostat', - long_description=io.open('README.rst', encoding='UTF-8').read(), + 'Nest™ Thermostat through new Google API', + long_description_content_type="text/markdown", + long_description=io.open('README.md', encoding='UTF-8').read(), keywords='nest thermostat', - author='Jason Kölker', - author_email='jason@koelker.net', - url='https://github.com/jkoelker/python-nest/', + author='Jonathan Diamond', + author_email='feros32@gmail.com', + url='https://github.com/axlan/python-nest/', + classifiers=[ + "Programming Language :: Python :: 3", + "Operating System :: OS Independent", + ], + python_requires=">=3.6", packages=['nest'], - install_requires=['requests>=1.0.0', - 'six>=1.10.0', - 'sseclient-py', - 'python-dateutil'], + install_requires=[ + # Tested with requests_oauthlib==1.3.0 + 'requests_oauthlib' + ], entry_points={ 'console_scripts': ['nest=nest.command_line:main'], }