import framebuf import struct import time import binascii class Font: def __init__(self, ws=None): self.ws = ws self.cache = {} # Simple cache for font bitmaps: {code: bytes} def set_ws(self, ws): self.ws = ws def text(self, tft, text, x, y, color, bg=0x0000): """ Draw text on ST7789 display using WebSocket to fetch fonts """ # Pre-calculate color bytes color_bytes = struct.pack(">H", color) bg_bytes = struct.pack(">H", bg) # Create LUT for current color/bg lut = [bytearray(16) for _ in range(256)] for i in range(256): for bit in range(8): # bit 7 is first pixel (leftmost) # target index: (7-bit)*2 val = (i >> bit) & 1 idx = (7 - bit) * 2 if val: lut[i][idx] = color_bytes[0] lut[i][idx+1] = color_bytes[1] else: lut[i][idx] = bg_bytes[0] lut[i][idx+1] = bg_bytes[1] initial_x = x # 1. Identify missing fonts missing_codes = set() for char in text: if ord(char) > 127: code = ord(char) if code not in self.cache: missing_codes.add(code) # 2. Batch request missing fonts if missing_codes and self.ws: # Convert to list for consistent order/string missing_list = list(missing_codes) # Limit batch size? Maybe 20 chars at a time? # For short ASR result, usually < 20 chars. req_str = ",".join([str(c) for c in missing_list]) print(f"Batch requesting fonts: {req_str}") try: self.ws.send(f"GET_FONTS_BATCH:{req_str}") self._wait_for_fonts(missing_codes) except Exception as e: print(f"Batch font request failed: {e}") # 3. Draw text for char in text: # Handle newlines if char == '\n': x = initial_x y += 16 continue # Boundary check if x + 16 > tft.width: x = initial_x y += 16 if y + 16 > tft.height: break is_chinese = False buf_data = None # Check if it's Chinese (or non-ASCII) if ord(char) > 127: code = ord(char) if code in self.cache: buf_data = self.cache[code] is_chinese = True else: # Still missing after batch request? # Could be timeout or invalid char. pass if is_chinese and buf_data: # Draw Chinese character (16x16) self._draw_bitmap(tft, buf_data, x, y, 16, 16, lut) x += 16 else: # Draw ASCII (8x16) using built-in framebuf font (8x8 actually) # If char is not ASCII, replace with '?' to avoid framebuf errors if ord(char) > 127: char = '?' self._draw_ascii(tft, char, x, y, color, bg) x += 8 def _wait_for_fonts(self, target_codes): """ Blocking wait for a set of font codes. Buffers other messages to self.ws.unread_messages. """ if not self.ws or not target_codes: return start = time.ticks_ms() self.local_deferred = [] # 2 seconds timeout for batch while time.ticks_diff(time.ticks_ms(), start) < 2000 and target_codes: # Check unread_messages first? # Actually ws.recv() in our modified client already checks unread_messages. # But wait, if we put something BACK into unread_messages, we need to be careful not to read it again immediately if we are looping? # No, we only put NON-FONT messages back. We are looking for FONT messages. # So if we pop a non-font message, we put it back? # If we put it back at head, we will read it again next loop! Infinite loop! # # Solution: We should NOT use ws.recv() which pops from unread. # We should assume unread_messages might contain what we need? # # Actually, `ws.recv()` pops from `unread_messages`. # If we get a message that is NOT what we want, we should store it in a temporary list, and push them all back at the end? # Or append to `unread_messages` (if it's a queue). # But `unread_messages` is used as a LIFO or FIFO? # pop(0) -> FIFO. # If we append, it goes to end. # So: # 1. recv() -> gets msg. # 2. Is it font? # Yes -> process. # No -> append to `temp_buffer`. # 3. After function finishes (or timeout), extend `unread_messages` with `temp_buffer`? # Wait, `unread_messages` should be preserved order. # If we had [A, B] in unread. # recv() gets A. Not font. Temp=[A]. # recv() gets B. Not font. Temp=[A, B]. # recv() gets network C (Font). Process. # End. # Restore: unread = Temp + unread? (unread is empty now). # So unread becomes [A, B]. Correct. import uselect # Fast check if we can read # But we want to block until SOMETHING arrives. # If unread_messages is not empty, we should process them first. # But we can't peak easily without modifying recv again. # Let's just use recv() and handle the buffering logic here. try: # Use a poller for the socket part to implement timeout # But recv() handles logic. # If unread_messages is empty, we poll socket. can_read = False if self.ws.unread_messages: can_read = True else: poller = uselect.poll() poller.register(self.ws.sock, uselect.POLLIN) events = poller.poll(100) # 100ms if events: can_read = True if can_read: msg = self.ws.recv() # This will pop from unread or read from sock if msg is None: # Socket closed or error? # Or just timeout in recv (but we polled). continue if isinstance(msg, str): if msg == "FONT_BATCH_END": # Batch complete. Mark remaining as failed. # We need to iterate over a copy because we are modifying target_codes? # Actually we just clear it. # But wait, target_codes is passed by reference (set). # If we clear it, loop breaks. # But we also want to mark cache as None for missing ones. temp_missing = list(target_codes) for c in temp_missing: print(f"Batch missing/failed: {c}") self.cache[c] = None # Cache failure target_codes.clear() elif msg.startswith("FONT_DATA:"): # General font data handler parts = msg.split(":") if len(parts) >= 3: try: key_str = parts[1] if key_str.startswith("0x"): c = int(key_str, 16) else: c = int(key_str) d = binascii.unhexlify(parts[2]) self.cache[c] = d if c in target_codes: target_codes.remove(c) # print(f"Batch loaded: {c}") except: pass else: # Other message, e.g. START_PLAYBACK self.local_deferred.append(msg) elif msg is not None: # Binary message? Buffer it too. self.local_deferred.append(msg) except Exception as e: print(f"Wait font error: {e}") # End of wait. Restore deferred messages. if self.local_deferred: # We want new_list = local_deferred + old_list self.ws.unread_messages = self.local_deferred + self.ws.unread_messages self.local_deferred = [] def _wait_for_font(self, target_code_str): # Compatibility wrapper or deprecated? # The new logic uses batch wait. pass def _draw_bitmap(self, tft, bitmap, x, y, w, h, lut): # Convert 1bpp bitmap to RGB565 buffer using LUT # Optimize buffer allocation # bitmap length is w * h / 8 = 32 bytes for 16x16 # Create list of chunks chunks = [lut[b] for b in bitmap] # Join chunks into one buffer rgb_buf = b''.join(chunks) tft.blit_buffer(rgb_buf, x, y, w, h) def _draw_ascii(self, tft, char, x, y, color, bg): # Use framebuf for ASCII w, h = 8, 8 buf = bytearray(w * h // 8) fb = framebuf.FrameBuffer(buf, w, h, framebuf.MONO_VLSB) fb.fill(0) fb.text(char, 0, 0, 1) # Since framebuf.text is 8x8, we center it vertically in 16px height # Drawing pixel by pixel is slow but compatible # To optimize, we can build a small buffer # Create a 8x16 RGB565 buffer rgb_buf = bytearray(8 * 16 * 2) # Fill with background bg_high, bg_low = bg >> 8, bg & 0xFF color_high, color_low = color >> 8, color & 0xFF for i in range(0, len(rgb_buf), 2): rgb_buf[i] = bg_high rgb_buf[i+1] = bg_low # Draw the 8x8 character into the buffer (centered) # MONO_VLSB: each byte is a column of 8 pixels for col in range(8): # 0..7 byte = buf[col] for row in range(8): # 0..7 if (byte >> row) & 1: # Calculate position in rgb_buf # Target: x=col, y=row+4 pos = ((row + 4) * 8 + col) * 2 rgb_buf[pos] = color_high rgb_buf[pos+1] = color_low tft.blit_buffer(rgb_buf, x, y, 8, 16)