|
| 1 | +""" |
| 2 | +https://docs.micropython.org/en/latest/library/espnow.html |
| 3 | +""" |
| 4 | + |
| 5 | +from collections import deque |
| 6 | + |
| 7 | +import lvgl as lv |
| 8 | +import machine |
| 9 | +from micropython import const |
| 10 | +from mpos import Activity, MposKeyboard, TaskManager |
| 11 | +from mpos.time import localtime |
| 12 | + |
| 13 | +try: |
| 14 | + import aioespnow |
| 15 | +except ImportError: |
| 16 | + aioespnow = None |
| 17 | + |
| 18 | +try: |
| 19 | + import network |
| 20 | +except ImportError: |
| 21 | + network = None |
| 22 | + |
| 23 | +BROADCAST_MAC = const(b"\xbb\xbb\xbb\xbb\xbb\xbb") |
| 24 | + |
| 25 | + |
| 26 | +def pformat_mac(mac): |
| 27 | + if mac: |
| 28 | + return ":".join(f"{b:02x}" for b in mac) |
| 29 | + else: |
| 30 | + return "<no mac>" |
| 31 | + |
| 32 | + |
| 33 | +class EspNowChat(Activity): |
| 34 | + def onCreate(self): |
| 35 | + main_content = lv.obj() |
| 36 | + main_content.set_flex_flow(lv.FLEX_FLOW.COLUMN) |
| 37 | + main_content.set_style_pad_gap(10, 0) |
| 38 | + |
| 39 | + self.input_textarea = lv.textarea(main_content) |
| 40 | + self.input_textarea.set_placeholder_text("Message input...") |
| 41 | + self.input_textarea.set_one_line(True) |
| 42 | + self.input_textarea.set_style_text_font(lv.font_montserrat_16, lv.PART.MAIN) |
| 43 | + self.input_textarea.set_width(lv.pct(100)) |
| 44 | + self.input_textarea.add_event_cb(self.show_keyboard, lv.EVENT.CLICKED, None) |
| 45 | + |
| 46 | + self.keyboard = MposKeyboard(main_content) |
| 47 | + self.keyboard.set_textarea(self.input_textarea) |
| 48 | + self.keyboard.add_event_cb(self.keyboard_cb, lv.EVENT.READY, None) |
| 49 | + self.keyboard.add_flag(lv.obj.FLAG.HIDDEN) |
| 50 | + |
| 51 | + self.messages = lv.label(main_content) |
| 52 | + self.messages.set_style_text_font(lv.font_montserrat_14, 0) |
| 53 | + |
| 54 | + # Buffer to store and display the latest 20 messages: |
| 55 | + self.messages_buffer = deque((), 20) |
| 56 | + |
| 57 | + self.setContentView(main_content) |
| 58 | + |
| 59 | + if aioespnow and network: |
| 60 | + print("Initialize WLAN interface...") |
| 61 | + sta = network.WLAN(network.WLAN.IF_STA) |
| 62 | + sta.active(True) |
| 63 | + |
| 64 | + self.own_id = pformat_mac(machine.unique_id()) |
| 65 | + |
| 66 | + self.info("Initialize ESPNow...") |
| 67 | + self.espnow = aioespnow.AIOESPNow() |
| 68 | + self.espnow.active(True) |
| 69 | + self.espnow.add_peer(BROADCAST_MAC) |
| 70 | + |
| 71 | + if sta.isconnected(): |
| 72 | + self.info(f"Connected to WiFi: {sta.config('essid')}") |
| 73 | + self.info(f"Use WiFi Channel: {sta.config('channel')}") |
| 74 | + else: |
| 75 | + self.own_id = "<no espnow>" |
| 76 | + self.info("ESPNow not available on this platform") |
| 77 | + |
| 78 | + def info(self, text): |
| 79 | + now = localtime() |
| 80 | + hour, minute, second = now[3], now[4], now[5] |
| 81 | + message = f"{hour:02}:{minute:02}:{second:02} {text}" |
| 82 | + print(message) |
| 83 | + self.messages_buffer.appendleft(message) |
| 84 | + self.messages.set_text("\n".join(self.messages_buffer)) |
| 85 | + |
| 86 | + def keyboard_cb(self, event): |
| 87 | + message = self.input_textarea.get_text() |
| 88 | + if not message: |
| 89 | + print("Ignore empty input") |
| 90 | + else: |
| 91 | + self.input_textarea.set_text("") |
| 92 | + print(f"Create task to send {message=}...") |
| 93 | + TaskManager.create_task(self.send_messages(message)) |
| 94 | + |
| 95 | + def show_keyboard(self, event): |
| 96 | + print("Show keyboard") |
| 97 | + self.keyboard.remove_flag(lv.obj.FLAG.HIDDEN) |
| 98 | + |
| 99 | + async def send_messages(self, message): |
| 100 | + self.info(f"Sending: {message} ({self.own_id})") |
| 101 | + try: |
| 102 | + await self.espnow.asend(BROADCAST_MAC, message.encode()) |
| 103 | + except OSError as err: |
| 104 | + print(f"Error sending message: {err}") |
| 105 | + else: |
| 106 | + print(f"{message=} sent") |
| 107 | + |
| 108 | + async def receive_messages(self): |
| 109 | + await self.send_messages(f"{self.own_id} joins ESPNow chat.") |
| 110 | + async for mac, msg in self.espnow: |
| 111 | + if not msg: |
| 112 | + print("Ignore empty message from", pformat_mac(mac)) |
| 113 | + continue |
| 114 | + try: |
| 115 | + msg = msg.decode() |
| 116 | + except UnicodeError as err: |
| 117 | + msg = f"<invalid message: {err}>" |
| 118 | + self.info(f"{msg} ({pformat_mac(mac)})") |
| 119 | + raise RuntimeError("ESPNow receive loop exited, which shouldn't happen") |
| 120 | + |
| 121 | + def onResume(self, screen): |
| 122 | + super().onResume(screen) |
| 123 | + if aioespnow and network: |
| 124 | + TaskManager.create_task(self.receive_messages()) |
| 125 | + |
| 126 | + def onPause(self, screen): |
| 127 | + if aioespnow and network: |
| 128 | + self.espnow.send( |
| 129 | + BROADCAST_MAC, f"{self.own_id} leaves ESPNow chat.".encode() |
| 130 | + ) |
| 131 | + |
| 132 | + print("Stop ESPNow...") |
| 133 | + self.espnow.active(False) |
| 134 | + print("ESPNow deactivated") |
| 135 | + |
| 136 | + super().onPause(screen) |
0 commit comments