Skip to content

Instantly share code, notes, and snippets.

@ggravlingen
Created November 15, 2017 21:15
Show Gist options
  • Save ggravlingen/459787e88bab4785fbd1a982970dba1f to your computer and use it in GitHub Desktop.
Save ggravlingen/459787e88bab4785fbd1a982970dba1f to your computer and use it in GitHub Desktop.
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