wordclock: finish implementing mqtt dimming support

.gitignore

@@ -1 +1,2 @@
 *.stamp
+local_secrets.py

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "third_party/micropython"]
+	path = third_party/micropython
+	url = https://github.com/micropython/micropython.git

src/Makefile

@@ -11,11 +11,25 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-SRC = $(wildcard *.py)
+SRC = $(wildcard *.py) $(wildcard lib/*.py)
 
-put: $(SRC:%.py=put-%.stamp)
+MPY_CROSS = ../third_party/micropython/mpy-cross/build/mpy-cross
+
+put: $(SRC:%.py=build/put-%.stamp)
 	ampy -p /dev/ttyUSB? ls
 
-put-%.stamp: %.py
-	ampy -p /dev/ttyUSB? put $*.py
-	cp -a $< $@
+build/put-%.stamp: %.py
+	mkdir -p $(@D)
+	ampy -p /dev/ttyUSB? put $< $(<F)
+	touch -r $< $@
+
+build/%.mpy: %.py $(MPY_CROSS)
+	mkdir -p $(@D)
+	$(MPY_CROSS) -O1 -march=xtensa -o $@ $<
+
+$(MPY_CROSS):
+	$(MAKE) -C $(dir $(@D))
+
+clean:
+	rm -rf build
+

src/asynced.py

@@ -13,21 +13,25 @@
 # limitations under the License.
 import time
 
+import compat
 
-def delay(secs: float) -> generator:
-    end = time.monotonic() + secs
-    while time.monotonic() < end:
+from typing import Generator
+
+
+def delay(secs: float) -> Generator[None, None, None]:
+    end = compat.monotonic() + secs
+    while compat.monotonic() < end:
         yield None
 
 
-def fps(fps: float) -> generator:
+def fps(fps: float) -> Generator[None, None, None]:
     yield None
     dt = 1 / fps
-    until = time.monotonic() + dt
+    until = compat.monotonic() + dt
     while True:
-        remain = until - time.monotonic()
+        remain = until - compat.monotonic()
         if remain > 0:
-            time.sleep(remain)
+            compat.sleep(remain)
         until += dt
         if remain < -dt:
             # Catch up a bit
@@ -35,16 +39,12 @@ def fps(fps: float) -> generator:
         yield None
 
 
-def wait(stop: int) -> generator:
-    start = time.monotonic()
+def wait(stop_ms: int) -> Generator[int, None, None]:
+    """Waits for stop_ms to pass, yielding how many ms have passed so far."""
+    start = compat.monotonic()
     while True:
-        elapsed = int((time.monotonic() - start) * 1000)
-        elapsed = min(stop, elapsed)
+        elapsed = int((compat.monotonic() - start) * 1000)
+        elapsed = min(stop_ms, elapsed)
         yield elapsed
-        if elapsed >= stop:
+        if elapsed >= stop_ms:
             break
-
-
-def test():
-    for f in delay(2.0):
-        print(utime.ticks_ms())

src/compat.py

@@ -0,0 +1,29 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import random
+import time
+
+import utime
+
+
+def monotonic() -> float:
+    return utime.ticks_ms() * 1e-3
+
+
+def sleep(secs: float) -> None:
+    utime.sleep_ms(int(secs * 1e3))
+
+
+def randint(start: int, limit: int) -> int:
+    return start + random.getrandbits(30) % (limit - start + 1)

src/iptime.py

@@ -12,32 +12,31 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import random
-import urllib.urequest
+import time
+from typing import Generator, Optional
+
+import urequests
 
 import asynced
-import ujson
-import time
+import compat
 
 
-def _fetch():
+def _fetch() -> Optional[dict]:
     try:
-        resp = urllib.urequest.urlopen('https://juju.nz/api/iptime/now')
-        t = ujson.load(resp)
-        resp.close()
-        return t
+        return urequests.get("http://worldtimeapi.org/api/ip").json()
     except OSError as ex:
-        print(ex)
+        print("OSError", ex)
         return None
     except ValueError as ex:
-        print(ex)
+        print("ValueError", ex)
         return None
 
 
 def _jitter(mid: int) -> int:
-    return random.randint(mid, mid * 120 // 100)
+    return compat.randint(mid, mid * 120 // 100)
 
 
-def _sync() -> generator:
+def _sync() -> Generator[Optional[dict], None, None]:
     while True:
         # Poll quickly until the first result comes in.
         while True:
@@ -45,7 +44,7 @@ def _sync() -> generator:
             if got is not None:
                 yield got
                 break
-            yield from asynced.delay(_jitter(15))
+            yield from asynced.delay(_jitter(5))
 
         # Poll slowly until the connection drops.
         while True:
@@ -56,37 +55,48 @@ def _sync() -> generator:
             yield got
 
 
-def day_sec() -> generator:
+def _get_day_sec(resp):
+    parts = resp.get("datetime", "").split("T")
+    if len(parts) != 2:
+        return None
+    hms = parts[1].split("+")[0].split(":")
+    if len(hms) != 3:
+        return None
+    return float(hms[0]) * 3600 + float(hms[1]) * 60 + float(hms[2])
+
+
+def day_sec() -> Generator[Optional[float], None, None]:
     s = _sync()
     # Spin until the first result comes in.
     for got in s:
         if got is None:
-            yield None, None
+            yield None
             continue
-        local = time.monotonic()
-        base = got.get('day_sec', None)
+        local = compat.monotonic()
+        base = _get_day_sec(got)
         if base is not None:
             break
     good = got
 
+    assert base is not None
     for got in s:
-        now = base + time.monotonic() - local
-        yield now % (60 * 60 * 24), good
+        now = base + compat.monotonic() - local
+        yield now % (60 * 60 * 24)
 
         if got is not None:
             # Update the baseline.
-            b2 = got.get('day_sec', None)
+            b2 = _get_day_sec(got)
             if b2 is not None:
-                local = time.monotonic()
+                local = compat.monotonic()
                 base = b2
                 good = got
 
 
 def test():
-    for secs, meta in day_sec():
+    for secs in day_sec():
         print(secs)
-        time.sleep(0.3)
+        compat.sleep(0.3)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     test()

src/lib/typing.py

@@ -0,0 +1,7 @@
+Generator = None
+
+Optional = None
+
+Tuple = None
+
+Iterable = None

src/main.py

@@ -13,18 +13,24 @@
 # limitations under the License.
 import time
 
-import microcontroller
+import machine
+import network
 
+import compat
 import wordclock
 
 
 def main():
-    time.sleep(1)
+    compat.sleep(1)
     wordclock.main()
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     try:
         main()
+    except Exception as ex:
+        print(ex)
     finally:
-        microcontroller.reset()
+        print("Resetting")
+        compat.sleep(10)
+        machine.reset()

src/mqtt.py

@@ -0,0 +1,166 @@
+import socket
+
+import asynced
+
+from typing import Generator, Optional, Tuple
+
+_CONNECT = 1
+_CONNACK = 2
+_PUBLISH = 3
+_PUBACK = 4
+_PUBREC = 5
+_PUBREL = 6
+_PUBCOMP = 7
+_SUBSCRIBE = 8
+_SUBACK = 9
+_UNSUBSCRIBE = 10
+_UNSUBACK = 11
+_PINGREQ = 12
+_PINGRESP = 13
+_DISCONNECT = 14
+
+_DUP = 8
+_QOS = 2
+_RETAIN = 1
+
+_UPDATE_SECONDS = 120
+
+
+class Message:
+    def __init__(self, topic: str, value: str):
+        self.topic = topic
+        self.value = value
+
+    topic: str
+    value: str
+
+
+def _getch() -> Generator[None, Optional[int], int]:
+    """Returns the first non-None value sent to the generator"""
+    while True:
+        got = yield None
+        if got is not None:
+            return got
+
+
+def _get_packet() -> Generator[None, Optional[int], Tuple[int, bytearray]]:
+    """Decodes and returns the next packet"""
+    control = yield from _getch()
+    size = 0
+    shift = 0
+    while True:
+        got = yield from _getch()
+        size |= (got & 0x7F) << shift
+        shift += 7
+        if (got & 0x80) == 0:
+            break
+    payload = bytearray(size)
+    for i in range(len(payload)):
+        payload[i] = yield from _getch()
+    return control, payload
+
+
+def _encode_string(value: str) -> bytearray:
+    """Encodes a length-value string"""
+    encoded = value.encode()
+    return bytearray((len(encoded) >> 8, len(encoded) & 0xFF)) + encoded
+
+
+def _write_packet(
+    sock,
+    control: int,
+    payload: bytearray | bytes,
+    packet_id: Optional[int] = None,
+) -> None:
+    if packet_id is not None:
+        header = bytearray(
+            (control, len(payload) + 2, (packet_id >> 8) & 0xFF, packet_id & 0xFF)
+        )
+    else:
+        header = bytearray((control, len(payload)))
+    print("mqtt: >", header, payload)
+    sock.write(header + payload)
+
+
+def _send_connect(sock: socket.socket, username: str, password: str):
+    flags = 0x02
+    if username:
+        flags |= 0x80
+    if password:
+        flags |= 0x40
+    payload = (
+        _encode_string("MQTT")
+        + b"\x04"  # Protocol level
+        + bytearray((flags,))
+        + bytearray((0, _UPDATE_SECONDS * 2))  # Keep alive seconds
+        + _encode_string("")  # Client identifier
+    )
+    if username:
+        payload += _encode_string(username)
+    if password:
+        payload += _encode_string(password)
+    _write_packet(sock, _CONNECT << 4, payload)
+
+
+def _send_subscribe(sock: socket.socket, topic: str):
+    packet_id = hash(topic) & 0xFFFF
+    _write_packet(
+        sock,
+        (_SUBSCRIBE << 4) | 2,
+        _encode_string(topic) +
+        # QOS
+        b"\x00",
+        packet_id,
+    )
+
+
+def _send_publish(sock: socket.socket, topic: str, value: str):
+    _write_packet(sock, _PUBLISH << 4, _encode_string(topic) + value.encode())
+
+
+def _sender(sock: socket.socket, path: str) -> Generator[None, None, None]:
+    _send_subscribe(sock, path + "/#")
+
+    while True:
+        _send_publish(sock, path + "/available", "online")
+        yield from asynced.delay(_UPDATE_SECONDS)
+
+
+def _receiver() -> Generator[Optional[Message], Optional[int], None]:
+    while True:
+        control, payload = yield from _get_packet()
+        if (control >> 4) == _PUBLISH:
+            qos = (control // _QOS) & 3
+            topic_length = (payload[0] << 8) | payload[1]
+            topic = payload[2 : 2 + topic_length].decode()
+            has_packet_id = qos == 1 or qos == 2
+            packet_id_length = 2 if has_packet_id else 0
+            value = payload[2 + topic_length + packet_id_length :].decode()
+            print("mqtt: topic update:", topic, value)
+            yield Message(topic, value)
+
+
+def client(
+    sock: socket.socket,
+    path: str,
+    username: str = "",
+    password: str = "",
+) -> Generator[Optional[Message], Optional[int], None]:
+    _send_connect(sock, username, password)
+
+    connack, payload = yield from _get_packet()
+    if connack != _CONNACK << 4:
+        raise Exception("Expected CONNACK")
+
+    if len(payload) < 2:
+        raise Exception("CONNACK is too short")
+    if payload[1] != 0:
+        raise Exception("CONNACK has an error code")
+
+    receiver = _receiver()
+    sender = _sender(sock, path)
+
+    got = yield None
+    while True:
+        sender.send(None)
+        got = yield receiver.send(got)

src/wordclock.py

@@ -12,17 +12,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import array
-import random
-import time
+import binascii
+import sys
+import socket
 
-import asynced
-import iptime
-
-from neopixel_write import neopixel_write
-import board
 import machine
 import neopixel
+import network
+import micropython
 
+import asynced
+import compat
+import iptime
+import mqtt
+import local_secrets
+
+from typing import Optional, Generator, Iterable, Tuple
+
+if "ESP32" in str(sys.implementation):
+    LED_BUILTIN = 2
+else:
+    LED_BUILTIN = 16
+
+# fmt: off
 RANGES = {
     'dude': 0,
     'ok': 5,
@@ -75,9 +87,11 @@ AMPM = ('AM', 'PM')
 ITS = ('its', 'it is', 'its dude', 'it is dude', 'its ok', 'it is ok')
 
 COLOUR = (200, 255, 200)
+# fmt: on
 
 
-def split(secs: int) -> tuple:
+def _split(secs: int) -> Tuple[str, str, str]:
+    """Converts a seconds-of-day into minutes, hours, and AM/PM"""
     secs = int(secs)
     mins = (secs // 60) % 60
     minute = mins // 5
@@ -97,19 +111,17 @@ class Frame:
     Max = 255
     Zeros = tuple([0] * (11 * 10))
 
-    def __init__(self, n: int = 11 * 10, pixels: list = None):
-        if pixels is None:
-            pixels = list(self.Zeros)
-        self.pixels = pixels
+    def __init__(self, n: int = 11 * 10, pixels: Optional[array.array] = None):
+        self.pixels = pixels if pixels else list(self.Zeros)
 
-    def set(self, idx: int) -> 'Frame':
-        self.pixels[idx] = self.Max
+    def set(self, idx: int, value: int) -> "Frame":
+        self.pixels[idx] = value
         return self
 
     def __add__(self, other):
-        return Frame(pixels=[
-            min(self.Max, x + y) for x, y in zip(self.pixels, other.pixels)
-        ])
+        return Frame(
+            pixels=[min(self.Max, x + y) for x, y in zip(self.pixels, other.pixels)]
+        )
 
     def muldiv(self, num: int, den: int):
         return Frame(pixels=[x * num // den for x in self.pixels])
@@ -117,10 +129,11 @@ class Frame:
 
 @micropython.native
 def merge(f1: Frame, f2: Frame, num1: int, num2: int, den: int) -> Frame:
+    """Merges two frames together."""
     m = Frame.Max
     p1 = f1.pixels
     p2 = f2.pixels
-    o = array.array('H', Frame.Zeros)
+    o = array.array("H", Frame.Zeros)
     for i in range(len(p1)):
         k = (p1[i] * num1 + p2[i] * num2) // den
         if k >= m:
@@ -129,116 +142,196 @@ def merge(f1: Frame, f2: Frame, num1: int, num2: int, den: int) -> Frame:
     return Frame(pixels=o)
 
 
-def render(words: list) -> Frame:
+def _render(words: Iterable[str], brightness: int) -> Frame:
+    """Renders a list of words into a frame"""
     f = Frame()
     for word in words:
         for w in word.split():
             idx = RANGES.get(w, None)
             if idx is not None:
                 for i in range(len(w)):
-                    f.set(idx + i)
+                    f.set(idx + i, brightness)
             else:
-                print('warning: %s is missing' % w)
+                print("warning: %s is missing" % w)
     return f
 
 
-def prefix() -> tuple:
-    return (ITS[random.randint(0, len(ITS) - 1)], )
+def _prefix() -> Tuple[str]:
+    """Returns a random suffix word such as 'dude'"""
+    return (ITS[compat.randint(0, len(ITS) - 1)],)
 
 
-def show(f: Frame, n):
+def _show(f: Frame, n):
+    """Renders a frame to a NeoPixel handler"""
     p = f.pixels
     buf = bytearray(len(p) * 3)
     r, g, b = COLOUR
     i = 0
 
     for c in p:
-        buf[i + 1] = (r * c) >> 8
-        buf[i + 0] = (g * c) >> 8
-        buf[i + 2] = (b * c) >> 8
+        n.buf[i + 1] = (r * c) >> 8
+        n.buf[i + 0] = (g * c) >> 8
+        n.buf[i + 2] = (b * c) >> 8
         i += 3
-    neopixel_write(n.pin, buf)
+    n.write()
 
 
-def scan() -> generator:
+def _scan() -> Generator[Frame, None, None]:
+    """Returns a stream of frames that scan across the LEDs"""
     f = Frame()
     while True:
         for i in range(len(f.pixels)):
-            yield f.set(i)
+            yield f.set(i, Frame.Max)
             f = f.muldiv(80, 100)
 
 
-def intensity(num: int, den: int) -> int:
+def _intensity(num: int, den: int) -> int:
     idx = num * len(CIEL8) // den
     if idx >= len(CIEL8):
         return 255
     return CIEL8[idx]
 
 
-def brightness(secs):
-    hour = int(secs) // 60 // 60
-    if hour >= 7 and hour < 22:
-        return 255
-
-    return 1200
+def _crossfade(src: Frame, dest: Frame, fade_time: int):
+    for at in asynced.wait(fade_time):
+        yield merge(
+            src,
+            dest,
+            _intensity(fade_time - at, fade_time),
+            _intensity(at, fade_time),
+            Frame.Max,
+        )
 
 
-def run() -> generator:
+def _run(brightness_queue: list[int]) -> Generator[Optional[Frame], None, None]:
     words = None
-    fade = 2000
+    fade_time = 2000
     secs = iptime.day_sec()
+    brightness = 255
 
     # Flash until time is synced.
-    sc = scan()
-    for s, meta in secs:
-        if s is not None and meta is not None:
+    sc = _scan()
+    for s in secs:
+        if s is not None:
             break
         yield next(sc)
 
     src = Frame()
     while True:
-        yield None
-        s, m2 = next(secs)
-        nxt = split(s)
-        if nxt == words:
+        changed = False
+        if brightness_queue:
+            brightness = brightness_queue[-1]
+            brightness_queue.clear()
+            changed = True
+
+        now = next(secs)
+        next_words = _split(now)
+        if words != next_words:
+            words = next_words
+            changed = True
+
+        if changed:
+            next_frame = _render(_prefix() + next_words, brightness)
+            yield from _crossfade(src, next_frame, fade_time)
+            src = next_frame
+        else:
             yield from asynced.delay(2)
-            continue
-        dest = render(prefix() + nxt)
-        for at in asynced.wait(fade):
-            yield merge(src, dest, intensity(fade - at, fade),
-                        intensity(at, fade), brightness(s))
-        words = nxt
-        src = dest
 
 
-def blink() -> generator:
-    led = machine.Pin(16, machine.Pin.OUT)
+def blink() -> Generator[None, None, None]:
+    led = machine.Pin(LED_BUILTIN, machine.Pin.OUT)
     while True:
-        led.off()
-        yield from asynced.delay(0.03)
         led.on()
+        yield from asynced.delay(0.03)
+        led.off()
         yield from asynced.delay(2.9)
 
 
-def bench() -> generator:
+def bench() -> Generator[None, None, None]:
     yield None
     i = 30
     while True:
-        start = time.monotonic()
-        yield from range(i)
-        elapsed = time.monotonic() - start
-        print('%d in %.3f %.3f/1 %.1f FPS' %
-              (i, elapsed, elapsed / i, i / elapsed))
+        start = compat.monotonic()
+        for _ in range(i):
+            yield None
+        elapsed = compat.monotonic() - start
+        if elapsed > 0:
+            print("%d in %.3f %.3f/1 %.1f FPS" % (i, elapsed, elapsed / i, i / elapsed))
+
+
+def _connect(wlan: network.WLAN) -> Generator[None, None, None]:
+    yield None
+
+    while True:
+        if not wlan.isconnected():
+            wlan.active(True)
+            wlan.connect(local_secrets.WLAN_ESSID, local_secrets.WLAN_PASSWORD)
+        else:
+            # Override the DNS server as MicroPython and dnsmasq seem
+            # to be incompatible.
+            cfg = list(wlan.ifconfig())
+            wlan.ifconfig(cfg[:3] + ["8.8.8.8"])
+        yield from asynced.delay(10)
+
+
+def _mqtt_client(
+    is_connected, brightness_queue: list[int]
+) -> Generator[None, None, None]:
+    while True:
+        yield from asynced.delay(3)
+
+        if not is_connected():
+            continue
+
+        sock = socket.socket()
+        try:
+            target = socket.getaddrinfo(local_secrets.MQTT_HOST, 1883)[0][-1]
+            print("mqtt: connecting to", target)
+            sock.connect(target)
+            sock.setblocking(False)
+
+            topic = "light/wc-" + binascii.hexlify(machine.unique_id()).decode()
+            client = mqtt.client(
+                sock, topic, local_secrets.MQTT_USERNAME, local_secrets.MQTT_PASSWORD
+            )
+            while True:
+                got = sock.read()
+                if not got:
+                    client.send(None)
+                    yield None
+                    continue
+
+                for ch in got:
+                    message = client.send(ch)
+                    if not message:
+                        continue
+                    client.send(None)
+                    if message.topic.endswith("/brightness"):
+                        brightness_queue.append(int(message.value))
+        except Exception as ex:
+            print("mqtt: exception:", ex)
+        finally:
+            print("mqtt: close")
+            sock.close()
 
 
 def main():
     num = 11 * 10
-    n = neopixel.NeoPixel(board.GPIO2, num, brightness=1, auto_write=False)
-    routines = (run(), blink(), asynced.fps(30), bench())
+    n = neopixel.NeoPixel(machine.Pin(2), num)
+    wlan = network.WLAN(network.STA_IF)
+    brightness_queue = []
+    routines = (
+        _connect(wlan),
+        _mqtt_client(wlan.isconnected, brightness_queue),
+        _run(brightness_queue),
+        blink(),
+        asynced.fps(30),
+        bench(),
+    )
     while True:
         for r in routines:
             f = next(r)
             if f is None:
                 continue
             elif isinstance(f, Frame):
-                show(f, n)
+                _show(f, n)

third_party/micropython

@@ -0,0 +1 @@
+Subproject commit 678707c8b07323c5b914778708a2858387c3b60c