michaelh/wordclock
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Compare commits

base: michaelh:d537de05fc43d4a4c6d1a5c2c60a4cd45336e223
Branches Tags
michaelh:master
michaelh:v1
michaelh:v0.0
..
compare: michaelh:ffead18f6e2c5f63200b9c9184e03817eab015d6
Branches Tags
michaelh:master
michaelh:v1
michaelh:v0.0

No commits in common. "d537de05fc43d4a4c6d1a5c2c60a4cd45336e223" and "ffead18f6e2c5f63200b9c9184e03817eab015d6" have entirely different histories.
d537de05fc ... ffead18f6e

6 changed files with 53 additions and 121 deletions
Show stats Expand all files Collapse all files

1
.gitignore vendored
View file

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

16
src/asynced.py
View file

@ -13,23 +13,21 @@
# limitations under the License. # limitations under the License.
import time import time
import compat
def delay(secs: float) -> generator: def delay(secs: float) -> generator:
end = compat.monotonic() + secs end = time.monotonic() + secs
while compat.monotonic() < end: while time.monotonic() < end:
yield None yield None
def fps(fps: float) -> generator: def fps(fps: float) -> generator:
yield None yield None
dt = 1 / fps dt = 1 / fps
until = compat.monotonic() + dt until = time.monotonic() + dt
while True: while True:
remain = until - compat.monotonic() remain = until - time.monotonic()
if remain > 0: if remain > 0:
compat.sleep(remain) time.sleep(remain)
until += dt until += dt
if remain < -dt: if remain < -dt:
# Catch up a bit # Catch up a bit
@ -38,9 +36,9 @@ def fps(fps: float) -> generator:
def wait(stop: int) -> generator: def wait(stop: int) -> generator:
start = compat.monotonic() start = time.monotonic()
while True: while True:
elapsed = int((compat.monotonic() - start) * 1000) elapsed = int((time.monotonic() - start) * 1000)
elapsed = min(stop, elapsed) elapsed = min(stop, elapsed)
yield elapsed yield elapsed
if elapsed >= stop: if elapsed >= stop:

27
src/compat.py
View file

@ -1,27 +0,0 @@
# 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
def monotonic():
return time.ticks_ms() * 1e-3
def sleep(secs):
time.sleep_ms(int(secs * 1e3))
def randint(start, limit):
return start + random.getrandbits(30) % (limit - start + 1)

44
src/iptime.py
View file

@ -12,27 +12,29 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import random import random
import time import urllib.urequest
import urequests
import asynced import asynced
import compat import ujson
import time
def _fetch(): def _fetch():
try: try:
return urequests.get("http://worldtimeapi.org/api/ip").json() resp = urllib.urequest.urlopen('https://juju.nz/api/iptime/now')
t = ujson.load(resp)
resp.close()
return t
except OSError as ex: except OSError as ex:
print("OSError", ex) print(ex)
return None return None
except ValueError as ex: except ValueError as ex:
print("ValueError", ex) print(ex)
return None return None
def _jitter(mid: int) -> int: def _jitter(mid: int) -> int:
return compat.randint(mid, mid * 120 // 100) return random.randint(mid, mid * 120 // 100)
def _sync() -> generator: def _sync() -> generator:
@ -43,7 +45,7 @@ def _sync() -> generator:
if got is not None: if got is not None:
yield got yield got
break break
yield from asynced.delay(_jitter(5)) yield from asynced.delay(_jitter(15))
# Poll slowly until the connection drops. # Poll slowly until the connection drops.
while True: while True:
@ -54,16 +56,6 @@ def _sync() -> generator:
yield got yield got
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: def day_sec() -> generator:
s = _sync() s = _sync()
# Spin until the first result comes in. # Spin until the first result comes in.
@ -71,21 +63,21 @@ def day_sec() -> generator:
if got is None: if got is None:
yield None, None yield None, None
continue continue
local = compat.monotonic() local = time.monotonic()
base = _get_day_sec(got) base = got.get('day_sec', None)
if base is not None: if base is not None:
break break
good = got good = got
for got in s: for got in s:
now = base + compat.monotonic() - local now = base + time.monotonic() - local
yield now % (60 * 60 * 24), good yield now % (60 * 60 * 24), good
if got is not None: if got is not None:
# Update the baseline. # Update the baseline.
b2 = _get_day_sec(got) b2 = got.get('day_sec', None)
if b2 is not None: if b2 is not None:
local = compat.monotonic() local = time.monotonic()
base = b2 base = b2
good = got good = got
@ -93,8 +85,8 @@ def day_sec() -> generator:
def test(): def test():
for secs, meta in day_sec(): for secs, meta in day_sec():
print(secs) print(secs)
compat.sleep(0.3) time.sleep(0.3)
if __name__ == "__main__": if __name__ == '__main__':
test() test()

14
src/main.py
View file

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

72
src/wordclock.py
View file

@ -13,18 +13,16 @@
# limitations under the License. # limitations under the License.
import array import array
import random import random
import secrets
import time import time
import machine
import neopixel
import network
import asynced import asynced
import compat
import iptime import iptime
# fmt: off from neopixel_write import neopixel_write
import board
import machine
import neopixel
RANGES = { RANGES = {
'dude': 0, 'dude': 0,
'ok': 5, 'ok': 5,
@ -77,7 +75,6 @@ AMPM = ('AM', 'PM')
ITS = ('its', 'it is', 'its dude', 'it is dude', 'its ok', 'it is ok') ITS = ('its', 'it is', 'its dude', 'it is dude', 'its ok', 'it is ok')
COLOUR = (200, 255, 200) COLOUR = (200, 255, 200)
# fmt: on
def split(secs: int) -> tuple: def split(secs: int) -> tuple:
@ -105,14 +102,14 @@ class Frame:
pixels = list(self.Zeros) pixels = list(self.Zeros)
self.pixels = pixels self.pixels = pixels
def set(self, idx: int) -> "Frame": def set(self, idx: int) -> 'Frame':
self.pixels[idx] = self.Max self.pixels[idx] = self.Max
return self return self
def __add__(self, other): def __add__(self, other):
return Frame( return Frame(pixels=[
pixels=[min(self.Max, x + y) for x, y in zip(self.pixels, other.pixels)] min(self.Max, x + y) for x, y in zip(self.pixels, other.pixels)
) ])
def muldiv(self, num: int, den: int): def muldiv(self, num: int, den: int):
return Frame(pixels=[x * num // den for x in self.pixels]) return Frame(pixels=[x * num // den for x in self.pixels])
@ -123,7 +120,7 @@ def merge(f1: Frame, f2: Frame, num1: int, num2: int, den: int) -> Frame:
m = Frame.Max m = Frame.Max
p1 = f1.pixels p1 = f1.pixels
p2 = f2.pixels p2 = f2.pixels
o = array.array("H", Frame.Zeros) o = array.array('H', Frame.Zeros)
for i in range(len(p1)): for i in range(len(p1)):
k = (p1[i] * num1 + p2[i] * num2) // den k = (p1[i] * num1 + p2[i] * num2) // den
if k >= m: if k >= m:
@ -141,12 +138,12 @@ def render(words: list) -> Frame:
for i in range(len(w)): for i in range(len(w)):
f.set(idx + i) f.set(idx + i)
else: else:
print("warning: %s is missing" % w) print('warning: %s is missing' % w)
return f return f
def prefix() -> tuple: def prefix() -> tuple:
return (ITS[compat.randint(0, len(ITS) - 1)],) return (ITS[random.randint(0, len(ITS) - 1)], )
def show(f: Frame, n): def show(f: Frame, n):
@ -156,11 +153,11 @@ def show(f: Frame, n):
i = 0 i = 0
for c in p: for c in p:
n.buf[i + 1] = (r * c) >> 8 buf[i + 1] = (r * c) >> 8
n.buf[i + 0] = (g * c) >> 8 buf[i + 0] = (g * c) >> 8
n.buf[i + 2] = (b * c) >> 8 buf[i + 2] = (b * c) >> 8
i += 3 i += 3
n.write() neopixel_write(n.pin, buf)
def scan() -> generator: def scan() -> generator:
@ -208,13 +205,8 @@ def run() -> generator:
continue continue
dest = render(prefix() + nxt) dest = render(prefix() + nxt)
for at in asynced.wait(fade): for at in asynced.wait(fade):
yield merge( yield merge(src, dest, intensity(fade - at, fade),
src, intensity(at, fade), brightness(s))
dest,
intensity(fade - at, fade),
intensity(at, fade),
brightness(s),
)
words = nxt words = nxt
src = dest src = dest
@ -232,33 +224,17 @@ def bench() -> generator:
yield None yield None
i = 30 i = 30
while True: while True:
start = compat.monotonic() start = time.monotonic()
yield from range(i) yield from range(i)
elapsed = compat.monotonic() - start elapsed = time.monotonic() - start
if elapsed > 0: print('%d in %.3f %.3f/1 %.1f FPS' %
print("%d in %.3f %.3f/1 %.1f FPS" % (i, elapsed, elapsed / i, i / elapsed)) (i, elapsed, elapsed / i, i / elapsed))
def connect(wlan) -> generator:
yield None
while True:
if not wlan.isconnected():
wlan.active(True)
wlan.connect(secrets.WLAN_ESSID, 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 main(): def main():
num = 11 * 10 num = 11 * 10
n = neopixel.NeoPixel(machine.Pin(2), num) n = neopixel.NeoPixel(board.GPIO2, num, brightness=1, auto_write=False)
wlan = network.WLAN(network.STA_IF) routines = (run(), blink(), asynced.fps(30), bench())
routines = (connect(wlan), run(), blink(), asynced.fps(30), bench())
while True: while True:
for r in routines: for r in routines:
f = next(r) f = next(r)