Initial upload

Author: FRC4564

hue.py

@@ -0,0 +1,231 @@
+import socket
+from time import sleep
+import json
+try:
+    import requests
+except:
+    import urequests as requests
+    
+# UPnP SSDP Search request header
+HEADER = b"""M-SEARCH * HTTP/1.1\r
+HOST: 239.255.255.250:1900\r
+MAN: "ssdp:discover"\r
+ST: ssdp:all\r
+MX: 3\r
+\r
+"""
+
+class Bridge:
+    """Provides methods for connecting to and using Hue Bridge.  Supports
+    Micropython, Python 2, and 3."""
+    
+    def __init__(self,autosetup=True, debug=1):
+        self.debug = debug  #0=no prints, 1=messages, 2=debug
+        self.IP = None
+        self.username = None
+        if autosetup:
+            self.setup()
+
+
+    def show(self,str,level=1):
+        """ Show debug output. """
+        if self.debug >= level:
+            print(str)
+            
+
+    def setup(self):
+        """ Loads bridge settings or attempts to establish them, if needed."""
+        success = self.loadSettings()
+        if success:
+            # verify bridge settings work
+            try:
+                self.idLights()
+                success = True
+            except:
+                success = False
+        if not success:
+            if self.discover():
+                self.show('Bridge located at {}'.format(self.IP))
+                self.show('>>> Press link button on Hue bridge to register <<<')
+                if self.getUsername():
+                    success = self.saveSettings()
+                else:
+                    self.show("Couldn't get username from bridge.")                
+            else:
+                self.show("Couldn't find bridge on LAN.")
+        return success
+
+
+    def discover(self):
+        """ Locate Hue Bridge IP using UPnP SSDP search. Discovery will return
+        when bridge is found or 3 seconds after last device response. Returns IP
+        address or None."""
+        #On ESP8266, disable AP WLAN to force use of STA interface
+        #import network
+        #ap = network.WLAN(network.AP_IF)
+        #ap.active(False)
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.sendto(HEADER, ('239.255.255.250',1900))    #UPnP Multicast
+        s.settimeout(3)
+
+        IP = None
+        while IP == None:
+            data, addr = s.recvfrom(1024)
+            self.show(str(data),2)
+            lines = data.split(b'\r\n')
+            for l in lines:
+                tokens = l.split(b' ')
+                if tokens[0] == b'SERVER:':
+                    product = tokens[3].split(b'/')
+                    if product[0] == b'IpBridge':
+                        IP = str(addr[0])
+                        break
+
+        s.close()
+        self.IP = IP
+        return IP
+
+
+    def getUsername(self):
+        """ Get a developer API username from bridge.
+        Requires that the bridge link button be pressed sometime while polling.
+        Polls for 20 seconds (20 attempts at 1 second intervals).
+        Can timeout with error if bridge is non-responsive.
+        Returns username on success or None on failure."""      
+        url = 'http://{}/api'.format(self.IP)
+        data = '{"devicetype":"TapLight#mydevice"}'
+        username = None
+        count = 20
+        while count > 0 and username == None:
+            resp = requests.post(url,data=data)
+            if resp.status_code == 200:
+                j = resp.json()[0]
+                self.show(j,2)
+                if j.get('success'):
+                    username = str(j['success']['username'])
+                    self.username = username
+            sleep(1)
+            count -= 1
+        return username
+
+
+    def saveSettings(self): 
+        """ Save bridge IP and username to bridge.dat file.
+        Returns True on success."""
+        if self.IP and self.username:
+            f=open('bridge.dat','w')
+            f.write(json.dumps([self.IP,self.username]))
+            f.close()
+            return True
+        else:
+            return None
+
+
+    def loadSettings(self):
+        """ Load bridge IP and username from bridge.dat file and set base URL.
+        Returns True on success. """
+        try:
+            f=open('bridge.dat')
+        except:
+            return None
+        l = json.load(f)
+        f.close()
+        self.IP = str(l[0])
+        self.username = str(l[1])
+        self.show('Loaded settings {} {}'.format(self.IP,self.username),2)
+        return True
+
+
+    def resetSettings(self):
+        """Delete current saved bridge settings and reinitiate."""
+        from os import remove
+        remove('bridge.dat')
+        self.IP = None
+        self.username = None
+        self.setup()
+        
+
+    def url(self,path):
+        """Return url for API calls."""
+        return 'http://{}/api/{}/{}'.format(self.IP,self.username,path)
+
+
+    def get(self, path):
+        """Perform GET request and return json result."""
+        url = self.url(path)
+        self.show(url,2)
+        resp = requests.get(url).json()
+        self.show(resp,2)
+        return resp
+
+
+    def put(self, path, data):
+        """Perform PUT request and return response."""
+        url = self.url(path)
+        self.show(url,2)
+        data = json.dumps(data)
+        self.show(data,2)
+        resp = requests.put(url, data=data).json()
+        self.show(resp,2)
+        return resp
+
+
+    def allLights(self):
+        """Returns dictionary containing all lights, with detail."""
+        """Large return set, not ideal for controllers with limited RAM."""
+        return self.get('lights')
+
+
+    def idLights(self):
+        """Returns list of all light IDs."""
+        ids = self.get('groups/0')['lights']
+        for i in range(len(ids)):
+            ids[i] = int(ids[i])
+        return ids
+
+
+    def getLight(self,id):
+        """Returns dictionary of light details for given ID."""
+        return self.get('lights/{}'.format(str(id)))
+
+
+    def getLights(self):
+        """Iterates through each light to build and return a dictionary 
+        of light IDs and names."""
+        dict = {}
+        for i in self.idLights():
+            dict[i] = str(self.getLight(i)['name'])
+        return dict
+            
+
+    def setLight(self,id,**kwargs):
+        """Set one or more states of a light.  
+        Ex: setLight(1,on=True,bri=254,hue=50000,sat=254)"""
+        self.put('lights/{}/state'.format(str(id)),kwargs)
+
+
+    def allGroups(self):
+        """Returns dictionary containing all groups, with detail."""
+        return self.get('groups')
+
+
+    def getGroup(self,id):
+        """Returns dictionary of group details."""
+        return self.get('groups/{}'.format(str(id)))
+
+
+    def getGroups(self):
+        """Returns dictionary of group IDs and names."""
+        dict = {}
+        groups = self.allGroups()
+        for g in groups:
+            dict[int(g)] = str(groups[g]['name'])
+        return dict
+    
+
+    def setGroup(self,id,**kwargs):
+        """Set one or more states of a group.
+        Ex: setGroup(1,bri_inc=100,transitiontime=40)"""
+        self.put('groups/{}/action'.format(str(id)),kwargs)
+        
+