298 lines
11 KiB
Python
298 lines
11 KiB
Python
import framebuf
|
|
import struct
|
|
import time
|
|
import binascii
|
|
import gc
|
|
|
|
try:
|
|
import static_font_data
|
|
except ImportError:
|
|
static_font_data = None
|
|
|
|
class Font:
|
|
def __init__(self, ws=None):
|
|
self.ws = ws
|
|
self.cache = {}
|
|
self.pending_requests = set()
|
|
self.retry_count = {}
|
|
self.max_retries = 3
|
|
# Pre-allocate buffer for row drawing (16 pixels * 2 bytes = 32 bytes)
|
|
self.row_buf = bytearray(32)
|
|
|
|
def set_ws(self, ws):
|
|
self.ws = ws
|
|
|
|
def clear_cache(self):
|
|
"""清除字体缓存以释放内存"""
|
|
self.cache.clear()
|
|
gc.collect()
|
|
|
|
def get_cache_size(self):
|
|
"""获取当前缓存的字体数量"""
|
|
return len(self.cache)
|
|
|
|
def handle_message(self, msg):
|
|
"""处理字体相关消息，更新缓存
|
|
返回: 是否为字体消息
|
|
"""
|
|
if not isinstance(msg, str):
|
|
return False
|
|
|
|
if msg.startswith("FONT_BATCH_END:"):
|
|
# 批处理结束消息，目前主要用于阻塞等待时的退出条件
|
|
return True
|
|
|
|
elif msg.startswith("FONT_DATA:"):
|
|
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
|
|
# 清除重试计数（如果有）和 pending
|
|
if c in self.retry_count:
|
|
del self.retry_count[c]
|
|
if c in self.pending_requests:
|
|
self.pending_requests.remove(c)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Font data parse error: {e}")
|
|
return True
|
|
|
|
return False
|
|
|
|
def text(self, tft, text, x, y, color, bg=0x0000, wait=True):
|
|
"""在ST7789显示器上绘制文本"""
|
|
if not text:
|
|
return
|
|
|
|
color_bytes = struct.pack(">H", color)
|
|
bg_bytes = struct.pack(">H", bg)
|
|
|
|
# Create a mini-LUT for 4-bit chunks (16 entries * 8 bytes = 128 bytes)
|
|
# Each entry maps 4 bits (0-15) to 4 pixels (8 bytes)
|
|
mini_lut = []
|
|
for i in range(16):
|
|
chunk = bytearray(8)
|
|
for bit in range(4):
|
|
# bit 0 is LSB of nibble, corresponds to rightmost pixel of the 4 pixels
|
|
# Assuming standard MSB-first bitmap
|
|
val = (i >> (3 - bit)) & 1
|
|
idx = bit * 2
|
|
if val:
|
|
chunk[idx] = color_bytes[0]
|
|
chunk[idx+1] = color_bytes[1]
|
|
else:
|
|
chunk[idx] = bg_bytes[0]
|
|
chunk[idx+1] = bg_bytes[1]
|
|
mini_lut.append(bytes(chunk))
|
|
|
|
initial_x = x
|
|
|
|
missing_codes = set()
|
|
for char in text:
|
|
if ord(char) > 127:
|
|
code = ord(char)
|
|
# Check static font data first
|
|
if static_font_data and hasattr(static_font_data, 'FONTS') and code in static_font_data.FONTS:
|
|
continue
|
|
if code not in self.cache:
|
|
missing_codes.add(code)
|
|
|
|
if missing_codes and self.ws:
|
|
missing_list = list(missing_codes)
|
|
|
|
req_str = ",".join([str(c) for c in missing_list])
|
|
# Only print if waiting, to reduce log spam in async mode
|
|
if wait:
|
|
print(f"Batch requesting fonts: {req_str}")
|
|
try:
|
|
# Add Pending requests to retry count to avoid spamming
|
|
for c in missing_list:
|
|
if c not in self.pending_requests:
|
|
self.pending_requests.add(c)
|
|
|
|
self.ws.send(f"GET_FONTS_BATCH:{req_str}")
|
|
if wait:
|
|
self._wait_for_fonts(missing_codes)
|
|
except Exception as e:
|
|
print(f"Batch font request failed: {e}")
|
|
|
|
for char in text:
|
|
if char == '\n':
|
|
x = initial_x
|
|
y += 16
|
|
continue
|
|
|
|
if x + 16 > tft.width:
|
|
x = initial_x
|
|
y += 16
|
|
if y + 16 > tft.height:
|
|
break
|
|
|
|
is_chinese = False
|
|
buf_data = None
|
|
code = ord(char)
|
|
|
|
if code > 127:
|
|
if static_font_data and hasattr(static_font_data, 'FONTS') and code in static_font_data.FONTS:
|
|
buf_data = static_font_data.FONTS[code]
|
|
is_chinese = True
|
|
elif code in self.cache:
|
|
buf_data = self.cache[code]
|
|
is_chinese = True
|
|
else:
|
|
# Missing font data
|
|
if not wait:
|
|
# In async mode, draw a placeholder or space
|
|
# We use '?' for now so user knows something is missing
|
|
char = '?'
|
|
is_chinese = False
|
|
else:
|
|
if code in self.pending_requests:
|
|
retry = self.retry_count.get(code, 0)
|
|
if retry < self.max_retries:
|
|
self.retry_count[code] = retry + 1
|
|
self._request_single_font(code)
|
|
|
|
if is_chinese and buf_data:
|
|
self._draw_bitmap_optimized(tft, buf_data, x, y, mini_lut)
|
|
x += 16
|
|
else:
|
|
if code > 127:
|
|
char = '?'
|
|
self._draw_ascii(tft, char, x, y, color, bg)
|
|
x += 8
|
|
|
|
def _draw_bitmap_optimized(self, tft, bitmap, x, y, mini_lut):
|
|
"""使用优化方式绘制位图，减少内存分配"""
|
|
# Bitmap is 32 bytes (16x16 pixels)
|
|
# 2 bytes per row
|
|
|
|
for row in range(16):
|
|
# Get 2 bytes for this row
|
|
# Handle case where bitmap might be different length (safety)
|
|
if row * 2 + 1 < len(bitmap):
|
|
b1 = bitmap[row * 2]
|
|
b2 = bitmap[row * 2 + 1]
|
|
|
|
# Process b1 (Left 8 pixels)
|
|
# High nibble
|
|
self.row_buf[0:8] = mini_lut[(b1 >> 4) & 0x0F]
|
|
# Low nibble
|
|
self.row_buf[8:16] = mini_lut[b1 & 0x0F]
|
|
|
|
# Process b2 (Right 8 pixels)
|
|
# High nibble
|
|
self.row_buf[16:24] = mini_lut[(b2 >> 4) & 0x0F]
|
|
# Low nibble
|
|
self.row_buf[24:32] = mini_lut[b2 & 0x0F]
|
|
|
|
tft.blit_buffer(self.row_buf, x, y + row, 16, 1)
|
|
|
|
def _request_single_font(self, code):
|
|
"""请求单个字体"""
|
|
if self.ws:
|
|
try:
|
|
self.ws.send(f"GET_FONT_UNICODE:{code}")
|
|
except:
|
|
pass
|
|
|
|
def _wait_for_fonts(self, target_codes):
|
|
"""等待字体数据返回"""
|
|
if not self.ws or not target_codes:
|
|
return
|
|
|
|
start = time.ticks_ms()
|
|
self.local_deferred = []
|
|
|
|
while time.ticks_diff(time.ticks_ms(), start) < 3000 and target_codes:
|
|
try:
|
|
can_read = False
|
|
if hasattr(self.ws, 'unread_messages') and self.ws.unread_messages:
|
|
can_read = True
|
|
else:
|
|
import uselect
|
|
poller = uselect.poll()
|
|
poller.register(self.ws.sock, uselect.POLLIN)
|
|
events = poller.poll(100)
|
|
if events:
|
|
can_read = True
|
|
|
|
if can_read:
|
|
msg = self.ws.recv()
|
|
if msg is None:
|
|
continue
|
|
|
|
if self.handle_message(msg):
|
|
# 如果是批处理结束，检查是否有失败的
|
|
if msg.startswith("FONT_BATCH_END:"):
|
|
parts = msg[15:].split(":")
|
|
failed = int(parts[1]) if len(parts) > 1 else 0
|
|
|
|
if failed > 0:
|
|
temp_missing = list(target_codes)
|
|
for c in temp_missing:
|
|
if c not in self.cache:
|
|
print(f"Font failed after retries: {c}")
|
|
self.cache[c] = None # 标记为 None 避免死循环
|
|
if c in target_codes:
|
|
target_codes.remove(c)
|
|
|
|
# 清除所有剩余的目标，因为批处理结束了
|
|
# 但实际上可能只需要清除 failed 的。
|
|
# 无论如何，收到 BATCH_END 意味着本次请求处理完毕。
|
|
# 如果还有没收到的，可能是丢包了。
|
|
# 为了简单起见，我们认为结束了。
|
|
target_codes.clear()
|
|
|
|
# 检查是否有新缓存的字体满足了 target_codes
|
|
temp_target = list(target_codes)
|
|
for c in temp_target:
|
|
if c in self.cache:
|
|
target_codes.remove(c)
|
|
if c in self.retry_count:
|
|
del self.retry_count[c]
|
|
|
|
else:
|
|
self.local_deferred.append(msg)
|
|
|
|
except Exception as e:
|
|
print(f"Wait font error: {e}")
|
|
|
|
if self.local_deferred:
|
|
if hasattr(self.ws, 'unread_messages'):
|
|
self.ws.unread_messages = self.local_deferred + self.ws.unread_messages
|
|
self.local_deferred = []
|
|
|
|
def _draw_ascii(self, tft, char, x, y, color, bg):
|
|
"""绘制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)
|
|
|
|
rgb_buf = bytearray(8 * 16 * 2)
|
|
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
|
|
|
|
for col in range(8):
|
|
byte = buf[col]
|
|
for row in range(8):
|
|
if (byte >> row) & 1:
|
|
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)
|