128 lines
3.4 KiB
Python
128 lines
3.4 KiB
Python
# ============================================================+
|
|
# Identity IoT — main.py for Pico W2
|
|
# Config from config.json — OLED display integrated
|
|
# ============================================================+
|
|
|
|
import network
|
|
import utime
|
|
import ujson
|
|
from machine import Pin, I2C
|
|
from ssd1306 import SSD1306_I2C
|
|
from identity_iot import IdentityIoT
|
|
|
|
# ---- OLED ----
|
|
i2c = I2C(0, sda=Pin(4), scl=Pin(5))
|
|
oled = SSD1306_I2C(128, 64, i2c)
|
|
|
|
def show(lines):
|
|
oled.fill(0)
|
|
for i, line in enumerate(lines):
|
|
oled.text(line[:16], 0, i * 12)
|
|
oled.show()
|
|
|
|
def show_scroll(text, row=0, delay=0.04):
|
|
"""Scroll long text on a single row."""
|
|
if len(text) <= 16:
|
|
return
|
|
for i in range(len(text) - 15):
|
|
oled.fill_rect(0, row * 12, 128, 12, 0)
|
|
oled.text(text[i:i+16], 0, row * 12)
|
|
oled.show()
|
|
utime.sleep(delay)
|
|
|
|
# ---- Config ----
|
|
def load_config():
|
|
with open('config.json') as f:
|
|
return ujson.load(f)
|
|
|
|
def connect_wifi(cfg):
|
|
show(['Identity IoT', 'WIDE / AEL', '', 'Connecting...'])
|
|
wlan = network.WLAN(network.STA_IF)
|
|
wlan.active(True)
|
|
utime.sleep(2)
|
|
|
|
if not wlan.isconnected():
|
|
wlan.connect(cfg['wifi_ssid'], cfg['wifi_pass'])
|
|
timeout = 30
|
|
while timeout > 0:
|
|
if wlan.isconnected() and wlan.ifconfig()[0] != '0.0.0.0':
|
|
break
|
|
utime.sleep(1)
|
|
timeout -= 1
|
|
|
|
if not wlan.isconnected() or wlan.ifconfig()[0] == '0.0.0.0':
|
|
show(['WiFi FAILED', '', 'Check config', ''])
|
|
raise Exception('WiFi failed')
|
|
|
|
if cfg.get('wifi_static'):
|
|
wlan.ifconfig((
|
|
cfg['wifi_ip'],
|
|
cfg['wifi_mask'],
|
|
cfg['wifi_gw'],
|
|
cfg['wifi_dns'],
|
|
))
|
|
|
|
ip = wlan.ifconfig()[0]
|
|
show(['WiFi OK', ip, '', ''])
|
|
utime.sleep(1)
|
|
return wlan
|
|
|
|
# ---- Main ----
|
|
def main():
|
|
show(['Identity IoT', 'WIDE / AEL', 'v1.0', 'Boot...'])
|
|
utime.sleep(1)
|
|
|
|
cfg = load_config()
|
|
connect_wifi(cfg)
|
|
|
|
|
|
|
|
identity = IdentityIoT(
|
|
base_url = cfg['base_url'],
|
|
node_type = cfg.get('node_type', 'sensor'),
|
|
node_label = cfg.get('node_label', ''),
|
|
tenant_id = cfg['tenant_id'],
|
|
comm_key = cfg['comm_key'].encode(),
|
|
comm_iv = cfg['comm_iv'].encode(),
|
|
wifi_ssid = cfg['wifi_ssid'],
|
|
wifi_pass = cfg['wifi_pass'],
|
|
)
|
|
|
|
# Register con retry visibile
|
|
for attempt in range(3):
|
|
try:
|
|
show(['Registering...', 'attempt ' + str(attempt+1) + '/3', '', ''])
|
|
reg = identity.ensure_registered()
|
|
break
|
|
except Exception as e:
|
|
print('register failed:', e)
|
|
if attempt == 2:
|
|
show(['FAILED', 'Check network', '', ''])
|
|
raise
|
|
utime.sleep(3)
|
|
|
|
# Register
|
|
|
|
reg = identity.ensure_registered()
|
|
uid_short = reg.get('node_uid', '')[:12] + '...'
|
|
show(['Node ' + reg.get('result',''), uid_short, '', ''])
|
|
utime.sleep(1)
|
|
|
|
# Authenticate
|
|
show(['Challenge...', '', '', ''])
|
|
utime.sleep(0.5)
|
|
show(['Signing...', '', 'Please wait', ''])
|
|
token = identity.authenticate()
|
|
|
|
if token:
|
|
show(['VERIFIED', '', 'JWT issued', ':)'])
|
|
else:
|
|
show(['AUTH FAILED', '', 'Check backend', ''])
|
|
|
|
utime.sleep(2)
|
|
|
|
# Idle — show node UID
|
|
uid = identity.get_node_uid()
|
|
show(['Identity IoT', 'Active', uid[:16], uid[16:32]])
|
|
|
|
main() |