You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
128 lines
3.3 KiB
128 lines
3.3 KiB
try:
|
|
import uasyncio as asyncio
|
|
except ImportError:
|
|
import asyncio
|
|
|
|
import gc
|
|
|
|
from machine import Pin
|
|
from microdot_asyncio import Microdot, Response, send_file
|
|
from pixels import Pixels
|
|
from wifi import status as network_status
|
|
|
|
try:
|
|
from config import config
|
|
except ImportError:
|
|
raise Exception("Cannot open/find config.py")
|
|
|
|
app = Microdot()
|
|
current_task = None
|
|
try:
|
|
p = Pixels(config["pixel_pin"], config["pixel_count"])
|
|
except KeyError:
|
|
raise ValueError("'pixel_pin' or 'pixel_count' does not exist in config.py")
|
|
|
|
|
|
@app.before_request
|
|
async def pre_request_handler(request):
|
|
# Cancel any running preset if light related function is called
|
|
if request.path.startswith("/lights/") and current_task:
|
|
current_task.cancel()
|
|
|
|
|
|
@app.route("/")
|
|
async def hello(request):
|
|
return send_file("index.html")
|
|
|
|
|
|
@app.route("/shutdown")
|
|
async def shutdown(request):
|
|
print("Shutting Down App")
|
|
if current_task:
|
|
current_task.cancel()
|
|
|
|
request.app.shutdown()
|
|
gc.collect()
|
|
return {"message": "The server is shutting down..."}
|
|
|
|
@app.route("/status")
|
|
async def status(request):
|
|
results = {
|
|
"memory": {
|
|
"alloc": gc.mem_alloc(),
|
|
"free": gc.mem_free(),
|
|
"total": gc.mem_alloc() + gc.mem_free(),
|
|
},
|
|
"network": network_status(),
|
|
}
|
|
|
|
return results
|
|
|
|
|
|
@app.route("/lights/off")
|
|
async def lights_off(request):
|
|
p.clear()
|
|
return {"message": "Lights turned off"}
|
|
|
|
@app.route("/lights/on")
|
|
async def lights_on(request):
|
|
p.set_color(r=255, g=255, b=255)
|
|
return {"message": "Lights turned on"}
|
|
|
|
@app.route("/lights/rgb")
|
|
async def lights_rgb(request):
|
|
args = rgb_args(request.args)
|
|
p.set_color(**args)
|
|
return {"message": f"Light colors set: {args}"}
|
|
|
|
@app.route("/lights/preset/bounce")
|
|
async def lights_preset_bounce(request):
|
|
global current_task
|
|
args = rgb_args(request.args)
|
|
args["cycles"] = request.args.get("cycles", -1)
|
|
current_task = asyncio.create_task(p.preset_bounce(**args))
|
|
return {"message": "Running light preset: Bounce"}
|
|
|
|
@app.route("/lights/preset/cycle")
|
|
async def lights_preset_cycle(request):
|
|
global current_task
|
|
args = rgb_args(request.args)
|
|
args["cycles"] = request.args.get("cycles", -1)
|
|
current_task = asyncio.create_task(p.preset_cycle(**args))
|
|
return {"message": "Running light preset: Cycle"}
|
|
|
|
@app.route("/lights/preset/niagra")
|
|
async def lights_preset_niagra(request):
|
|
global current_task
|
|
cycles = request.args.get("cycles", -1)
|
|
current_task = asyncio.create_task(p.preset_niagra(cycles=cycles))
|
|
return {"message": "Running light preset: Niagra"}
|
|
|
|
@app.route("/lights/preset/rainbow")
|
|
async def lights_preset_rainbow(request):
|
|
global current_task
|
|
cycles = request.args.get("cycles", -1)
|
|
current_task = asyncio.create_task(p.preset_rainbow(cycles=cycles))
|
|
return {"message": "Running light preset: Rainbow"}
|
|
|
|
|
|
def rgb_args(args_in):
|
|
args = {}
|
|
for key in args_in.keys():
|
|
try:
|
|
args[key] = int(args_in[key])
|
|
except ValueError:
|
|
continue
|
|
return args
|
|
|
|
def start_server():
|
|
print("Starting Microdot App")
|
|
try:
|
|
app.run(port=80)
|
|
except KeyboardInterrupt:
|
|
print("Shutting Down App")
|
|
app.shutdown()
|
|
gc.collect()
|
|
|
|
|
|
start_server()
|