UIND
/
UiMetrix-M5-Firmware
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
UiMetrix-M5-Firmware/main.py

675 lines
22 KiB
Python
Raw Permalink Blame History

import os
import gc
import M5
from M5 import *
import network
import m5ui
from unit import RFIDUnit, RGBUnit
from hardware import I2C, Pin
import time
import utime
from umqtt.simple import MQTTClient
from esp32 import NVS
import ujson
from machine import unique_id
import ubinascii
import requests
import machine
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
WIFI_SSID = 'Utopia-WiFi'
WIFI_PASS = '@!nt3lGwn#1'
#WIFI_SSID = 'UI-Matrix'
#WIFI_PASS = 'Uimatrix01'
MQTT_BROKER = '192.168.2.174'
#MQTT_BROKER = 'scada.utopia.pk'
MQTT_PORT = 1883
MQTT_USER = 'utopia'
MQTT_PASSWORD = 'utopia01'
MQTT_SCAN_TOPIC = "send_receive_data" # published: one message per card tap
MQTT_RESPONSE_TOPIC = "response_data" # subscribed: the server's answer to a scan
MQTT_LAMP_TOPIC = "lamp_topic" # subscribed: light bar colour
MQTT_RESET_TOPIC = "reset-topic" # subscribed: payload "reset" clears the counter
DEVICE_TYPE = "tag_reader"
UPDATE_VERSION_URL = "https://git.utopiadeals.com/UIND/UiMetrix-M5-Firmware/raw/branch/main/version.txt"
UPDATE_FILE_URL = "https://git.utopiadeals.com/UIND/UiMetrix-M5-Firmware/raw/branch/main/main.py"
OTA_MAIN_PATH = "/flash/main.py" # the program the device runs at boot
OTA_TEMP_PATH = "/flash/main_ota_temp.py" # a download lands here first
OTA_BACKUP_PATH = "/flash/main_prev.py" # the previous program, kept for manual rollback
NVS_NAMESPACE = "rfid_data"
MAX_RETRIES = 5
ACK_TIMEOUT_MS = 10000 # how long a tap blocks the reader while the server answers
MQTT_RECONNECT_INTERVAL = 5000
WIFI_TIMEOUT_S = 10
# Card layout (MIFARE Classic blocks; 7 and 11 are sector trailers). Blocks 4-9 once held
# sku / colour / size / article / remarks, which are no longer published, so only these two
# are read now.
CARD_TYPE_BLOCK = 10 # "Product", "Operator" or "Reset"
OPERATOR_ID_BLOCK = 12 # employee id, on Operator cards
LAMP_COLOURS = {
"red": 0xff0000,
"green": 0x008000,
"yellow": 0xffff00,
"purple": 0x800080,
"blue": 0x0000FF,
}
DEFAULT_LAMP_COLOUR = 0x008000
DISCONNECTED_COLOUR = 0x0000FF # blue on the bar means "not connected"
IMG = "/flash/res/img/"
# ---------------------------------------------------------------------------
# Identity and firmware version
# ---------------------------------------------------------------------------
serial = ubinascii.hexlify(unique_id()).decode()
_version_nvs = NVS("storage")
def get_local_version():
try:
return _version_nvs.get_str("fw_version")
except Exception:
return "1.0" # default first install
def set_local_version(version):
try:
_version_nvs.set_str("fw_version", version)
_version_nvs.commit()
print("Version saved:", version)
except Exception as e:
print("NVS save error:", e)
local_version = get_local_version()
# ---------------------------------------------------------------------------
# Runtime state (module-level; functions declare `global` for what they change)
# ---------------------------------------------------------------------------
ui = {} # screen widgets by name; label_set() tolerates ones not created yet
wlan = None
i2c0 = None
rfid_0 = None
rgb_0 = None
battery_bar = None
mqtt_client = None
mqtt_last_try = 0
wifi_retry_count = 0
Operator_ID = "-"
lamp_color = DEFAULT_LAMP_COLOUR
tag_counter = 0 # the server's count: only a positive ACK increments it
last_uid_str = "-"
tap_handled = False # set once the card on the reader has been acted on, cleared when it leaves
rfid_needs_init = False
waiting_for_ack = False
ack_deadline = 0
pending_scan_type = None # card type of the scan the server has not answered yet
# ---------------------------------------------------------------------------
# OTA update (runs once at boot, when WiFi is up)
# ---------------------------------------------------------------------------
def _is_valid_program(code):
"""Reject error pages, truncated downloads and anything that does not compile."""
if not code or len(code) < 1000:
return False
if "def setup" not in code or "def loop" not in code:
return False
try:
compile(code, "main.py", "exec")
except SyntaxError:
return False
except MemoryError:
print("Not enough memory to verify the update")
return False
return True
def install_update(new_code, remote_version):
"""Write the new program to flash, swap it in as main.py, then record the version."""
with open(OTA_TEMP_PATH, "w") as f:
f.write(new_code)
# Keep the current program as a rollback copy (restore it over USB if the new one misbehaves).
try:
os.remove(OTA_BACKUP_PATH)
except OSError:
pass
try:
os.rename(OTA_MAIN_PATH, OTA_BACKUP_PATH)
except OSError:
pass # no main.py yet
try:
os.rename(OTA_TEMP_PATH, OTA_MAIN_PATH)
except OSError:
# Put the old program back so the device still boots. The old version stays
# in NVS, so the update is retried on the next boot.
try:
os.rename(OTA_BACKUP_PATH, OTA_MAIN_PATH)
except OSError:
pass
raise
set_local_version(remote_version)
def check_for_update():
try:
print("Checking for update...")
r = requests.get(UPDATE_VERSION_URL)
try:
status = getattr(r, "status_code", 200)
if status != 200:
print("Version check failed: HTTP", status)
return
remote_version = r.text.strip()
finally:
r.close()
print("Remote:", remote_version)
print("Local :", local_version)
# Compared as floats, so "1.10" is OLDER than "1.9".
if float(remote_version) <= float(local_version):
print("Already latest version")
return
print("New version found. Downloading...")
gc.collect()
r = requests.get(UPDATE_FILE_URL)
try:
status = getattr(r, "status_code", 200)
if status != 200:
print("Download failed: HTTP", status)
return
new_code = r.text
finally:
r.close()
if not _is_valid_program(new_code):
print("Update rejected: downloaded file is not a valid program")
return
install_update(new_code, remote_version)
print("Update installed. Restarting...")
machine.reset()
except Exception as e:
print("Update check failed:", e)
# ---------------------------------------------------------------------------
# Persistence (NVS namespace rfid_data: count, color, operator_id)
# ---------------------------------------------------------------------------
def load_from_nvs():
global tag_counter, Operator_ID, lamp_color
try:
nvs = NVS(NVS_NAMESPACE)
try:
tag_counter = nvs.get_i32("count")
except Exception:
tag_counter = 0
nvs.set_i32("count", tag_counter)
nvs.commit()
try:
lamp_color = nvs.get_i32("color")
except Exception:
lamp_color = DEFAULT_LAMP_COLOUR
nvs.set_i32("color", lamp_color)
nvs.commit()
try:
Operator_ID = nvs.get_str("operator_id")
if Operator_ID == "" or Operator_ID == "NA":
Operator_ID = "-"
except Exception:
Operator_ID = "-"
nvs.set_str("operator_id", Operator_ID)
nvs.commit()
print("Loaded from NVS: count", tag_counter, "colour", lamp_color, "operator", Operator_ID)
except Exception as e:
print("NVS Load Error:", e)
tag_counter = 0
lamp_color = DEFAULT_LAMP_COLOUR
Operator_ID = "-"
def save_to_nvs():
try:
nvs = NVS(NVS_NAMESPACE)
nvs.set_i32("count", tag_counter)
nvs.set_i32("color", lamp_color)
nvs.set_str("operator_id", Operator_ID)
nvs.commit()
print("Saved to NVS")
except Exception as e:
print("NVS Save Error:", e)
# ---------------------------------------------------------------------------
# Screen
# ---------------------------------------------------------------------------
def label_set(name, text):
label = ui.get(name)
if label is not None:
label.setText(str(text))
def set_msg(text):
label_set('msg', "MSG : " + str(text))
def show_wifi_icon(ok):
# Drawn over the previous icon at fixed coordinates; the two files differ slightly in size.
if ok:
ui['wifi_icon'] = Widgets.Image(IMG + "wifi.jpg", 149, 3, scale_x=0.9, scale_y=1)
else:
ui['wifi_icon'] = Widgets.Image(IMG + "wifi_error.jpg", 150, 3, scale_x=1.1, scale_y=1)
def show_mqtt_icon(ok):
if ok:
ui['mqtt_icon'] = Widgets.Image(IMG + "MQTT_Cloud.jpg", 100, 0)
else:
ui['mqtt_icon'] = Widgets.Image(IMG + "mqtt_error.jpg", 104, 1, scale_x=1, scale_y=1)
def show_battery():
level = Power.getBatteryLevel()
battery_bar.set_value(level, True)
label_set('battery', level)
icon = "charging.jpg" if Power.isCharging() else "not_charging.jpg"
ui['charging_icon'] = Widgets.Image(IMG + icon, 305, 11)
def init_ui():
try:
Widgets.fillScreen(0xffffff)
# Operator
ui['operator_box'] = Widgets.Rectangle(4, 101, 152, 55, 0x616161, 0xffffff)
ui['operator_img'] = Widgets.Image(IMG + "emp.jpg", 5, 102, scale_x=1, scale_y=1)
ui['operator'] = Widgets.Label(str(Operator_ID), 60, 124, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu18)
# Production count
ui['count_box'] = Widgets.Rectangle(160, 101, 155, 55, 0x616161, 0xffffff)
ui['count_img'] = Widgets.Image(IMG + "mach.jpg", 165, 102, scale_x=1, scale_y=1)
ui['count'] = Widgets.Label(str(tag_counter), 220, 124, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu18)
# Server message bar
ui['msg_box'] = Widgets.Rectangle(4, 160, 311, 25, 0x616161, 0xffffff)
ui['msg'] = Widgets.Label("MSG : ", 8, 166, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu12)
# Tag UID
ui['uid_box'] = Widgets.Rectangle(4, 189, 311, 47, 0x616161, 0xffffff)
ui['uid_img'] = Widgets.Image(IMG + "rfid-tag-log.jpg", 5, 190, scale_x=1, scale_y=0.9)
ui['uid'] = Widgets.Label("-", 60, 205, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu12)
# Device serial
ui['serial_box'] = Widgets.Rectangle(4, 47, 311, 50, 0x616161, 0xffffff)
ui['serial_img'] = Widgets.Image(IMG + "bar_c.jpg", 5, 58, scale_x=1.2, scale_y=1.2)
ui['serial'] = Widgets.Label("SR: " + serial, 70, 70, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu18)
# Status strip
show_wifi_icon(True)
show_mqtt_icon(False)
ui['ip'] = Widgets.Label("IP: Connecting...", 205, 35, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu9)
ui['logo'] = Widgets.Image(IMG + "Logo.jpg", 0, 0)
ui['battery'] = Widgets.Label("b%", 230, 15, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu12)
ui['version'] = Widgets.Label("V" + str(local_version), 275, 55, 1.0, 0x000000, 0xffffff, Widgets.FONTS.DejaVu12)
except Exception as e:
print("UI Init Error:", e)
# ---------------------------------------------------------------------------
# WiFi
# ---------------------------------------------------------------------------
def init_wifi():
global wlan
try:
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.config(reconnects=MAX_RETRIES)
connect_wifi()
except Exception as e:
print("WiFi Init Error:", e)
label_set('ip', "WiFi Error")
show_wifi_icon(False)
rgb_0.fill_color(DISCONNECTED_COLOUR)
def connect_wifi():
"""Blocking connect. Retries after a 2 s pause, MAX_RETRIES times over the device's life."""
global wifi_retry_count
if wlan.isconnected():
return
try:
label_set('ip', "WiFi: Connecting...")
show_wifi_icon(False)
rgb_0.fill_color(DISCONNECTED_COLOUR)
wlan.connect(WIFI_SSID, WIFI_PASS)
timeout = WIFI_TIMEOUT_S
while not wlan.isconnected() and timeout > 0:
time.sleep_ms(500)
timeout -= 0.5
M5.update()
if wlan.isconnected():
label_set('ip', "IP:" + wlan.ifconfig()[0])
show_wifi_icon(True)
rgb_0.fill_color(lamp_color)
return
label_set('ip', "WiFi: Timeout")
except Exception as e:
print("WiFi Error:", e)
label_set('ip', "WiFi: Error")
show_wifi_icon(False)
rgb_0.fill_color(DISCONNECTED_COLOUR)
wifi_retry_count += 1
if wifi_retry_count < MAX_RETRIES:
time.sleep(2)
connect_wifi()
# ---------------------------------------------------------------------------
# MQTT
# ---------------------------------------------------------------------------
def publish_scan(uid, cardtype, operator_id):
"""One message per card tap; the server routes on card_type / device_type."""
if not mqtt_client:
return
message = {
"card_type": cardtype,
"operator_id": operator_id,
"device_type": DEVICE_TYPE,
"uid": uid,
"device_serial": serial,
}
try:
mqtt_client.publish(MQTT_SCAN_TOPIC, ujson.dumps(message), qos=0)
except Exception as e:
print("MQTT Publish Error:", e)
reconnect_mqtt()
def start_ack_wait(cardtype):
"""Block the reader until the server answers this tap (or ACK_TIMEOUT_MS passes)."""
global waiting_for_ack, ack_deadline, pending_scan_type
set_msg("WAITING FOR RESPONSE")
waiting_for_ack = True
pending_scan_type = cardtype
ack_deadline = utime.ticks_add(utime.ticks_ms(), ACK_TIMEOUT_MS)
def reset_counter():
global tag_counter, last_uid_str
tag_counter = 0
last_uid_str = "-"
label_set('count', "0")
label_set('uid', "-")
set_msg("")
def handle_response(msg):
global waiting_for_ack, tag_counter
data = ujson.loads(msg)
if data.get("device_serial") != serial:
print("Message ignored, not for this device")
return
ack = data.get("ack")
output = data.get("output")
if ack == 1:
waiting_for_ack = False
# Only a Product scan is production: an accepted Operator or Reset scan is not a piece.
if pending_scan_type == "Product":
tag_counter += 1
label_set('count', tag_counter)
save_to_nvs()
set_msg(output)
elif ack == 0:
waiting_for_ack = False
set_msg(output)
else:
print("Message ignored, no ack field")
return
print(output)
def handle_lamp(msg):
global waiting_for_ack, lamp_color
data = ujson.loads(msg)
if data.get("serial") != serial:
print("Message ignored, serial does not match")
return
waiting_for_ack = False # a lamp message for this device also releases a pending tap
colour = LAMP_COLOURS.get(data.get("color"))
if colour is None:
return
print("Set light", data.get("color"))
rgb_0.fill_color(colour)
lamp_color = colour
save_to_nvs()
def mqtt_callback(*args):
"""Accepts both the 2-argument and 4-argument umqtt callback signatures."""
if len(args) == 2:
topic, msg = args
elif len(args) == 4:
topic, msg, _, _ = args
else:
print("Unsupported callback format")
return
try:
topic = topic.decode('utf-8') if isinstance(topic, bytes) else topic
msg = msg.decode('utf-8') if isinstance(msg, bytes) else msg
print(f"MQTT: {topic} -> {msg}")
if topic.endswith(MQTT_RESET_TOPIC) and msg.lower() == 'reset':
reset_counter() # keeps the operator
save_to_nvs()
print("System reset via MQTT")
elif topic.endswith(MQTT_RESPONSE_TOPIC):
handle_response(msg)
elif topic.endswith(MQTT_LAMP_TOPIC):
handle_lamp(msg)
except Exception as e:
print("MQTT Callback Error:", e)
def init_mqtt():
global mqtt_client
if not wlan.isconnected():
print("WiFi not connected")
return
try:
client = MQTTClient(
client_id=serial,
server=MQTT_BROKER,
port=MQTT_PORT,
user=MQTT_USER,
password=MQTT_PASSWORD,
keepalive=60,
ssl=False
)
client.set_callback(mqtt_callback)
client.connect()
for topic in (MQTT_RESET_TOPIC, MQTT_RESPONSE_TOPIC, MQTT_LAMP_TOPIC):
client.subscribe(topic)
mqtt_client = client
print("MQTT Connected")
show_mqtt_icon(True)
rgb_0.fill_color(lamp_color)
except Exception as e:
print("MQTT init error:", e)
mqtt_client = None
show_mqtt_icon(False)
def reconnect_mqtt():
"""Tear the client down and rebuild it, at most once per MQTT_RECONNECT_INTERVAL."""
global mqtt_client, mqtt_last_try
now = utime.ticks_ms()
if utime.ticks_diff(now, mqtt_last_try) < MQTT_RECONNECT_INTERVAL:
return
mqtt_last_try = now
print("MQTT reconnect attempt")
try:
if mqtt_client:
try:
mqtt_client.disconnect()
except Exception:
pass
mqtt_client = None
if not wlan.isconnected():
print("WiFi not connected, skipping MQTT reconnect")
show_wifi_icon(False)
return
init_mqtt()
except Exception as e:
print("MQTT reconnect error:", e)
show_mqtt_icon(False)
mqtt_client = None
# ---------------------------------------------------------------------------
# RFID
# ---------------------------------------------------------------------------
def init_rfid():
global i2c0, rfid_0, rfid_needs_init
i2c0 = I2C(0, scl=Pin(1), sda=Pin(2), freq=100000)
rfid_0 = RFIDUnit(i2c0)
rfid_needs_init = False
def read_card_fields():
"""(card type, operator id) as written on the card; None for a block that cannot be read."""
values = []
for block in (CARD_TYPE_BLOCK, OPERATOR_ID_BLOCK):
try:
values.append(rfid_0.read(block).decode('utf-8').strip('\x00'))
except Exception:
values.append(None)
return values[0], values[1]
def handle_card():
"""Act on the card on the reader: once per tap, by the type written on the card."""
global Operator_ID, last_uid_str, tap_handled
uid = rfid_0.read_card_uid()
if not uid:
return
uid_str = ':'.join('%02X' % b for b in uid)
label_set('uid', uid_str)
cardtype, card_operator_id = read_card_fields()
rfid_0.close()
if tap_handled:
return # the same card is still resting on the reader
if cardtype == "Reset":
tap_handled = True
reset_counter()
Operator_ID = "0"
label_set('operator', "-")
save_to_nvs()
publish_scan("RESET", "RESET", 0)
start_ack_wait("Reset")
return
last_uid_str = uid_str
if cardtype == "Operator" and card_operator_id and card_operator_id != '0':
tap_handled = True
Operator_ID = card_operator_id
label_set('operator', Operator_ID)
save_to_nvs()
# The operator card's OWN uid. Until 2026-09-15 this sent the previous card's uid
# ("-" after a reboot), so the server checked and logged the wrong card.
publish_scan(uid_str, cardtype, Operator_ID)
start_ack_wait(cardtype)
return
if cardtype == "Product":
tap_handled = True
publish_scan(uid_str, cardtype, Operator_ID)
start_ack_wait(cardtype)
# Any other type (or an unwritten card) is shown on screen and not sent.
# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------
def loop():
global mqtt_client, tap_handled, rfid_needs_init, waiting_for_ack
if wlan.isconnected():
label_set('ip', "IP:" + wlan.ifconfig()[0])
time.sleep(0.1)
show_battery()
if not wlan.isconnected():
connect_wifi()
if wlan.isconnected():
init_mqtt()
else:
time.sleep(1)
return
try:
if mqtt_client:
mqtt_client.check_msg()
except Exception as e:
print("MQTT error:", e)
mqtt_client = None
if waiting_for_ack and utime.ticks_diff(utime.ticks_ms(), ack_deadline) >= 0:
print("ACK TIMEOUT: No response received")
set_msg("NO RESPONSE RECEIVED")
waiting_for_ack = False
if wlan.isconnected() and mqtt_client is None:
reconnect_mqtt()
try:
if rfid_needs_init:
init_rfid()
print("RFID RE_INITIALIZED")
card_present = rfid_0.is_new_card_present() if rfid_0 else False
if waiting_for_ack:
# Further cards are ignored until the server answers or the deadline passes.
if card_present:
set_msg("WAITING FOR RESPONSE")
return
if card_present:
handle_card()
else:
label_set('uid', "-")
tap_handled = False
except Exception as e:
print("RFID Processing Error:", e)
label_set('uid', "RFID Error")
rfid_needs_init = True # rebuild the bus and reader on the next pass
# Liveness heartbeat, several times a second while idle.
if mqtt_client:
try:
mqtt_client.publish("M5devices/" + serial, "{}", qos=0)
except Exception as e:
print("MQTT heartbeat error:", e)
mqtt_client = None
M5.update()
time.sleep(0.1)
def setup():
global rgb_0, battery_bar, rfid_needs_init
M5.begin()
m5ui.init()
battery_bar = m5ui.M5Bar(x=256, y=7, w=49, h=25, min_value=0, max_value=100, value=25, bg_c=0x616161, color=0x21f398)
rgb_0 = RGBUnit((8, 9), 10)
load_from_nvs()
rgb_0.fill_color(DISCONNECTED_COLOUR)
init_ui()
init_wifi()
if wlan.isconnected():
check_for_update()
try:
init_rfid()
except Exception as e:
print("RFID Init Error:", e)
label_set('uid', "RFID Error")
rfid_needs_init = True
init_mqtt()
if __name__ == '__main__':
try:
setup()
while True:
loop()
except Exception as e:
print("Main Error:", e)
try:
from utility import print_error_msg
print_error_msg(e)
except ImportError:
print("Please update firmware")
Reference in New Issue View Git Blame Copy Permalink