busylight/ESP32/main.py

from microdot import Microdot, send_file
import machine, neopixel, time
import uasyncio as asyncio

app = Microdot()

# =========================
# NeoPixel setup
# =========================
nbPixels = 12 * 2
pinPixelStrip = 16
neoPixelStrip = neopixel.NeoPixel(machine.Pin(pinPixelStrip), nbPixels)


# =========================
# LED CONTROLLER
# =========================
class LedController:
    def __init__(self, np):
        self._np = np

        # async control
        self._flag = asyncio.ThreadSafeFlag()
        self._cmd = None
        self._task = None

        # state
        self.states = {
            'BUSY': (255, 0, 0),
            'AVAILABLE': (0, 255, 0),
            'AWAY': (246, 190, 0),
            'OFF': (0, 0, 0),
            'ON': (255, 255, 255)
        }

        self.color = self.states['OFF']
        self.status = 'off'
        self.previousStatus = 'off'
        self.brightness = 0.1

        self.blinking = False
        self.blinkFrequency = 0
        self.blinkDuration = 0
        self.blinkStart = 0

    # ---------- internal ----------
    def _apply_brightness(self, color):
        r, g, b = color
        return (
            int(r * self.brightness),
            int(g * self.brightness),
            int(b * self.brightness),
        )

    def _write(self, color):
        self._np.fill(self._apply_brightness(color))
        self._np.write()

    async def run(self):
        while True:
            await self._flag.wait()
            cmd = self._cmd

            # Cancel any ongoing task before starting a new one
            if self._task:
                self._task.cancel()
                try:
                    await self._task
                except asyncio.CancelledError:
                    pass

            # Process the new command
            t = cmd["type"]

            if t == "blink":
                self._task = asyncio.create_task(self._blink(cmd))
            elif t == "solid":
                self._task = asyncio.create_task(self._solid(cmd))
            elif t == "off":
                self._task = asyncio.create_task(self._off())

    async def _solid(self, cmd):
        self.color = cmd["color"]
        self.status = cmd.get("status", "colored")
        self._write(self.color)

    async def _off(self):
        self.color = (0, 0, 0)
        self.status = "off"
        self._write(self.color)

    async def _blink(self, cmd):
        try:
            self.blinking = True

            self.color = cmd["color"]
            self.previousStatus = cmd["previousStatus"]
            self.blinkFrequency = cmd["frequency"]
            self.blinkDuration = cmd["duration"]
            self.blinkStart = time.ticks_ms()

            freq = max(0.1, self.blinkFrequency)
            half_ms = int(500 / freq)

            while True:
                if self.blinkDuration != 0:
                    elapsed = time.ticks_diff(time.ticks_ms(), self.blinkStart)
                    if elapsed > self.blinkDuration * 1000:
                        break

                self._write(self.color)
                await asyncio.sleep_ms(half_ms)

                self._write((0, 0, 0))
                await asyncio.sleep_ms(half_ms)

        except asyncio.CancelledError:
            pass

        self.blinking = False
        self.status = self.previousStatus
        self._write(self.color)

    async def send(self, cmd):
        self._cmd = cmd
        self._flag.set()

    # ---------- public API ----------
    async def set_color(self, color, brightness=None):
        if brightness is not None:
            self.brightness = brightness

        await self.send({
            "type": "solid",
            "color": color
        })

    async def set_status(self, status):
        status = status.upper()
        color = self.states.get(status)

        if color is None:
            raise ValueError("invalid status")

        await self.send({
            "type": "solid",
            "color": color,
            "status": status.lower()
        })

    async def blink(self, frequency, duration):
        self.previousStatus = self.status

        await self.send({
            "type": "blink",
            "color": self.color,
            "frequency": frequency,
            "duration": duration,
            "previousStatus": self.previousStatus
        })

    async def stop(self):
        await self.send({"type": "off"})

    def get_state(self):
        if self.blinking and self.blinkDuration > 0:
            elapsed = time.ticks_diff(time.ticks_ms(), self.blinkStart) / 1000
            remains = max(0.0, self.blinkDuration - elapsed)
        else:
            remains = 0.0

        return {
            "status": self.status,
            "color": self.color,
            "brightness": self.brightness,
            "isblinking": self.blinking,
            "frequency": self.blinkFrequency,
            "duration": self.blinkDuration,
            "remains": remains
        }


# =========================
# INIT
# =========================
led = LedController(neoPixelStrip)


# =========================
# ROUTES
# =========================

@app.get('/static/<path:path>')
async def staticRoutes(request, path):
    if '..' in path:
        return {'error': '404 Not found'}, 404
    return send_file('static/' + path)


@app.get('/')
async def getIndex(request):
    return send_file('static/index.html')


@app.get('/api/brightness')
async def getBrightness(request):
    return {'brightness': led.brightness}


@app.post('/api/brightness')
async def setBrightness(request):
    brightness = request.json.get("brightness")

    if brightness is None:
        return {'error': 'missing brightness parameter'}, 400

    if not isinstance(brightness, (int, float)):
        return {'error': 'wrong brightness type'}, 400

    if brightness < 0 or brightness > 1:
        return {'error': 'brightness out of bound'}, 400

    current_status = led.status
    current_color = led.color

    await led.set_color(current_color, brightness)

    led.status = current_status

    return {'brightness': led.brightness}


@app.post('/api/color')
async def setColor(request):
    r = request.json.get("r")
    g = request.json.get("g")
    b = request.json.get("b")

    if r is None or g is None or b is None:
        return {'error': 'missing color'}, 400

    if not all(isinstance(v, int) for v in (r, g, b)):
        return {'error': 'wrong color type'}, 400

    if not (0 <= r <= 255 and 0 <= g <= 255 and 0 <= b <= 255):
        return {'error': 'color out of bound'}, 400

    brightness = request.json.get("brightness")

    await led.set_color((r, g, b), brightness)

    return {'status': led.status}


@app.get('/api/color')
async def getColor(request):
    r, g, b = led.color
    return {'r': r, 'g': g, 'b': b, 'brightness': led.brightness}


@app.get('/api/status')
async def getStatus(request):
    return {'status': led.status}


@app.route('/api/status/<status>', methods=['GET', 'POST'])
async def setStatus(request, status):
    try:
        await led.set_status(status)
    except ValueError:
        return {'error': 'unknown status'}, 404

    return {'status': led.status}


@app.post('/api/blink')
async def setBlink(request):
    freq = request.json.get('frequency')
    duration = request.json.get('duration')

    if freq is None or duration is None:
        return {'error': 'missing frequency or duration'}, 400

    if not isinstance(freq, (int, float)) or freq <= 0:
        return {'error': 'invalid frequency'}, 400

    if not isinstance(duration, (int, float)) or duration < 0:
        return {'error': 'invalid duration'}, 400

    await led.blink(freq, duration)

    return {'status': 'blinking', 'frequency': freq, 'duration': duration}


@app.get('/api/blink')
async def getBlink(request):
    state = led.get_state()
    return {
        'isblinking': state["isblinking"],
        'frequency': state["frequency"],
        'duration': state["duration"],
        'remains': state["remains"]
    }


@app.post('/api/blink/stop')
async def blinkStop(request):
    await led.stop()
    await led.set_status(led.previousStatus)
    return {'status': led.status}


@app.get('/api/debug')
async def getDebugInfo(request):
    r, g, b = led.color
    dr, dg, db = led._np[0]

    return {
        'status': led.status,
        'brightness': led.brightness,
        'color': {'r': r, 'g': g, 'b': b},
        'dimColor': {'r': dr, 'g': dg, 'b': db},
        'blinking': led.blinking,
        'blinkFrequency': led.blinkFrequency,
        'blinkDuration': led.blinkDuration
    }


@app.post('/api/mutedeck-webhook')
async def mutedeckWebhook(request):

    if request.json.get('control') != 'system':
        if request.json.get('call') == 'active':
            isMuted = request.json.get('mute') == 'active'

            if isMuted:
                await led.set_status('away')
            else:
                await led.set_status('busy')
    else:
        await led.set_status('available')

    return {'status': led.status}


@app.post('/shutdown')
async def shutdown(request):
    request.app.shutdown()
    return 'The server is shutting down...'


# =========================
# START
# =========================
async def main():
    asyncio.create_task(led.run())

    # startup animation
    await led.set_status('busy')
    await asyncio.sleep_ms(200)
    await led.set_status('away')
    await asyncio.sleep_ms(300)
    await led.set_status('available')
    await asyncio.sleep_ms(500)
    await led.set_status('off')

    await app.start_server(port=80)


asyncio.run(main())