#!/usr/bin/env python3

import json

import requests

import sys

import os

import time

import socket

import copy

from pprint import pprint

prefs = {}

prefs_old = None

GROUP_ALL = 0 # A special group containing all lights in the system,

# and is not returned by the ‘get all groups’ command.

class CliFatalError(Exception):

pass

def preffilename():

if sys.platform == 'darwin':

tpl = '~/Library/Preferences/huectl.json'

else:

tpl = '%s/huectl/config.json' % (os.environ.get('XDG_CONFIG_HOME', '~/.config'), )

return os.path.expanduser(tpl)

def prefs_load():

global prefs, prefs_old

try:

with open(preffilename(), 'r') as fp:

prefs = json.load(fp)

prefs_old = copy.deepcopy(prefs)

except:

prefs = {}

def prefs_save():

global prefs, prefs_old

if prefs_old == prefs:

return

with open(preffilename(), 'w') as fp:

json.dump(prefs, fp)

def discover():

"""Find hue bridge using meethue rendezvous point. As this is quite slow,

try a cached IP first."""

global prefs

if 'bridge_cache' in prefs:

try:

bridgeip = prefs['bridge_cache']['ip']

reply = requests.get('http://%s/api/' % (bridgeip), timeout=3).json()

if len(reply) > 0 and 'error' in reply[0] and reply[0]['error']['type'] == 4:

# good bridge, use it

return bridgeip

except requests.exceptions.ConnectTimeout:

# fallback to rendezvous point

pass

print("Discovering bridge...")

try:

bridgeip = requests.get('https://www.meethue.com/api/nupnp').json()[0]['internalipaddress']

prefs['bridge_cache'] = {'ip': bridgeip}

return bridgeip

except Exception as except_inst:

print("Bridge discovery failed:", except_inst)

raise CliFatalError()

def link(bridgeip):

"""Request username from bridge. User has to press the LINK button while

doing this."""

body = json.dumps({'devicetype': 'huectl#' + socket.gethostname()})

maxtries = 30

sys.stdout.write("Press link button on bridge %s now..." % bridgeip)

sys.stdout.flush()

for try_ in range(0, maxtries):

reg = requests.post('http://%s/api' % (bridgeip, ), body).json()

if 'error' in reg[0] and reg[0]['error']['type'] == 101:

sys.stdout.write(".")

sys.stdout.flush()

else:

prefs['username'] = reg[0]['success']['username']

prefs_save()

print()

return

time.sleep(1)

print("\nExpired. Please try again.")

raise CliFatalError()

class Api(object):

def __init__(self, bridgeip, username):

self.url = 'http://%s/api/%s/' % (bridgeip, username)

self.s = requests.Session()

def lights(self):

return self.s.get(self.url + 'lights').json()

def info(self, lightid):

return self.s.get(self.url + ('lights/%d' % (lightid, ))).json()

def set_onoff(self, lightid, onoff):

body = json.dumps({'on': bool(onoff)})

return self.s.put(self.url + ('lights/%d/state' % (lightid, )), body).json()

def dim(self, lightid, pct):

body = json.dumps({'bri': int(pct * 254 / 100), 'on': True})

return self.s.put(self.url + ('lights/%d/state' % (lightid, )), body).json()

def scene(self, sceneid):

return self.s.get(self.url + 'scenes/%s' % (sceneid, )).json()

def scenes(self):

return self.s.get(self.url + 'scenes').json()

def sensors(self):

return self.s.get(self.url + 'sensors').json()

def group_action(self, groupid, changes):

body = json.dumps(changes)

return self.s.put(self.url + ('groups/%s/action' % (groupid, )), body).json()

def recall_scene(self, sceneid):

changes = {'scene': sceneid}

return self.group_action(GROUP_ALL, changes)

def config(self, changes=None):

url = self.url + 'config'

if not changes:

return self.s.get(url).json()

else:

body = json.dumps(changes)

return self.s.put(url, body).json()

def do_swupdate(api):

config = api.config()

print("Current bridge software version:", config['swversion'])

if not config['portalstate']['signedon']:

print("Bridge not connected to portal, swupdate not possible.")

raise CliFatalError()

api.config({'swupdate': {'checkforupdate': True}})

config = api.config()

while config['swupdate']['checkforupdate'] is True:

time.sleep(1)

config = api.config()

if config['swupdate']['updatestate'] == 1:

print("Software download in progress...")

while ['swupdate']['updatestate'] == 1:

time.sleep(1)

config = api.config()

if config['swupdate']['updatestate'] == 2:

print("New software \"%s\" found, starting update..." % (config['swupdate']['text']))

print("Update will affect these components:")

if config['swupdate']['devicetypes']['bridge']:

print(" - bridge")

for light in config['swupdate']['devicetypes']['lights']:

print(" - light %s", (light, ))

for sensor in config['swupdate']['devicetypes']['sensors']:

print(" - sensor %s", (sensor, ))

api.config({'swupdate': {'updatestate': 3}})

elif config['swupdate']['updatestate'] == 0:

print("No new software found.")

return

elif config['swupdate']['updatestate'] == 3:

print("Update already in progress...")

else:

print("Unknown updatestate %s", (config['swupdate']['updatestate'], ))

raise CliFatalError()

while True:

try:

config = api.config()

if config['swupdate']['notify']:

break

time.sleep(1)

except requests.exceptions.ConnectTimeout:

print("Bridge unavailable, waiting...")

time.sleep(3)

print(config['swupdate']['url'])

print(config['swupdate']['text'])

# clear notify flag

api.config({'swupdate': {'notify': False}})

def main():

global prefs

prefs_load()

bridgeip = discover()

if 'username' not in prefs:

link(bridgeip)

api = Api(bridgeip, prefs['username'])

if len(sys.argv) == 1:

command = 'list'

args = []

else:

command = sys.argv[1]

args = sys.argv[2:]

if command == 'list':

items = api.lights()

for id_, detail in sorted(items.items(), key=lambda kv: kv[0]):

print("%-5s %-30s %-25s %s" % (id_, detail['name'], detail['type'], detail['state']))

elif command == 'dim':

light = int(args[0])

pct = int(args[1])

print(api.dim(light, pct))

elif command == 'on':

light = int(args[0])

print(api.set_onoff(light, True))

elif command == 'off':

light = int(args[0])

print(api.set_onoff(light, False))

elif command == 'scene':

print(api.recall_scene(args[0]))

elif command == 'scene-detail':

scene = api.scene(args[0])

flags = [('locked' if scene.get('locked', False) else ''),

('recycle' if scene.get('recycle', True) else ''),

('v' + str(scene.get('version', 1)))]

flags = [x for x in flags if x]

print("ID: ", args[0])

print("Name: ", scene['name'])

print("Lights: ", ' '.join(sorted(scene['lights'])))

print("Flags: ", ' '.join(flags))

if scene.get('owner', 'none') != 'none':

# 1.11+

print("Owner: ", scene['owner'])

if scene.get('lastupdated', None):

# 1.11+, may be null for v1 data

print("Changed: ", scene['lastupdated'])

if scene.get('lightstates', None):

# 1.11+, may be null for v1 data

for id_, detail in sorted(scene.get('lightstates', None).items(), key=lambda kv: kv[0]):

print("%-15s %3s %-5s %-10s %-10s %-10s" % (id_, 'on' if detail['on'] else 'off',

detail.get('bri', ''), detail.get('xy', ''), detail.get('hs', ''), detail.get('ct', '')))

# we don't print appdata, picture (for now)

elif command == 'scenes':

items = api.scenes()

for id_, detail in sorted(items.items(), key=lambda kv: kv[0]):

print("%-15s %-32s %-10s" % (id_, detail['name'], ' '.join(detail['lights'])))

elif command == 'sensors':

items = api.sensors()

for id_, detail in sorted(items.items(), key=lambda kv: kv[0]):

print("%-5s %-30s %-15s %s" % (id_, detail['name'], detail['type'], detail['state']))

elif command == 'swupdate':

do_swupdate(api)

elif command == 'help':

print("Usage:")

print("huectl list - list lights")

print("huectl dim ID pct - dim light ID to percent pct%")

print("huectl on ID - turn on light ID")

print("huectl off ID - turn off light ID")

print("huectl scenes - list scenes")

print("huectl scene ID - activate scene ID")

print("huectl scene-detail ID - get scene ID (bridge v1.11+)")

print("huectl sensors - list sensors")

print("huectl swupdate - update hue system software (no prompting)")

else:

print("Unknown command given.")

raise CliFatalError()

prefs_save()

try:

main()

except CliFatalError:

sys.exit(1)