Created
November 15, 2017 21:15
-
-
Save ggravlingen/459787e88bab4785fbd1a982970dba1f to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import logging | |
import sys | |
import argparse | |
import uuid | |
import time | |
from pytradfri import Gateway | |
from pytradfri.api.aiocoap_api import APIFactory | |
import logging | |
_LOGGER = logging.getLogger(__name__) | |
parser = argparse.ArgumentParser() | |
parser.add_argument('-H', '--hostname', dest='host', required=True, | |
help='IP Address of your Tradfri gateway') | |
parser.add_argument('-K', '--key', dest='key', required=False, | |
help='Key found on your Tradfri gateway') | |
args = parser.parse_args() | |
CONFIG_FILE = 'tradfri_standalone_psk.conf' | |
"""JSON utility functions.""" | |
from typing import Union, List, Dict | |
import json | |
root = logging.getLogger() | |
root.setLevel(logging.INFO) | |
try: | |
# pylint: disable=ungrouped-imports | |
from asyncio import ensure_future | |
except ImportError: | |
# Python 3.4.3 and earlier has this as async | |
# pylint: disable=unused-import | |
from asyncio import async | |
ensure_future = async | |
def load_json(filename: str) -> Union[List, Dict]: | |
"""Load JSON data from a file and return as dict or list. | |
Defaults to returning empty dict if file is not found. | |
""" | |
try: | |
with open(filename, encoding='utf-8') as fdesc: | |
return json.loads(fdesc.read()) | |
except FileNotFoundError: | |
# This is not a fatal error | |
_LOGGER.debug('JSON file not found: %s', filename) | |
except ValueError as error: | |
_LOGGER.exception('Could not parse JSON content: %s', filename) | |
raise HomeAssistantError(error) | |
except OSError as error: | |
_LOGGER.exception('JSON file reading failed: %s', filename) | |
raise HomeAssistantError(error) | |
return {} # (also evaluates to False) | |
def save_json(filename: str, config: Union[List, Dict]): | |
"""Save JSON data to a file. | |
Returns True on success. | |
""" | |
try: | |
data = json.dumps(config, sort_keys=True, indent=4) | |
with open(filename, 'w', encoding='utf-8') as fdesc: | |
fdesc.write(data) | |
return True | |
except TypeError as error: | |
_LOGGER.exception('Failed to serialize to JSON: %s', | |
filename) | |
raise HomeAssistantError(error) | |
except OSError as error: | |
_LOGGER.exception('Saving JSON file failed: %s', | |
filename) | |
raise HomeAssistantError(error) | |
return False | |
@asyncio.coroutine | |
def run(): | |
conf = load_json(CONFIG_FILE) | |
try: | |
identity = conf[args.host].get('identity') | |
psk = conf[args.host].get('key') | |
api_factory = APIFactory(host=args.host, psk_id=identity, psk=psk) | |
except KeyError: | |
identity = uuid.uuid4().hex | |
api_factory = APIFactory(host=args.host, psk_id=identity) | |
try: | |
psk = yield from api_factory.generate_psk(args.key) | |
print('Generated PSK: ', psk) | |
conf[args.host] = {'identity': identity, | |
'key': psk} | |
save_json(CONFIG_FILE, conf) | |
except TypeError: | |
print("asdasd") | |
api = api_factory.request | |
gateway = Gateway() | |
devices_command = gateway.get_devices() | |
devices_commands = yield from api(devices_command) | |
devices = yield from api(devices_commands) | |
lights = [dev for dev in devices if dev.has_light_control] | |
# Print all lights | |
print(lights) | |
def observe_callback(updated_device): | |
light = updated_device.light_control.lights[0] | |
print("Received message for: %s" % light) | |
def observe_err_callback(err): | |
print('observe error:', err) | |
yield from asyncio.sleep(120) | |
asyncio.get_event_loop().run_until_complete(run()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment