test yeelight ok

Author: Marty

Diff for: esp01_mlx90614_yeelight.py

@@ -0,0 +1,27 @@
+import time
+import mlx90614
+from machine import I2C, Pin
+from uyeelight import *
+
+bulbs = Bulb.search(timeout=1)
+
+if len(bulbs)==0:
+    raise Exception("bulb not found.")
+
+ip = list(bulbs.keys())[0]
+bulb = Bulb(ip)
+bulb.turn_on()
+bulb.set_rgb(255,255,255, effect=EFFECT.SUDDEN, duration=0.1)
+time.sleep(1)
+bulb.set_rgb(255,100,0, effect=EFFECT.SUDDEN, duration=0.1)
+
+
+i2c = I2C(scl=Pin(2), sda=Pin(0))
+sensor = mlx90614.MLX90614(i2c)
+
+while True:
+    ambient = sensor.read_ambient_temp()
+    objTemp = sensor.read_object_temp()
+    print(objTemp*10)
+    bulb.set_rgb(255,0,objTemp*10, effect=EFFECT.SUDDEN, duration=0.1)
+    time.sleep_ms(100)

Diff for: esp01_ssd1306.py

@@ -0,0 +1,9 @@
+from machine import Pin,I2C
+import ssd1306
+ 
+i2c = I2C(scl=Pin(0), sda=Pin(2), freq=100000)  #Init i2c
+lcd=ssd1306.SSD1306_I2C(64,48,i2c) #create LCD object,Specify col and row
+lcd.text("ESP8266",0,0)
+lcd.text("test",0,16)
+lcd.text("123456",0,32)
+lcd.show()

Diff for: esp01_yeelight.py

@@ -0,0 +1,11 @@
+from uyeelight import *
+
+bulbs = Bulb.search(timeout=1)
+
+if len(bulbs)==0:
+    raise Exception("bulb not found.")
+
+ip = list(bulbs.keys())[0]
+bulb = Bulb(ip)
+bulb.turn_on()
+bulb.set_rgb(255,0,25, effect=EFFECT.SUDDEN, duration=0.1)

Diff for: lib/uyeelight.py

@@ -1,5 +1,5 @@
 import usocket as socket
-import json
+import json, struct
 
 
 class YeeLightException(Exception):
@@ -232,6 +232,24 @@ def _send_message(self, method, params=None):
 
         return recv_data
 
+    def search(timeout=2):
+        msg = "\r\n".join(["M-SEARCH * HTTP/1.1", "HOST: 239.255.255.250:1982", 'MAN: "ssdp:discover"', "ST: wifi_bulb"])
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind(('', 1982))
+        s.sendto(msg.encode(), ('239.255.255.250', 1982))
+        s.settimeout(timeout)
+        bulbs = {}
+        while True:
+            try:
+                data, addr = s.recvfrom(1024)
+                capabilities = dict([x.strip("\r").split(": ") for x in data.decode().split("\n") if ":" in x])
+                key = capabilities['Location'].split(':')[1][2:]
+                bulbs[key] = capabilities
+            except:
+                break
+        return bulbs
+
     def _handle_response(self, response):
         response = json.loads(response.decode('utf-8'))
 
@@ -245,4 +263,4 @@ def _handle_response(self, response):
         elif "error" in response:
             raise YeeLightException(response["error"])
         else:
-            raise YeeLightException("Unknown Exception occurred.")
+            raise YeeLightException("Unknown Exception occurred.")

Diff for: main.py

@@ -1,67 +1,11 @@
-import struct, socket
 from uyeelight import *
 
-# see yeelight app for IP
-bulb = Bulb("192.168.0.95")
+bulbs = Bulb.search(timeout=1)
 
-try:
-    if not bulb.is_on:
-        bulb.turn_on()
-except YeeLightException as e:
-    print(e)
-
-bulb.set_rgb(255,0,255, effect=EFFECT.SUDDEN, duration=1)
-
-def discover_bulbs(timeout=2, interface=False):
-    """
-    Discover all the bulbs in the local network.
-    :param int timeout: How many seconds to wait for replies. Discovery will
-                        always take exactly this long to run, as it can't know
-                        when all the bulbs have finished responding.
-    :param string interface: The interface that should be used for multicast packets.
-                             Note: it *has* to have a valid IPv4 address. IPv6-only
-                             interfaces are not supported (at the moment).
-                             The default one will be used if this is not specified.
-    :returns: A list of dictionaries, containing the ip, port and capabilities
-              of each of the bulbs in the network.
-    """
-    msg = "\r\n".join(["M-SEARCH * HTTP/1.1", "HOST: 239.255.255.250:1982", 'MAN: "ssdp:discover"', "ST: wifi_bulb"])
-    MCAST_GRP = "239.255.255.250"
-    MCAST_PORT= 1982
-    # Set up UDP socket
-    """
-    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    s.bind((MCAST_GRP, MCAST_PORT))
-    # '\xc0\xa8\x01\x01' socket.inet_aton(MCAST_GRP)
-    mreq = struct.pack("4sl",'\xef\xff\xff\xfa' , socket.INADDR_ANY)
-    s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
-
-    s.sendto(msg.encode(), (MCAST_GRP, MCAST_PORT))
-    """
-    
-    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    s.bind(('', 1982))
-    s.sendto(msg.encode(), ('239.255.255.250', 1982))    
-    
-    bulbs = []
-    bulb_ips = set()
-    while True:
-        data, addr = s.recvfrom(65507)
-
-        capabilities = dict([x.strip("\r").split(": ") for x in data.decode().split("\n") if ":" in x])
-        parsed_url = urlparse(capabilities["Location"])
-
-        bulb_ip = (parsed_url.hostname, parsed_url.port)
-        if bulb_ip in bulb_ips:
-            continue
-
-        capabilities = {key: value for key, value in capabilities.items() if key.islower()}
-        bulbs.append({"ip": bulb_ip[0], "port": bulb_ip[1], "capabilities": capabilities})
-        bulb_ips.add(bulb_ip)
-
-    return bulbs
-
-print(discover_bulbs())
+if len(bulbs)==0:
+    raise Exception("bulb not found.")
 
+ip = list(bulbs.keys())[0]
+bulb = Bulb(ip)
+bulb.turn_on()
+bulb.set_rgb(255,0,25, effect=EFFECT.SUDDEN, duration=0.1)