Remove files found not needed(builtin options). Implement sleeping and catch some errors.
This commit is contained in:
188
inky_helper.py
Normal file
188
inky_helper.py
Normal file
@@ -0,0 +1,188 @@
|
||||
from pimoroni_i2c import PimoroniI2C
|
||||
from pcf85063a import PCF85063A
|
||||
import math
|
||||
from machine import Pin, PWM, Timer
|
||||
import time
|
||||
import inky_frame
|
||||
import json
|
||||
import network
|
||||
import os
|
||||
|
||||
# Pin setup for VSYS_HOLD needed to sleep and wake.
|
||||
HOLD_VSYS_EN_PIN = 2
|
||||
hold_vsys_en_pin = Pin(HOLD_VSYS_EN_PIN, Pin.OUT)
|
||||
|
||||
# intialise the pcf85063a real time clock chip
|
||||
I2C_SDA_PIN = 4
|
||||
I2C_SCL_PIN = 5
|
||||
i2c = PimoroniI2C(I2C_SDA_PIN, I2C_SCL_PIN, 100000)
|
||||
rtc = PCF85063A(i2c)
|
||||
|
||||
led_warn = Pin(6, Pin.OUT)
|
||||
|
||||
# set up for the network LED
|
||||
network_led_pwm = PWM(Pin(7))
|
||||
network_led_pwm.freq(1000)
|
||||
network_led_pwm.duty_u16(0)
|
||||
|
||||
|
||||
# set the brightness of the network led
|
||||
def network_led(brightness):
|
||||
brightness = max(0, min(100, brightness)) # clamp to range
|
||||
# gamma correct the brightness (gamma 2.8)
|
||||
value = int(pow(brightness / 100.0, 2.8) * 65535.0 + 0.5)
|
||||
network_led_pwm.duty_u16(value)
|
||||
|
||||
|
||||
network_led_timer = Timer(-1)
|
||||
network_led_pulse_speed_hz = 1
|
||||
|
||||
|
||||
def network_led_callback(t):
|
||||
# updates the network led brightness based on a sinusoid seeded by the current time
|
||||
brightness = (math.sin(time.ticks_ms() * math.pi * 2 / (1000 / network_led_pulse_speed_hz)) * 40) + 60
|
||||
value = int(pow(brightness / 100.0, 2.8) * 65535.0 + 0.5)
|
||||
network_led_pwm.duty_u16(value)
|
||||
|
||||
|
||||
# set the network led into pulsing mode
|
||||
def pulse_network_led(speed_hz=1):
|
||||
global network_led_timer, network_led_pulse_speed_hz
|
||||
network_led_pulse_speed_hz = speed_hz
|
||||
network_led_timer.deinit()
|
||||
network_led_timer.init(period=50, mode=Timer.PERIODIC, callback=network_led_callback)
|
||||
|
||||
|
||||
# turn off the network led and disable any pulsing animation that's running
|
||||
def stop_network_led():
|
||||
global network_led_timer
|
||||
network_led_timer.deinit()
|
||||
network_led_pwm.duty_u16(0)
|
||||
|
||||
|
||||
def sleep(t):
|
||||
# Time to have a little nap until the next update
|
||||
rtc.clear_timer_flag()
|
||||
rtc.set_timer(t, ttp=rtc.TIMER_TICK_1_OVER_60HZ)
|
||||
rtc.enable_timer_interrupt(True)
|
||||
|
||||
# Set the HOLD VSYS pin to an input
|
||||
# this allows the device to go into sleep mode when on battery power.
|
||||
hold_vsys_en_pin.init(Pin.IN)
|
||||
|
||||
# Regular time.sleep for those powering from USB
|
||||
time.sleep(60 * t)
|
||||
|
||||
|
||||
# Turns off the button LEDs
|
||||
def clear_button_leds():
|
||||
inky_frame.button_a.led_off()
|
||||
inky_frame.button_b.led_off()
|
||||
inky_frame.button_c.led_off()
|
||||
inky_frame.button_d.led_off()
|
||||
inky_frame.button_e.led_off()
|
||||
|
||||
|
||||
def network_connect(SSID, PSK):
|
||||
# Enable the Wireless
|
||||
wlan = network.WLAN(network.STA_IF)
|
||||
wlan.active(True)
|
||||
|
||||
# Number of attempts to make before timeout
|
||||
max_wait = 10
|
||||
|
||||
# Sets the Wireless LED pulsing and attempts to connect to your local network.
|
||||
pulse_network_led()
|
||||
wlan.config(pm=0xa11140) # Turn WiFi power saving off for some slow APs
|
||||
wlan.connect(SSID, PSK)
|
||||
|
||||
while max_wait > 0:
|
||||
if wlan.status() < 0 or wlan.status() >= 3:
|
||||
break
|
||||
max_wait -= 1
|
||||
print('waiting for connection...')
|
||||
time.sleep(1)
|
||||
|
||||
stop_network_led()
|
||||
network_led_pwm.duty_u16(30000)
|
||||
|
||||
# Handle connection error. Switches the Warn LED on.
|
||||
if wlan.status() != 3:
|
||||
stop_network_led()
|
||||
led_warn.on()
|
||||
|
||||
def network_connect_better(SSID, PSK):
|
||||
# Enable the Wireless
|
||||
wlan = network.WLAN(network.STA_IF)
|
||||
wlan.active(True)
|
||||
|
||||
ssids = wlan.scan()
|
||||
for ssid in ssids:
|
||||
if ssid[0].decode('UTF-8') == SSID:
|
||||
# Number of attempts to make before timeout
|
||||
max_wait = 10
|
||||
|
||||
# Sets the Wireless LED pulsing and attempts to connect to your local network.
|
||||
pulse_network_led()
|
||||
wlan.config(pm=0xa11140) # Turn WiFi power saving off for some slow APs
|
||||
wlan.connect(SSID, PSK)
|
||||
|
||||
while max_wait > 0:
|
||||
if wlan.status() < 0 or wlan.status() >= 3:
|
||||
break
|
||||
max_wait -= 1
|
||||
print('waiting for connection...')
|
||||
time.sleep(1)
|
||||
|
||||
stop_network_led()
|
||||
network_led_pwm.duty_u16(30000)
|
||||
|
||||
# Handle connection error. Switches the Warn LED on.
|
||||
if wlan.status() != 3:
|
||||
stop_network_led()
|
||||
led_warn.on()
|
||||
|
||||
return True
|
||||
raise RuntimeError("Can't find SSID")
|
||||
|
||||
|
||||
state = {"run": None}
|
||||
app = None
|
||||
|
||||
|
||||
def file_exists(filename):
|
||||
try:
|
||||
return (os.stat(filename)[0] & 0x4000) == 0
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def clear_state():
|
||||
if file_exists("state.json"):
|
||||
os.remove("state.json")
|
||||
|
||||
|
||||
def save_state(data):
|
||||
with open("/state.json", "w") as f:
|
||||
f.write(json.dumps(data))
|
||||
f.flush()
|
||||
|
||||
|
||||
def load_state():
|
||||
global state
|
||||
data = json.loads(open("/state.json", "r").read())
|
||||
if type(data) is dict:
|
||||
state = data
|
||||
|
||||
|
||||
def update_state(running):
|
||||
global state
|
||||
state['run'] = running
|
||||
save_state(state)
|
||||
|
||||
|
||||
def launch_app(app_name):
|
||||
global app
|
||||
app = __import__(app_name)
|
||||
print(app)
|
||||
update_state(app_name)
|
||||
88
main.py
88
main.py
@@ -1,51 +1,63 @@
|
||||
import uNextcloud
|
||||
import inky_helper as ih
|
||||
import gc
|
||||
import sdcard, uos
|
||||
import inky_helper as ih
|
||||
import inky_frame
|
||||
import gc
|
||||
import random
|
||||
from picographics import PicoGraphics, DISPLAY_INKY_FRAME_7 as DISPLAY
|
||||
from jpegdec import JPEG
|
||||
|
||||
|
||||
sd_spi = machine.SPI(0, sck=machine.Pin(18, machine.Pin.OUT), mosi=machine.Pin(19, machine.Pin.OUT), miso=machine.Pin(16, machine.Pin.OUT))
|
||||
sd = sdcard.SDCard(sd_spi, machine.Pin(22))
|
||||
uos.mount(sd, "/sd")
|
||||
gc.collect()
|
||||
|
||||
from secrets import WIFI_SSID, WIFI_PASSWORD, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD
|
||||
try:
|
||||
ih.network_connect_better(WIFI_SSID, WIFI_PASSWORD)
|
||||
except RuntimeError as e:
|
||||
print(e)
|
||||
with open('/sd/errors.log', 'a+') as error_log:
|
||||
error_log.write(f'{e} \n')
|
||||
#Likely a network oops, try starting again
|
||||
machine.soft_reset()
|
||||
|
||||
def get_new_image():
|
||||
sd_spi = machine.SPI(0, sck=machine.Pin(18, machine.Pin.OUT), mosi=machine.Pin(19, machine.Pin.OUT),
|
||||
miso=machine.Pin(16, machine.Pin.OUT))
|
||||
sd = sdcard.SDCard(sd_spi, machine.Pin(22))
|
||||
uos.mount(sd, "/sd")
|
||||
gc.collect()
|
||||
|
||||
ih.network_connect(WIFI_SSID, WIFI_PASSWORD)
|
||||
|
||||
try:
|
||||
nc = uNextcloud.uNextcloud("https://cloud.eyecreate.org")
|
||||
nc.set_auth(NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD)
|
||||
response = nc.get_folder_items("eink-frame")
|
||||
except OSError as e:
|
||||
print(e)
|
||||
with open('/sd/errors.log', 'a+') as error_log:
|
||||
error_log.write(f'{e} \n')
|
||||
#Likely a network oops, try starting again
|
||||
machine.soft_reset()
|
||||
|
||||
try:
|
||||
last = open('/sd/last_wallpaper', 'r')
|
||||
last_file = last.read()
|
||||
last.close()
|
||||
except OSError:
|
||||
last_file = ""
|
||||
try:
|
||||
last = open('/sd/last_wallpaper', 'r')
|
||||
last_file = last.read()
|
||||
last.close()
|
||||
except OSError:
|
||||
last_file = ""
|
||||
|
||||
if response != None:
|
||||
for item in response:
|
||||
if item.url_path == last_file:
|
||||
response.remove(item)
|
||||
ran = random.choice(response)
|
||||
if ran.mimetype == "image/jpeg":
|
||||
update_last = open('/sd/last_wallpaper', 'w+')
|
||||
if response != None:
|
||||
for item in response:
|
||||
if item.url_path == last_file:
|
||||
response.remove(item)
|
||||
ran = random.choice(response)
|
||||
if ran.mimetype == "image/jpeg":
|
||||
with open('/sd/last_wallpaper', 'w+') as update_last:
|
||||
update_last.write(ran.url_path)
|
||||
update_last.close()
|
||||
print(ran.url_path)
|
||||
gc.collect()
|
||||
nc.download_file_to_path(ran.url_path, "/sd/current_image.jpg")
|
||||
graphics = PicoGraphics(DISPLAY)
|
||||
j = JPEG(graphics)
|
||||
|
||||
j.open_file("/sd/current_image.jpg")
|
||||
j.decode()
|
||||
graphics.update()
|
||||
#sleep?
|
||||
|
||||
print(ran.url_path)
|
||||
gc.collect()
|
||||
nc.download_file_to_path(ran.url_path, "/sd/current_image.jpg")
|
||||
gc.collect()
|
||||
from picographics import PicoGraphics, DISPLAY_INKY_FRAME_7 as DISPLAY
|
||||
from jpegdec import JPEG
|
||||
graphics = PicoGraphics(DISPLAY)
|
||||
j = JPEG(graphics)
|
||||
|
||||
j.open_file("/sd/current_image.jpg")
|
||||
j.decode()
|
||||
graphics.update()
|
||||
|
||||
inky_frame.sleep_for(180)
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import ElementTree
|
||||
import urequests
|
||||
from urequests import auth
|
||||
import gc
|
||||
import gc, sys
|
||||
import ElementTree
|
||||
|
||||
|
||||
class uNextcloud:
|
||||
@@ -28,7 +27,7 @@ class uNextcloud:
|
||||
self.password = password
|
||||
|
||||
def get_folder_items(self, folder_path=""):
|
||||
response = urequests.request("PROPFIND", self.url+"/remote.php/dav/files/"+self.username+"/"+folder_path, auth=urequests.auth.HTTPBasicAuth(self.username, self.password))
|
||||
response = urequests.request("PROPFIND", self.url+"/remote.php/dav/files/"+self.username+"/"+folder_path, auth=(self.username, self.password))
|
||||
if 200 <= response.status_code < 300:
|
||||
file_list = []
|
||||
try:
|
||||
@@ -53,7 +52,7 @@ class uNextcloud:
|
||||
return None
|
||||
|
||||
def download_file_to_path(self, url_path, destination_path):
|
||||
response = urequests.request("GET", self.url+url_path, auth=urequests.auth.HTTPBasicAuth(self.username, self.password))
|
||||
response = urequests.request("GET", self.url+url_path, auth=(self.username, self.password))
|
||||
if 200 <= response.status_code < 300:
|
||||
dest = open(destination_path, 'wb')
|
||||
data = bytearray(1024)
|
||||
@@ -65,7 +64,6 @@ class uNextcloud:
|
||||
break
|
||||
len += 1024
|
||||
dest.write(data)
|
||||
print(gc.mem_free())
|
||||
dest.close()
|
||||
response.close()
|
||||
gc.collect()
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
# (c) 2016-2021 Paul Sokolovsky, MIT license, https://github.com/pfalcon/pycopy-lib
|
||||
import usocket
|
||||
|
||||
class Request:
|
||||
pass
|
||||
|
||||
|
||||
class Response:
|
||||
|
||||
def __init__(self, f):
|
||||
self.raw = f
|
||||
self.encoding = "utf-8"
|
||||
self._cached = None
|
||||
|
||||
def close(self):
|
||||
if self.raw:
|
||||
self.raw.close()
|
||||
self.raw = None
|
||||
self._cached = None
|
||||
|
||||
@property
|
||||
def content(self):
|
||||
if self._cached is None:
|
||||
try:
|
||||
self._cached = self.raw.read()
|
||||
finally:
|
||||
self.raw.close()
|
||||
self.raw = None
|
||||
return self._cached
|
||||
|
||||
@property
|
||||
def text(self):
|
||||
return str(self.content, self.encoding)
|
||||
|
||||
def json(self):
|
||||
import ujson
|
||||
return ujson.loads(self.content)
|
||||
|
||||
|
||||
def request(method, url, data=None, json=None, headers={}, auth=None, stream=None, parse_headers=True):
|
||||
redir_cnt = 1
|
||||
if json is not None:
|
||||
assert data is None
|
||||
import ujson
|
||||
data = ujson.dumps(json)
|
||||
|
||||
while True:
|
||||
try:
|
||||
proto, dummy, host, path = url.split("/", 3)
|
||||
except ValueError:
|
||||
proto, dummy, host = url.split("/", 2)
|
||||
path = ""
|
||||
if proto == "http:":
|
||||
port = 80
|
||||
elif proto == "https:":
|
||||
import ussl
|
||||
port = 443
|
||||
else:
|
||||
raise ValueError("Unsupported protocol: " + proto)
|
||||
|
||||
if ":" in host:
|
||||
host, port = host.split(":", 1)
|
||||
port = int(port)
|
||||
|
||||
if auth is not None:
|
||||
req = Request()
|
||||
req.method = method
|
||||
req.url = url
|
||||
if not headers:
|
||||
# Fresh local dict, not a copy of anything.
|
||||
headers = {}
|
||||
req.headers = headers
|
||||
req = auth(req)
|
||||
headers = req.headers
|
||||
|
||||
ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
|
||||
ai = ai[0]
|
||||
|
||||
resp_d = None
|
||||
if parse_headers is not False:
|
||||
resp_d = {}
|
||||
|
||||
s = usocket.socket(ai[0], ai[1], ai[2])
|
||||
try:
|
||||
s.connect(ai[-1])
|
||||
if proto == "https:":
|
||||
s = ussl.wrap_socket(s, server_hostname=host)
|
||||
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
|
||||
if not "Host" in headers:
|
||||
s.write(b"Host: %s\r\n" % host)
|
||||
# Iterate over keys to avoid tuple alloc
|
||||
for k in headers:
|
||||
s.write(k)
|
||||
s.write(b": ")
|
||||
s.write(headers[k])
|
||||
s.write(b"\r\n")
|
||||
if json is not None:
|
||||
s.write(b"Content-Type: application/json\r\n")
|
||||
if data:
|
||||
s.write(b"Content-Length: %d\r\n" % len(data))
|
||||
s.write(b"Connection: close\r\n\r\n")
|
||||
if data:
|
||||
s.write(data)
|
||||
|
||||
l = s.readline()
|
||||
#print(l)
|
||||
l = l.split(None, 2)
|
||||
status = int(l[1])
|
||||
reason = ""
|
||||
if len(l) > 2:
|
||||
reason = l[2].rstrip()
|
||||
while True:
|
||||
l = s.readline()
|
||||
if not l or l == b"\r\n":
|
||||
break
|
||||
#print(l)
|
||||
|
||||
if l.startswith(b"Transfer-Encoding:"):
|
||||
if b"chunked" in l:
|
||||
raise ValueError("Unsupported " + l.decode())
|
||||
elif l.startswith(b"Location:") and 300 <= status <= 399:
|
||||
if not redir_cnt:
|
||||
raise ValueError("Too many redirects")
|
||||
redir_cnt -= 1
|
||||
url = l[9:].decode().strip()
|
||||
#print("redir to:", url)
|
||||
status = 300
|
||||
break
|
||||
|
||||
if parse_headers is False:
|
||||
pass
|
||||
elif parse_headers is True:
|
||||
l = l.decode()
|
||||
k, v = l.split(":", 1)
|
||||
resp_d[k] = v.strip()
|
||||
else:
|
||||
parse_headers(l, resp_d)
|
||||
except OSError:
|
||||
s.close()
|
||||
raise
|
||||
|
||||
if status != 300:
|
||||
break
|
||||
|
||||
resp = Response(s)
|
||||
resp.status_code = status
|
||||
resp.reason = reason
|
||||
if resp_d is not None:
|
||||
resp.headers = resp_d
|
||||
return resp
|
||||
|
||||
|
||||
def head(url, **kw):
|
||||
return request("HEAD", url, **kw)
|
||||
|
||||
def get(url, **kw):
|
||||
return request("GET", url, **kw)
|
||||
|
||||
def post(url, **kw):
|
||||
return request("POST", url, **kw)
|
||||
|
||||
def put(url, **kw):
|
||||
return request("PUT", url, **kw)
|
||||
|
||||
def patch(url, **kw):
|
||||
return request("PATCH", url, **kw)
|
||||
|
||||
def delete(url, **kw):
|
||||
return request("DELETE", url, **kw)
|
||||
@@ -1,11 +0,0 @@
|
||||
# (c) 2021 Paul Sokolovsky, MIT license, https://github.com/pfalcon/pycopy-lib
|
||||
import uwwwauth
|
||||
|
||||
|
||||
class HTTPBasicAuth:
|
||||
def __init__(self, user, passwd):
|
||||
self.auth = uwwwauth.basic_resp(user, passwd)
|
||||
|
||||
def __call__(self, r):
|
||||
r.headers["Authorization"] = self.auth
|
||||
return r
|
||||
@@ -1,108 +0,0 @@
|
||||
# RFC2617, WWW-Authenticate: Basic/Digest module
|
||||
# (c) 2018 Paul Sokolovsky, MIT license
|
||||
import uhashlib
|
||||
import ubinascii
|
||||
|
||||
|
||||
# Private functions - do not use, will change
|
||||
|
||||
def md5_concat(arg1, arg2, arg3):
|
||||
h = uhashlib.md5(arg1)
|
||||
h.update(b":")
|
||||
h.update(arg2)
|
||||
if arg3 is not None:
|
||||
h.update(b":")
|
||||
h.update(arg3)
|
||||
return ubinascii.hexlify(h.digest()).decode()
|
||||
|
||||
|
||||
def make_digest_ha1(a1, method, uri, nonce):
|
||||
a2 = md5_concat(method, uri, None)
|
||||
digest = md5_concat(a1, nonce, a2)
|
||||
return digest
|
||||
|
||||
|
||||
def make_digest(realm, username, passwd, method, uri, nonce):
|
||||
a1 = md5_concat(username, realm, passwd)
|
||||
return make_digest_ha1(a1, method, uri, nonce)
|
||||
|
||||
|
||||
def parse_auth_req(line):
|
||||
typ, line = line.split(None, 1)
|
||||
d = {"type": typ}
|
||||
for kv in line.split(","):
|
||||
k, v = kv.split("=", 1)
|
||||
assert v[0] == '"' and v[-1] == '"'
|
||||
d[k.strip()] = v[1:-1]
|
||||
return d
|
||||
|
||||
|
||||
def format_resp(resp_d):
|
||||
fields = []
|
||||
for k, v in resp_d.items():
|
||||
if k in ("type", "passwd"):
|
||||
continue
|
||||
fields.append('%s="%s"' % (k, v))
|
||||
resp_auth = ", ".join(fields)
|
||||
|
||||
resp_auth = "Digest " + resp_auth
|
||||
return resp_auth
|
||||
|
||||
|
||||
def _digest_resp(auth_d, username, passwd, method, URL):
|
||||
#print(auth_d)
|
||||
|
||||
resp_d = {}
|
||||
resp_d["username"] = username
|
||||
resp_d["uri"] = URL
|
||||
resp_d["realm"] = auth_d["realm"]
|
||||
resp_d["nonce"] = auth_d["nonce"]
|
||||
|
||||
digest = make_digest(auth_d["realm"], username, passwd, method, URL, auth_d["nonce"])
|
||||
resp_d["response"] = digest
|
||||
#print(resp_d)
|
||||
|
||||
return format_resp(resp_d)
|
||||
|
||||
|
||||
# Helper functions - may change
|
||||
|
||||
def basic_resp(username, passwd):
|
||||
return "Basic " + ubinascii.b2a_base64("%s:%s" % (username, passwd))[:-1].decode()
|
||||
|
||||
|
||||
def auth_resp(auth_line, username, passwd, method=None, URL=None):
|
||||
auth_d = parse_auth_req(auth_line)
|
||||
if auth_d["type"] == "Basic":
|
||||
return basic_resp(username, passwd)
|
||||
elif auth_d["type"] == "Digest":
|
||||
assert method and URL
|
||||
return _digest_resp(auth_d, username, passwd, method, URL)
|
||||
else:
|
||||
raise ValueError(auth_d["type"])
|
||||
|
||||
|
||||
# Public interface
|
||||
|
||||
class WWWAuth:
|
||||
|
||||
def __init__(self, username, passwd):
|
||||
self.username = username
|
||||
self.passwd = passwd
|
||||
self.cached_auth_line = None
|
||||
|
||||
def resp(self, auth_line, method, URL):
|
||||
if auth_line.startswith("Basic"):
|
||||
return basic_resp(self.username, self.passwd)
|
||||
elif auth_line.startswith("Digest"):
|
||||
auth_d = parse_auth_req(auth_line)
|
||||
if auth_line != self.cached_auth_line:
|
||||
self.ha1 = md5_concat(self.username, auth_d["realm"], self.passwd)
|
||||
self.cached_auth_line = auth_line
|
||||
digest = make_digest_ha1(self.ha1, method, URL, auth_d["nonce"])
|
||||
auth_d["username"] = self.username
|
||||
auth_d["uri"] = URL
|
||||
auth_d["response"] = digest
|
||||
return format_resp(auth_d)
|
||||
else:
|
||||
raise ValueError("Unsupported auth: " + auth_line)
|
||||
Reference in New Issue
Block a user