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")