homecontrol-neopixel: work with homecontrol
This commit is contained in:
commit
1367690f70
6 changed files with 174 additions and 0 deletions
2
.gitignore
vendored
Normal file
2
.gitignore
vendored
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
.idea/*
|
||||||
|
Pipfile.lock
|
14
Pipfile
Normal file
14
Pipfile
Normal file
|
@ -0,0 +1,14 @@
|
||||||
|
[[source]]
|
||||||
|
name = "pypi"
|
||||||
|
url = "https://pypi.org/simple"
|
||||||
|
verify_ssl = true
|
||||||
|
|
||||||
|
[dev-packages]
|
||||||
|
|
||||||
|
[packages]
|
||||||
|
pyserial = ">=3.3,<4.0"
|
||||||
|
docopt = "<0.7,>=0.6.2"
|
||||||
|
adafruit-ampy = ">=1.0.5,<1.1"
|
||||||
|
|
||||||
|
[requires]
|
||||||
|
python_version = "3.7"
|
1
boot.py
Normal file
1
boot.py
Normal file
|
@ -0,0 +1 @@
|
||||||
|
# not doing anything
|
112
main.py
Normal file
112
main.py
Normal file
|
@ -0,0 +1,112 @@
|
||||||
|
import machine
|
||||||
|
import neopixel
|
||||||
|
import utime as time
|
||||||
|
import urequests as requests
|
||||||
|
import uuid
|
||||||
|
import ujson
|
||||||
|
import math
|
||||||
|
import wifi
|
||||||
|
|
||||||
|
CONFIG = "config.json"
|
||||||
|
ADDRESS = "http://192.168.11.21:5000"
|
||||||
|
TIMEOUT_UPDATE = 1
|
||||||
|
|
||||||
|
|
||||||
|
class ActorNP:
|
||||||
|
def __init__(self):
|
||||||
|
self.config = {}
|
||||||
|
try:
|
||||||
|
config_file = open(CONFIG, "r")
|
||||||
|
self.config = ujson.load(config_file)
|
||||||
|
except Exception as ex:
|
||||||
|
print("Exception\n\ttype: %s\n\targs: %s" % (type(ex).__name__, ex.args))
|
||||||
|
|
||||||
|
update_config = False
|
||||||
|
if not self.config.get("id"):
|
||||||
|
update_config = True
|
||||||
|
self.config["id"] = uuid.uuid4().hex
|
||||||
|
if not self.config.get("type"):
|
||||||
|
update_config = True
|
||||||
|
self.config["type"] = "neopixel"
|
||||||
|
if not self.config.get("levels"):
|
||||||
|
update_config = True
|
||||||
|
self.config["levels"] = 0xffffff
|
||||||
|
if update_config:
|
||||||
|
with open(CONFIG, "w") as config_file:
|
||||||
|
ujson.dump(self.config, config_file)
|
||||||
|
|
||||||
|
self.id_a = self.config["id"]
|
||||||
|
self.type_a = self.config["type"]
|
||||||
|
self.levels = self.config["levels"]
|
||||||
|
self.color = (0, 0, 0)
|
||||||
|
self.np = neopixel.NeoPixel(machine.Pin(2), 16)
|
||||||
|
|
||||||
|
|
||||||
|
def update(self):
|
||||||
|
url = "%s/actors/update" % ADDRESS
|
||||||
|
level = ((self.color[0] << 16) + (self.color[1] << 8) + self.color[2]) & 0xffffff
|
||||||
|
data = {"id": self.id_a, "type": self.type_a, "levels": self.levels, "level": level}
|
||||||
|
try:
|
||||||
|
res = requests.post(url, json=data)
|
||||||
|
content = res.json()
|
||||||
|
if "command" in content:
|
||||||
|
command = content["command"]
|
||||||
|
if command == "set_level":
|
||||||
|
data = content["data"]
|
||||||
|
level = int(data["level"])
|
||||||
|
self.set_level(level)
|
||||||
|
# remember to close
|
||||||
|
res.close()
|
||||||
|
except Exception as ex:
|
||||||
|
print("Exception\n\tdata: %s\n\ttype: %s\n\targs: %s" % (data, type(ex).__name__, ex.args))
|
||||||
|
|
||||||
|
|
||||||
|
def set_level(self, level):
|
||||||
|
print("set_level: level: 0x%06x" % (level))
|
||||||
|
t = 1
|
||||||
|
steps = 25
|
||||||
|
t_step = t/steps
|
||||||
|
br_step = 1/steps
|
||||||
|
# turn off
|
||||||
|
br = 1
|
||||||
|
r = self.color[0]
|
||||||
|
g = self.color[1]
|
||||||
|
b = self.color[2]
|
||||||
|
if (r or g or b):
|
||||||
|
for i in range(steps):
|
||||||
|
br = br - br_step
|
||||||
|
self.np.fill(self.col(r, g, b, br))
|
||||||
|
self.np.write()
|
||||||
|
time.sleep(t_step)
|
||||||
|
|
||||||
|
# turn on
|
||||||
|
r = (level >> 16) & 0xff
|
||||||
|
g = (level >> 8) & 0xff
|
||||||
|
b = level & 0xff
|
||||||
|
br = 0
|
||||||
|
if (r or g or b):
|
||||||
|
for i in range(steps):
|
||||||
|
br = br + br_step
|
||||||
|
self.np.fill(self.col(r, g, b, br))
|
||||||
|
self.np.write()
|
||||||
|
time.sleep(t_step)
|
||||||
|
|
||||||
|
self.color = (r, g, b)
|
||||||
|
|
||||||
|
|
||||||
|
def col(self, r, g, b, br):
|
||||||
|
rr = int(math.ceil(r * br))
|
||||||
|
gg = int(math.ceil(g * br))
|
||||||
|
bb = int(math.ceil(b * br))
|
||||||
|
c = (rr, gg, bb)
|
||||||
|
return(c)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
w = wifi.wifi()
|
||||||
|
w.connect()
|
||||||
|
np = ActorNP()
|
||||||
|
while True:
|
||||||
|
np.update()
|
||||||
|
time.sleep(TIMEOUT_UPDATE)
|
||||||
|
|
28
uuid.py
Normal file
28
uuid.py
Normal file
|
@ -0,0 +1,28 @@
|
||||||
|
import uos
|
||||||
|
import ubinascii
|
||||||
|
|
||||||
|
|
||||||
|
class UUID:
|
||||||
|
def __init__(self, bytes):
|
||||||
|
if len(bytes) != 16:
|
||||||
|
raise ValueError('bytes arg must be 16 bytes long')
|
||||||
|
self._bytes = bytes
|
||||||
|
|
||||||
|
@property
|
||||||
|
def hex(self):
|
||||||
|
return ubinascii.hexlify(self._bytes).decode()
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
h = self.hex
|
||||||
|
return '-'.join((h[0:8], h[8:12], h[12:16], h[16:20], h[20:32]))
|
||||||
|
|
||||||
|
def __repr__(self):
|
||||||
|
return "<UUID: %s>" % str(self)
|
||||||
|
|
||||||
|
|
||||||
|
def uuid4():
|
||||||
|
"""Generates a random UUID compliant to RFC 4122 pg.14"""
|
||||||
|
random = bytearray(uos.urandom(16))
|
||||||
|
random[6] = (random[6] & 0x0F) | 0x40
|
||||||
|
random[8] = (random[8] & 0x3F) | 0x80
|
||||||
|
return UUID(bytes=random)
|
17
wifi.py
Normal file
17
wifi.py
Normal file
|
@ -0,0 +1,17 @@
|
||||||
|
import network
|
||||||
|
NET_SSID="omnomnom"
|
||||||
|
NET_PASS="Ne!HabIchNicht!"
|
||||||
|
|
||||||
|
class wifi:
|
||||||
|
def __init__(self):
|
||||||
|
self.if_ap = network.WLAN(network.AP_IF)
|
||||||
|
self.if_sta = network.WLAN(network.STA_IF)
|
||||||
|
|
||||||
|
def connect(self):
|
||||||
|
self.if_ap.active(False)
|
||||||
|
self.if_sta.active(True)
|
||||||
|
self.if_sta.connect(NET_SSID, NET_PASS)
|
||||||
|
while not self.if_sta.isconnected():
|
||||||
|
pass
|
||||||
|
print("network config: ", self.if_sta.ifconfig())
|
||||||
|
|
Loading…
Reference in a new issue