from mpp.mp4_format import *
from mpp.mp4_format_struct import *
from media.vencoder import *
from media.sensor import *
from media.media import *
from media.display import *
from media.uvc import *
import uctypes
import time
from machine import TOUCH, PWM, FPIOA, Pin, ADC
import image
from libs.YOLO import YOLOv5
from libs.Utils import *
import os, sys, gc
import ulab.numpy as np
import _thread
import socket
import network
import errno
import cv_lite
import cv2
from media.media import *
from mpp.mp4_format import *
from mpp.mp4_format_struct import *
from media.pyaudio import *
import media.g711 as g711
from mpp.payload_struct import *
import media.vdecoder as vdecoder
from media.display import *
import uctypes
import time
import _thread
import os
from libs.Utils import hwc2chw
from libs.PipeLine import PipeLine
from libs.AIBase import AIBase
from libs.AI2D import Ai2d
from libs.Utils import *
import os,sys,ujson,gc,math
from media.media import *
import nncase_runtime as nn
import ulab.numpy as np
import image
import aicube
import os
from media.mjpeg import MJPEGEncoder
import gc
import errno
# ===================== UVC模拟多路快照全局缓存 =====================
uvc_raw_frame = None
uvc_frame_lock = _thread.allocate_lock()
def ALIGN_UP(x, align):
return ((x + align - 1) // align) * align
def _uvc_fetch_one_frame():
"""真正从UVC取一帧，返回RGB888 image.Image，外部加锁调用"""
global uvc_raw_frame
img = UVC.snapshot()
if img is None:
return None
# UVC输出MJPEG，转RGB888
rgb888_img = img.to_rgb888()
del img
return rgb888_img
def uvc_snapshot_sim(chn=0):
"""
模拟原sensor.snapshot(chn=ID)
实际UVC只有一路；取1帧，复制返回；chn参数仅做兼容，不做硬件通道区分
CAM_CHN_ID_0 / 1 /2 全部走这里
"""
global uvc_raw_frame
uvc_frame_lock.acquire()
try:
# 获取新帧
new_frame = _uvc_fetch_one_frame()
if new_frame is not None:
if uvc_raw_frame is not None:
del uvc_raw_frame
uvc_raw_frame = new_frame
if uvc_raw_frame is None:
return None
# copy一份返回，模拟不同通道取出独立image对象
out = uvc_raw_frame.copy()
return out
finally:
uvc_frame_lock.release()
def socket_send_all(sock, data, timeout_ms=5000):
"""MicroPython兼容sendall，处理EAGAIN/EWOULDBLOCK，超时抛出异常"""
total = len(data)
sent = 0
start = time.ticks_ms()
while sent < total:
os.exitpoint()
elapsed = time.ticks_diff(time.ticks_ms(), start)
if elapsed > timeout_ms:
raise OSError("socket_send_all timeout")
try:
n = sock.send(data[sent:])
sent += n
except OSError as e:
if e.args[0] in (errno.EAGAIN):
# 缓冲区满，短暂让出CPU，再重试
time.sleep_ms(2)
continue
else:
raise e
return sent
def blend_argbg_to_rgb(rgb_src_img, argb_osd_img):
"""
rgb_src_img: image.Image RGB888
argb_osd_img: image.Image ARGB8888
return: image.Image 融合后的RGB888图像
"""
out_img = rgb_src_img.copy()
# ARGB图层叠加到RGB888画布
out_img.draw_image(argb_osd_img, 0, 0)
# 【保险】强制转换格式，重置元数据，规避draw_image跨格式叠加元数据损坏bug
fixed_img = out_img.to_rgb888()
del out_img
return fixed_img
def image_to_jpeg_bytes(img_obj, quality=60):
"""
纯CPU cv2编码；支持ARGB8888 / RGB888
img_obj: image.Image
return: buf对象，拥有tobytes()
"""
try:
img_np = img_obj.to_numpy_ref()
fmt = img_obj.format()
if fmt == image.ARGB8888:
A,R,G,B = cv2.split(img_np)
bgr_np = cv2.merge([B, G, R])
elif fmt == image.RGB888:
R,G,B = cv2.split(img_np)
bgr_np = cv2.merge([B, G, R])
else:
print(f"[image_to_jpeg_bytes]不支持图像格式:{fmt}")
return None
success, buf = cv2.imencode(".jpg", bgr_np, [95, 60])
if success:
return buf
else:
print("[image_to_jpeg_bytes]cv2 imencode返回失败")
return None
except Exception as e:
print(f"[image_to_jpeg_bytes]cv2编码异常:{e}")
import sys
sys.print_exception(e)
return None
def send_image_over_tcp():
global client_conn, yolo_running, yolores, ocr_boxes, ocr_texts
tcp_lock.acquire()
conn = client_conn
tcp_lock.release()
if conn is None:
print("[send_image_over_tcp] 无客户端连接，跳过发送图片")
return False
photo_lock.acquire()
tmp_taking = taking_photo
photo_lock.release()
if tmp_taking:
print("[send_image_over_tcp] 正在拍照，跳过发送")
return False
cam_img = None
cam_copy = None
temp_osd = None
try:
sensor_lock.acquire()
cam_img = uvc_snapshot_sim(chn=CAM_CHN_ID_2)
if cam_img is None:
sensor_lock.release()
return False
cam_copy = cam_img.copy()
sensor_lock.release()
temp_osd = image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.ARGB8888)
temp_osd.clear()
yolo_lock.acquire()
running = yolo_running
local_yolo_res = yolores
yolo_lock.release()
ocr_lock.acquire()
local_ocr_box = ocr_boxes
local_ocr_txt = ocr_texts
ocr_lock.release()
if running and local_yolo_res:
yolo.draw_result(local_yolo_res, temp_osd)
if running and ocr_det_rec is not None:
ocr_det_rec.draw_result_to_img(temp_osd, local_ocr_box, local_ocr_txt)
FONT = 24
rec_color = (255,0,0,255) if recording else (0,255,0,255)
temp_osd.draw_string_advanced(10,10, FONT, f"录制:{'中' if recording else '空闲'}", rec_color)
tcp_lock.acquire()
tcp_str = "客户端:已连接" if client_conn else "客户端:未连接"
tcp_lock.release()
temp_osd.draw_string_advanced(10,40, FONT, tcp_str, (0,0,255,255))
temp_osd.draw_string_advanced(10,70, FONT, f"YOLO+OCR:{'运行中' if running else '已停止'}", (0,255,120,255))
temp_osd.draw_string_advanced(10,100, FONT, f"ADC:{adcdianya}", (0,255,0,255))
temp_osd.draw_string_advanced(10,130,FONT, djshow_status, djstatus_color + (255,))
ip_str = f"IP:{wifi_ip}" if wifi_ip else "IP:未连接"
temp_osd.draw_string_advanced(600,10, FONT, ip_str, (255,255,0,255))
yolo_btn_text = "停止检测" if running else "开始检测"
buttons = [
[0, 400, 200, 80, "结束录制" if recording else "开始录制", "rec"],
[400, 400, 200, 80, "图库", "tuku"],
[600, 400, 200, 80, yolo_btn_text, "yolo_btn"],
[200, 400, 200, 80, "拍照", "pic"],
]
for x,y,w,h,t,bid in buttons:
temp_osd.draw_rectangle(x,y,w,h,(220,220,220,255),fill=True)
temp_osd.draw_rectangle(x,y,w,h,(180,180,180,255),2)
temp_osd.draw_string_advanced(x+40,y+30,18,t,(0,0,0,255))
blend_img = blend_argbg_to_rgb(cam_copy, temp_osd)
blend_np = blend_img.to_numpy_ref().copy()
blend_bgr = cv2.cvtColor(blend_np, cv2.COLOR_RGB2BGR)
success, buf = cv2.imencode(".jpg", blend_bgr, [95, 60])
if not success:
print("[send_image_over_tcp] cv2 imencode返回失败")
return False
jpeg_data = buf
img_len = len(jpeg_data)
header = f"IMG_START:{img_len}\n".encode("utf-8")
tail = b"\nIMG_END\n"
tcp_lock.acquire()
try:
socket_send_all(conn, header)
socket_send_all(conn, jpeg_data)
socket_send_all(conn, tail)
except OSError as e:
print(f"[send_image_over_tcp]发送异常 E={e}")
cid = id(conn)
map_lock.acquire()
if cid in conn_valid_map:
conn_valid_map[cid] = False
map_lock.release()
tcp_lock.release()
return False
finally:
tcp_lock.release()
print(f"[send_image_over_tcp] 发送融合画面成功，字节数:{img_len}")
return True
except Exception as e:
print(f"[send_image_over_tcp] 异常: {e}")
import sys
sys.print_exception(e)
return False
finally:
if cam_img:
del cam_img
if cam_copy:
del cam_copy
if temp_osd:
del temp_osd
gc.collect()
# 自定义OCR检测类
class OCRDetectionApp(AIBase):
def __init__(self,kmodel_path,model_input_size,mask_threshold=0.5,box_threshold=0.2,rgb888p_size=[224,224],display_size=[1920,1080],debug_mode=0):
super().__init__(kmodel_path,model_input_size,rgb888p_size,debug_mode)
self.kmodel_path=kmodel_path
self.model_input_size=model_input_size
self.mask_threshold=mask_threshold
self.box_threshold=box_threshold
self.rgb888p_size=[ALIGN_UP(rgb888p_size[0],16),rgb888p_size[1]]
self.display_size=[ALIGN_UP(display_size[0],16),display_size[1]]
self.debug_mode=debug_mode
self.ai2d=Ai2d(debug_mode)
self.ai2d.set_ai2d_dtype(nn.ai2d_format.NCHW_FMT,nn.ai2d_format.NCHW_FMT,np.uint8, np.uint8)
def config_preprocess(self,input_image_size=None):
with ScopedTiming("set preprocess config",self.debug_mode > 0):
ai2d_input_size=input_image_size if input_image_size else self.rgb888p_size
top,bottom,left,right,_=letterbox_pad_param(self.rgb888p_size,self.model_input_size)
self.ai2d.pad([0,0,0,0,top,bottom,left,right], 0, [0,0,0])
self.ai2d.resize(nn.interp_method.tf_bilinear, nn.interp_mode.half_pixel)
self.ai2d.build([1,3,ai2d_input_size[1],ai2d_input_size[0]],[1,3,self.model_input_size[1],self.model_input_size[0]])
def postprocess(self,results):
with ScopedTiming("postprocess",self.debug_mode > 0):
hwc_array=self.chw2hwc(self.cur_img)
det_boxes = aicube.ocr_post_process(results[0][:,:,:,0].reshape(-1), hwc_array.reshape(-1),self.model_input_size,self.rgb888p_size, self.mask_threshold, self.box_threshold)
return det_boxes
def chw2hwc(self,features):
ori_shape = (features.shape[0], features.shape[1], features.shape[2])
c_hw_ = features.reshape((ori_shape[0], ori_shape[1] * ori_shape[2]))
hw_c_ = c_hw_.transpose()
new_array = hw_c_.copy()
hwc_array = new_array.reshape((ori_shape[1], ori_shape[2], ori_shape[0]))
del c_hw_
del hw_c_
del new_array
return hwc_array
# 自定义OCR识别任务类
class OCRRecognitionApp(AIBase):
def __init__(self,kmodel_path,model_input_size,dict_path,rgb888p_size=[1920,1080],display_size=[1920,1080],debug_mode=0):
super().__init__(kmodel_path,model_input_size,rgb888p_size,debug_mode)
self.kmodel_path=kmodel_path
self.model_input_size=model_input_size
self.dict_path=dict_path
self.rgb888p_size=[ALIGN_UP(rgb888p_size[0],16),rgb888p_size[1]]
self.display_size=[ALIGN_UP(display_size[0],16),display_size[1]]
self.debug_mode=debug_mode
self.dict_word=None
self.read_dict()
self.ai2d=Ai2d(debug_mode)
self.ai2d.set_ai2d_dtype(nn.ai2d_format.RGB_packed,nn.ai2d_format.NCHW_FMT,np.uint8, np.uint8)
def config_preprocess(self,input_image_size=None,input_np=None):
with ScopedTiming("set preprocess config",self.debug_mode > 0):
ai2d_input_size=input_image_size if input_image_size else self.rgb888p_size
top,bottom,left,right,_=letterbox_pad_param(ai2d_input_size,self.model_input_size)
self.ai2d.pad([0,0,0,0,top,bottom,left,right], 0, [0,0,0])
self.ai2d.resize(nn.interp_method.tf_bilinear, nn.interp_mode.half_pixel)
self.ai2d.build([input_np.shape[0],input_np.shape[1],input_np.shape[2],input_np.shape[3]],[1,3,self.model_input_size[1],self.model_input_size[0]])
def postprocess(self,results):
with ScopedTiming("postprocess",self.debug_mode > 0):
preds = np.argmax(results[0], axis=2).reshape((-1))
output_txt = ""
for i in range(len(preds)):
if preds[i] != (len(self.dict_word) - 1) and (not (i > 0 and preds[i - 1] == preds[i])):
output_txt = output_txt + self.dict_word[preds[i]]
return output_txt
def read_dict(self):
if self.dict_path!="":
with open(self.dict_path, 'r') as file:
line_one = file.read(100000)
line_list = line_one.split("\r\n")
self.dict_word = {num: char.replace("\r", "").replace("\n", "") for num, char in enumerate(line_list)}
class OCRDetRec:
def __init__(self,ocr_det_kmodel,ocr_rec_kmodel,det_input_size,rec_input_size,dict_path,mask_threshold=0.25,box_threshold=0.3,rgb888p_size=[1920,1080],display_size=[1920,1080],debug_mode=0):
self.ocr_det_kmodel=ocr_det_kmodel
self.ocr_rec_kmodel=ocr_rec_kmodel
self.det_input_size=det_input_size
self.rec_input_size=rec_input_size
self.dict_path=dict_path
self.mask_threshold=mask_threshold
self.box_threshold=box_threshold
self.rgb888p_size=[ALIGN_UP(rgb888p_size[0],16),rgb888p_size[1]]
self.display_size=[ALIGN_UP(display_size[0],16),display_size[1]]
self.debug_mode=debug_mode
self.ocr_det=OCRDetectionApp(self.ocr_det_kmodel,model_input_size=self.det_input_size,mask_threshold=self.mask_threshold,box_threshold=self.box_threshold,rgb888p_size=self.rgb888p_size,display_size=self.display_size,debug_mode=0)
self.ocr_rec=OCRRecognitionApp(self.ocr_rec_kmodel,model_input_size=self.rec_input_size,dict_path=self.dict_path,rgb888p_size=self.rgb888p_size,display_size=self.display_size)
self.ocr_det.config_preprocess()
def run(self,input_np):
det_res=self.ocr_det.run(input_np)
boxes=[]
ocr_res=[]
for det in det_res:
self.ocr_rec.config_preprocess(input_image_size=[det[0].shape[2],det[0].shape[1]],input_np=det[0])
ocr_str=self.ocr_rec.run(det[0])
ocr_res.append(ocr_str)
boxes.append(det[1])
gc.collect()
return boxes,ocr_res
def draw_result_to_img(self, osd_img, det_res, rec_res):
if not det_res:
return
for j in range(len(det_res)):
pts = det_res[j]
for i in range(4):
x1 = pts[i*2] / self.rgb888p_size[0] * self.display_size[0]
y1 = pts[i*2+1] / self.rgb888p_size[1] * self.display_size[1]
x2 = pts[((i+1)*2)%8] / self.rgb888p_size[0] * self.display_size[0]
y2 = pts[((i+1)*2+1)%8] / self.rgb888p_size[1] * self.display_size[1]
osd_img.draw_line(int(x1),int(y1),int(x2),int(y2),color=(255,0,0,255),thickness=4)
tx = int(pts[0]/self.rgb888p_size[0]*self.display_size[0])
ty = int(pts[1]/self.rgb888p_size[1]*self.display_size[1])
text = rec_res[j]
osd_img.draw_string_advanced(tx, ty-28, 24, text, (0,0,255))
# ===================== 全局配置区 =====================
WIFI_SSID = "MNV"
WIFI_PASSWORD = "mnv@2021"
SERVER_PORT = 8080
BUFFER_SIZE = 1024
PWM_CHANNEL = 0
PWM_PIN = 42
PWM_FREQ = 1000
PWM_DEFAULT_DUTY = 50
CONSTRUCT_WITH_FPIOA = False
DISPLAY_WIDTH = ALIGN_UP(800,16)
DISPLAY_HEIGHT = 480
CAM_CHN_ID_0 = 0
CAM_CHN_ID_1 = 1
CAM_CHN_ID_2 = 2
VIDEO_WIDTH = 1280
VIDEO_HEIGHT = 720
VENC_PAYLOAD_TYPE = K_PT_H264
SAVE_DIR = "/data/video"
# ===================== OCR DET+REC 全局配置 =====================
OCR_DET_KMODEL="/sdcard/examples/kmodel/ocr_det_int16.kmodel"
OCR_REC_KMODEL="/sdcard/examples/kmodel/ocr_rec_int16.kmodel"
OCR_DICT_PATH="/sdcard/examples/utils/dict.txt"
OCR_MASK_THRESHOLD=0.25
OCR_BOX_THRESHOLD=0.3
OCR_RGB_SIZE=[640,360]
OCR_DET_INPUT=[640,640]
OCR_REC_INPUT=[512,32]
ocr_det_rec = None
ocr_lock = _thread.allocate_lock()
ocr_boxes = []
ocr_texts = []
# ===================== 图库全局变量 =====================
gallery_path = SAVE_DIR
gallery_file_list = []
gallery_page = 0
gallery_page_size = 6
gallery_mode = False
gallery_file_type = ""
gallery_selected_idx = -1
gallery_refresh_flag = True
thumb_cache = {}
image_view_mode = False
video_view_mode = False
current_media_path = ""
current_media_name = ""
wifi_ip = ""
client_conn = None
tcp_lock = _thread.allocate_lock()
yolores = []
yolo = None
yolo_running = False
yolo_lock = _thread.allocate_lock()
osd_buf_lock = _thread.allocate_lock()
draw_buf = None
show_buf = None
last_det_payload = ""
media_buf = None
recording = False
stop_record = False
video_index = 1
record_lock = _thread.allocate_lock()
pwm = None
pwm_running = False
last_click_time = 0
CLICK_COOLDOWN = 1000
dianyanum = 0
adccaiyang = 0
adcdianya = 0
djshow_status = ""
djstatus_color = (0,0,0)
adc = ADC(0)
GALLERY_THUMB_W = 210
GALLERY_THUMB_H = 105
key_short_flag = False
last_key_short_flag = False
key_lock = _thread.allocate_lock()
sensor_lock = _thread.allocate_lock()
taking_photo = False
photo_lock = _thread.allocate_lock()
BEEP_FREQ = 2000
BEEP_DUTY = 50
BEEP_DURATION_MS = 1000
beep_busy = False
person_continuous_frame = 0
PERSON_TRIGGER_FRAME = 10
MAX_CACHE_IMG = 12
def get_mp4_first_frame_rgb565(fpath):
global thumb_cache
thumb_jpg_path = fpath.replace(".mp4", ".thumb.jpg")
cache_key = f"mp4thumb_{fpath}"
if cache_key in thumb_cache:
return thumb_cache[cache_key]
try:
src_img = image.Image(thumb_jpg_path).to_rgb565()
if len(thumb_cache) >= MAX_CACHE_IMG:
del_key = list(thumb_cache.keys())[0]
del thumb_cache[del_key]
thumb_cache[cache_key] = src_img
return src_img
except Exception as e:
print(f"读取mp4配套缩略图失败 {thumb_jpg_path}: {e}")
gc.collect()
return None
def get_origin_jpg_rgb565(fpath):
global thumb_cache
cache_key = f"origin_{fpath}"
if cache_key in thumb_cache:
return thumb_cache[cache_key]
try:
src_img = image.Image(fpath).to_rgb565()
if len(thumb_cache) >= MAX_CACHE_IMG:
del_key = list(thumb_cache.keys())[0]
del thumb_cache[del_key]
thumb_cache[cache_key] = src_img
return src_img
except Exception as e:
print(f"读取原图失败 {fpath}: {e}")
gc.collect()
return None
def refresh_gallery_files():
global gallery_file_list, gallery_refresh_flag, thumb_cache
flist = []
try:
os.stat(gallery_path)
except OSError:
os.mkdir(gallery_path)
try:
all_files = list(os.listdir(gallery_path))
for fname in all_files:
if not isinstance(fname, str):
continue
full_path = f"{gallery_path}/{fname}"
if fname[-4:] in [".jpg",".mp4"] and not fname.endswith(".thumb.jpg"):
stat = os.stat(full_path)
item = (stat[8], fname, full_path, fname.split(".")[-1].lower())
flist.append(item)
flist.sort(key=lambda x: x[0], reverse=True)
gallery_file_list = flist
gallery_refresh_flag = False
thumb_cache.clear()
print(f"图库刷新完成，共{len(gallery_file_list)}个媒体文件")
except Exception as e:
print("刷新图库失败:", e)
gallery_file_list = []
gallery_refresh_flag = False
def get_gallery_total_page():
total = len(gallery_file_list)
if total == 0:
return 0
return (total + gallery_page_size - 1) // gallery_page_size
def get_current_page_files():
start = gallery_page * gallery_page_size
end = start + gallery_page_size
return gallery_file_list[start:end]
def draw_image_view_ui(buf):
buf.clear()
src_img = None
try:
src_img = image.Image(current_media_path).to_rgb565()
img_h = src_img.height()
img_w = src_img.width()
view_h = DISPLAY_HEIGHT - 100
view_w = DISPLAY_WIDTH
scale = min(view_w / img_w, view_h / img_h)
draw_x = (DISPLAY_WIDTH - int(img_w * scale)) // 2
draw_y = 50 + (view_h - int(img_h * scale)) // 2
buf.draw_image(src_img, draw_x, draw_y, x_scale=scale, y_scale=scale)
show_name = current_media_name[:24] + "..." if len(current_media_name) > 24 else current_media_name
buf.draw_string_advanced(30, 15, 24, show_name, (255,255,255))
except Exception as e:
buf.draw_string_advanced(280, 200, 28, "图片加载失败", (255,0,0))
print(f"图片查看异常: {e}")
finally:
del src_img
gc.collect()
btn_h = 60
btn_y = DISPLAY_HEIGHT - btn_h - 10
buf.draw_rectangle(320, btn_y, 160, btn_h, (80,80,80), fill=True)
buf.draw_rectangle(320, btn_y, 160, btn_h, (255,255,255), 2)
buf.draw_string_advanced(370, btn_y+18, 24, "返回", (255,255,255))
def draw_video_view_ui(buf):
buf.clear()
buf.draw_string_advanced(30, 30, 32, "视频详情", (255,255,255))
buf.draw_line(30, 75, 770, 75, (100,100,100), 2)
info_y = 110
buf.draw_string_advanced(30, info_y, 24, f"文件名: {current_media_name}", (255,255,255))
info_y += 60
try:
stat = os.stat(current_media_path)
size_kb = stat[6] / 1024
size_str = f"{size_kb:.2f} KB" if size_kb < 1024 else f"{size_kb/1024:.2f} MB"
buf.draw_string_advanced(30, info_y, 24, f"文件大小: {size_str}", (255,255,255))
info_y += 60
t = time.localtime(stat[8])
time_str = f"{t[0]}-{t[1]:02d}-{t[2]:02d} {t[3]:02d}:{t[4]:02d}:{t[5]:02d}"
buf.draw_string_advanced(30, info_y, 24, f"创建时间: {time_str}", (255,255,255))
info_y += 60
except Exception as e:
buf.draw_string_advanced(30, info_y, 24, "获取文件信息失败", (255,0,0))
buf.draw_string_advanced(30, info_y + 40, 22, "提示：设备端暂不支持视频播放，可导出至电脑查看", (255,200,0))
btn_h = 60
btn_y = DISPLAY_HEIGHT - btn_h - 10
buf.draw_rectangle(320, btn_y, 160, btn_h, (80,80,80), fill=True)
buf.draw_rectangle(320, btn_y, 160, btn_h, (255,255,255), 2)
buf.draw_string_advanced(370, btn_y+18, 24, "返回", (255,255,255))
def draw_gallery_ui(buf):
global gallery_page, gallery_refresh_flag, thumb_cache
buf.clear()
total_page = get_gallery_total_page()
title = f"媒体图库 | {gallery_page+1}/{total_page}" if total_page > 0 else "媒体图库 | 无文件"
buf.draw_string_advanced(30, 10, 28, title, (255,255,255))
if gallery_refresh_flag:
refresh_gallery_files()
thumb_cache.clear()
gc.collect()
current_files = get_current_page_files()
item_w = 230
item_h = 160
gap_x = 20
gap_y = 20
start_x = 20
start_y = 60
thumb_w = item_w - 20
thumb_h = item_h - 55
page_item_with_pos = []
for idx, file_info in enumerate(current_files):
ctime, fname, fpath, ftype = file_info
row = idx // 3
col = idx % 3
x = start_x + col * (item_w + gap_x)
y = start_y + row * (item_h + gap_y)
is_selected = (gallery_page * gallery_page_size + idx) == gallery_selected_idx
border_color = (0,255,255) if is_selected else (200,200,200)
buf.draw_rectangle(x, y, item_w, item_h, (60,60,60), fill=True)
buf.draw_rectangle(x, y, item_w, item_h, border_color, 2)
thumb_x = x + 10
thumb_y = y + 10
src_img = None
if ftype == "jpg":
src_img = get_origin_jpg_rgb565(fpath)
elif ftype == "mp4":
src_img = get_mp4_first_frame_rgb565(fpath)
if src_img:
ori_w = src_img.width()
ori_h = src_img.height()
scale = min(thumb_w / ori_w, thumb_h / ori_h)
buf.draw_image(src_img, thumb_x, thumb_y, x_scale=scale, y_scale=scale)
else:
tip = "图片" if ftype == "jpg" else "视频"
color = (0,255,0) if ftype == "jpg" else (255,100,0)
buf.draw_rectangle(thumb_x, thumb_y, thumb_w, thumb_h, (80,80,80), fill=True)
buf.draw_string_advanced(thumb_x + (thumb_w - len(tip)*12)//2, thumb_y + thumb_h//2 - 12, 24, tip, color)
show_name = fname[:14] + "..." if len(fname) > 14 else fname
buf.draw_string_advanced(x+10, y + item_h - 35, 18, show_name, (255,255,255))
page_item_with_pos.append((ctime, fname, fpath, ftype, x, y, item_w, item_h))
btn_h = 70
btn_y = DISPLAY_HEIGHT - btn_h - 10
buf.draw_rectangle(20, btn_y, 160, btn_h, (80,80,80), fill=True)
buf.draw_rectangle(20, btn_y, 160, btn_h, (255,255,255), 2)
buf.draw_string_advanced(60, btn_y+20, 24, "返回", (255,255,255))
buf.draw_rectangle(200, btn_y, 160, btn_h, (80,80,80), fill=True)
buf.draw_rectangle(200, btn_y, 160, btn_h, (255,255,255), 2)
buf.draw_string_advanced(240, btn_y+20, 24, "上一页", (255,255,255))
buf.draw_rectangle(380, btn_y, 160, btn_h, (80,80,80), fill=True)
buf.draw_rectangle(380, btn_y, 160, btn_h, (255,255,255), 2)
buf.draw_string_advanced(420, btn_y+20, 24, "下一页", (255,255,255))
buf.draw_rectangle(560, btn_y, 200, btn_h, (120,20,20), fill=True)
buf.draw_rectangle(560, btn_y, 200, btn_h, (255,100,100), 2)
buf.draw_string_advanced(610, btn_y+20, 24, "删除选中", (255,255,255))
gc.collect()
return page_item_with_pos
tcp_task_queue = []
queue_lock = _thread.allocate_lock()
conn_valid_map = {}
map_lock = _thread.allocate_lock()
def get_time_str():
t = time.localtime()
return f"{t[0]}-{t[1]:02d}-{t[2]:02d}-{t[3]:02d}-{t[4]:02d}-{t[5]:02d}"
def wifi_connect():
global wifi_ip
sta = network.WLAN(0)
sta.active(True)
if sta.isconnected():
wifi_ip = sta.ifconfig()[0]
return wifi_ip
print(f"连接WiFi: {WIFI_SSID}...")
sta.connect(WIFI_SSID, WIFI_PASSWORD)
retry = 0
while not sta.isconnected() and retry < 50:
os.exitpoint()
time.sleep_ms(200)
retry += 1
if not sta.isconnected():
raise Exception("WiFi连接失败")
wifi_ip = sta.ifconfig()[0]
print(f"WiFi已连接 | IP:{wifi_ip}")
return wifi_ip
def key_detect_thread():
global djshow_status, djstatus_color, key_short_flag, last_key_short_flag
key = Pin(2, Pin.IN, pull=Pin.PULL_UP)
print("硬件按键检测线程已启动，实时监控Pin2短接状态")
while True:
os.exitpoint()
lv = key.value()
key_lock.acquire()
last_key_short_flag = key_short_flag
if lv == 0:
djshow_status = "按键:已短接"
djstatus_color = (0,255,0)
key_short_flag = True
if not last_key_short_flag:
print("已短接，开始拍照，临时暂停YOLO")
ts = get_time_str()
save_path = f"{SAVE_DIR}/pic-{ts}.jpg"
photo_lock.acquire()
taking_photo = True
photo_lock.release()
sensor_lock.acquire()
try:
img = uvc_snapshot_sim(chn=CAM_CHN_ID_2)
if img is not None:
img_np = img.to_numpy_ref().copy()
resize_img = cv2.resize(img_np.copy(), (640, 480), interpolation=cv2.INTER_LINEAR)
resize_img_bgr = cv2.cvtColor(resize_img, cv2.COLOR_RGB2BGR)
cv2.imwrite(save_path, resize_img_bgr)
print(f"拍照完成：{save_path}")
gallery_refresh_flag = True
except Exception as e:
print("拍照失败：", e)
finally:
sensor_lock.release()
photo_lock.acquire()
taking_photo = False
photo_lock.release()
else:
djshow_status = "按键:未短接"
djstatus_color = (255,0,0)
key_short_flag = False
key_lock.release()
time.sleep_ms(50)
def handle_client_message(msg, conn):
map_lock.acquire()
conn_valid_map[id(conn)] = True
map_lock.release()
queue_lock.acquire()
tcp_task_queue.append((msg, conn))
queue_lock.release()
return f"ACCEPT:{msg}"
def tcp_worker_thread():
global pwm_running,yolo_running
while True:
os.exitpoint()
task = None
queue_lock.acquire()
if len(tcp_task_queue) > 0:
task = tcp_task_queue.pop(0)
queue_lock.release()
if task is None:
time.sleep_ms(50)
continue
msg, conn = task
cid = id(conn)
alive = False
map_lock.acquire()
if cid in conn_valid_map and conn_valid_map[cid]:
alive = True
map_lock.release()
if not alive:
print(f"连接已失效，丢弃任务:{msg}")
continue
resp = ""
if msg == "PING":
resp = "PONG"
elif msg == "YOLO_ON":
yolo_lock.acquire()
yolo_running = True
yolo_lock.release()
resp = "OK:YOLO_ON"
elif msg == "YOLO_OFF":
yolo_lock.acquire()
yolo_running = False
yolo_lock.release()
resp = "OK:YOLO_OFF"
elif msg == "WIFI_INFO":
resp = f"IP:{wifi_ip},PORT:{SERVER_PORT}"
elif msg == "GET_YOLO":
if not yolores:
resp = "YOLO:NO_DETECTION"
else:
res_str = ";".join([f"{i[0]}:{i[1]:.2f}" for i in yolores])
resp = f"YOLO:{res_str}"
elif msg == "PWM_ON":
if pwm and not pwm_running:
pwm.duty(PWM_DEFAULT_DUTY)
pwm_running = True
resp = "OK:PWM_ON"
else:
resp = "FAIL:PWM"
elif msg == "PWM_OFF":
if pwm and pwm_running:
pwm.duty(0)
pwm_running = False
resp = "OK:PWM_OFF"
else:
resp = "FAIL:PWM"
elif msg == "BTN_SHORT":
key_lock.acquire()
lv = 1 if key_short_flag else 0
key_lock.release()
resp = f"BTN_SHORT:{lv}"
elif msg == "SEND_IMG":
if not gallery_mode and not image_view_mode and not video_view_mode:
send_image_over_tcp()
resp = "OK:SEND_IMG"
else:
resp = "FAIL:SEND_IMG:not in main camera view"
else:
resp = f"UNKNOWN:{msg}"
try:
send_data = (resp + "\n").encode("utf-8")
conn.send(send_data)
print(f"后台回复客户端: {resp}")
except Exception as e:
print(f"后台发送失败 {msg}: {e}")
map_lock.acquire()
if cid in conn_valid_map:
conn_valid_map[cid] = False
map_lock.release()
def tcp_server_thread():
global client_conn
_thread.start_new_thread(tcp_worker_thread, ())
print("TCP后台工作线程已启动")
try:
wifi_connect()
pwm_init()
except Exception as e:
print("WiFi/硬件初始化失败:", e)
return
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.settimeout(2.0)
try:
server_sock.bind((wifi_ip, SERVER_PORT))
server_sock.listen(3)
print(f"TCP服务器启动: {wifi_ip}:{SERVER_PORT}")
except Exception as e:
print(f"服务器绑定失败: {e}")
return
def heartbeat_send_loop(conn, addr, cid):
time.sleep_ms(1000)
while True:
os.exitpoint()
alive = False
map_lock.acquire()
if cid in conn_valid_map and conn_valid_map[cid]:
alive = True
map_lock.release()
if not alive:
break
try:
ocr_lock.acquire()
ocr_cnt = len(ocr_boxes)
ocr_lock.release()
base_msg = f"PONG-{1 if recording else 0}-{1 if yolo_running else 0}-OCR:{ocr_cnt}"
yolo_part = ""
yolo_lock.acquire()
tmp_yolo = yolores
yolo_lock.release()
MIN_CONF = 0.20
if tmp_yolo and len(tmp_yolo)==3 and len(tmp_yolo[1])>0:
boxes_arr = tmp_yolo[0]
clsid_list = tmp_yolo[1]
conf_list = tmp_yolo[2]
yolo_items = []
for idx in range(len(clsid_list)):
try:
cid = int(clsid_list[idx])
conf = float(conf_list[idx])
if conf < MIN_CONF:
continue
box_np = boxes_arr[idx]
x1 = float(box_np[0])
y1 = float(box_np[1])
w = float(box_np[2])
h = float(box_np[3])
x2 = x1 + w
y2 = y1 + h
yolo_items.append(f"[{cid},{conf:.3f},{x1:.1f},{y1:.1f},{x2:.1f},{y2:.1f}]")
except Exception as e:
print(f"[心跳组装yolo单条异常 idx={idx}, e={e}]")
continue
yolo_part = ";".join(yolo_items)
ocr_part = ""
ocr_lock.acquire()
tmp_ocr_boxes = ocr_boxes
tmp_ocr_texts = ocr_texts
ocr_lock.release()
ocr_items = []
for box, txt in zip(tmp_ocr_boxes, tmp_ocr_texts):
box_str = ",".join([str(int(v)) for v in box])
safe_txt = txt.replace("@","").replace("|","").replace(";","")
ocr_items.append(f"[{box_str}@{safe_txt}]")
if len(ocr_items)>0:
ocr_part = ";".join(ocr_items)
if yolo_part != "":
seg_yolo = f"YOLO|{yolo_part}"
else:
seg_yolo = "YOLO|"
if ocr_part != "":
seg_ocr = f"OCR|{ocr_part}"
else:
seg_ocr = "OCR|"
det_payload = f"DET_RESULT:{seg_yolo}||{seg_ocr}"
full_message = base_msg + "|" + det_payload
message = full_message + "\n"
tcp_lock.acquire()
try:
socket_send_all(conn, message.encode("utf-8"))
except OSError as e:
print(f"心跳发送失败 {addr}: {e}")
map_lock.acquire()
if cid in conn_valid_map:
conn_valid_map[cid] = False
map_lock.release()
break
finally:
tcp_lock.release()
except Exception as e:
print(f"心跳外层异常 {addr}: {e}")
import sys
sys.print_exception(e)
map_lock.acquire()
if cid in conn_valid_map:
conn_valid_map[cid] = False
map_lock.release()
break
time.sleep_ms(1000)
def session_loop(conn, addr):
global client_conn
print(f"客户端已连接: {addr}")
conn.settimeout(1.0)
cid = id(conn)
tcp_lock.acquire()
client_conn = conn
tcp_lock.release()
map_lock.acquire()
conn_valid_map[cid] = True
map_lock.release()
_thread.start_new_thread(heartbeat_send_loop, (conn, addr, cid))
cache = b""
session_alive = True
while session_alive:
os.exitpoint()
try:
data = conn.recv(BUFFER_SIZE)
cache += data
while b"\n" in cache:
line, cache = cache.split(b"\n", 1)
line = line.strip()
if not line:
continue
msg = line.decode("utf-8")
print(f"收到[{addr}]：{msg}")
if msg == "HELLO":
resp = "HELLO_OK\n"
tcp_lock.acquire()
try:
conn.send(resp.encode("utf-8"))
print("握手成功，回复HELLO_OK")
except Exception as e:
print(f"握手回复失败 {e}")
session_alive = False
break
finally:
tcp_lock.release()
continue
elif msg == "BYE":
session_alive = False
break
tmp_resp = handle_client_message(msg, conn)
tcp_lock.acquire()
try:
conn.send((tmp_resp + "\n").encode("utf-8"))
print(f"即时ACK回复: {tmp_resp}")
except Exception as e:
print(f"ACK发送失败 {msg}: {e}")
session_alive = False
break
finally:
tcp_lock.release()
except socket.timeout:
continue
except OSError as e:
if e.args[0] == errno.EAGAIN or e.args[0] == errno.EWOULDBLOCK:
continue
print(f"客户端[{addr}]OS异常断开: {e}")
session_alive = False
except Exception as e:
print(f"客户端[{addr}]会话异常: {e}")
session_alive = False
map_lock.acquire()
if cid in conn_valid_map:
conn_valid_map[cid] = False
map_lock.release()
tcp_lock.acquire()
if client_conn == conn:
client_conn = None
tcp_lock.release()
try:
conn.close()
except:
pass
print(f"客户端 {addr} 连接已关闭\n")
while True:
os.exitpoint()
conn = None
addr = None
try:
conn, addr = server_sock.accept()
_thread.start_new_thread(session_loop, (conn, addr))
except OSError as e:
if e.args[0] != errno.EAGAIN and e.args[0] != errno.EWOULDBLOCK:
print(f"TCP服务端accept异常: {e}")
time.sleep(1)
except Exception:
pass
def pwm_init():
global pwm
if pwm:
return
try:
pwm = PWM(PWM_PIN, freq=PWM_FREQ, duty=0)
print("[PWM] 通道初始化完成，同时支持输出与蜂鸣")
except Exception as e:
pwm = None
print("[WARN] PWM初始化失败:", e)
def beep_alert_thread():
global beep_busy, pwm, pwm_running
if not pwm or beep_busy:
return
beep_busy = True
old_state = pwm_running
pwm.freq(BEEP_FREQ)
pwm.duty(BEEP_DUTY)
time.sleep_ms(BEEP_DURATION_MS)
pwm.freq(PWM_FREQ)
if old_state:
pwm.duty(PWM_DEFAULT_DUTY)
else:
pwm.duty(0)
beep_busy = False
print("连续10帧检测到人，蜂鸣1秒完成，恢复PWM原有状态")
def yolo_detect_thread():
global yolores, yolo, ocr_det_rec, ocr_boxes, ocr_texts
kmodel_path="/sdcard/examples/zhadai.kmodel"
labels = ['cable_tie']
try:
model_input_size=[640,640]
display_size=[800, 480]
rgb888p_size=[224,224]
confidence_threshold = 0.1
nms_threshold=0.1
yolo=YOLOv5(task_type="detect",mode="video",kmodel_path=kmodel_path,labels=labels,
rgb888p_size=rgb888p_size,model_input_size=model_input_size,display_size=display_size,
conf_thresh=confidence_threshold,nms_thresh=nms_threshold,max_boxes_num=50,debug_mode=0)
yolo.config_preprocess()
print("YOLO模型加载完成（等待按钮启动）")
except Exception as e:
print("YOLO模型加载失败:", e)
return
try:
ocr_det_rec=OCRDetRec(OCR_DET_KMODEL,
OCR_REC_KMODEL,
det_input_size=OCR_DET_INPUT,
rec_input_size=OCR_REC_INPUT,
dict_path=OCR_DICT_PATH,
mask_threshold=OCR_MASK_THRESHOLD,
box_threshold=OCR_BOX_THRESHOLD,
rgb888p_size=OCR_RGB_SIZE,
display_size=[DISPLAY_WIDTH,DISPLAY_HEIGHT],
debug_mode=0)
print("OCR Det+Rec 检测识别模型加载完成")
except Exception as e:
print("OCR Det+Rec模型加载失败:", e)
ocr_det_rec = None
OCR_RUN_INTERVAL = 3
ocr_frame_cnt = 0
while True:
with ScopedTiming("total",0):
os.exitpoint()
photo_lock.acquire()
tmp_taking = taking_photo
photo_lock.release()
if tmp_taking:
yolores = []
ocr_lock.acquire()
ocr_boxes = []
ocr_texts = []
ocr_lock.release()
time.sleep_ms(50)
continue
yolo_lock.acquire()
running = yolo_running
yolo_lock.release()
if not running:
yolores = []
ocr_lock.acquire()
ocr_boxes = []
ocr_texts = []
ocr_lock.release()
time.sleep_ms(100)
continue
try:
sensor_lock.acquire()
img = uvc_snapshot_sim(chn=CAM_CHN_ID_2)
sensor_lock.release()
if img is None:
time.sleep_ms(20)
continue
img_np = img.to_numpy_ref()
img_resize = cv2.resize(img_np, (224, 224), interpolation=cv2.INTER_LINEAR)
img_chw = hwc2chw(img_resize)
yolores = yolo.run(img_chw)
res_str = ""
has_person = False
tmp_yolo=yolores
if yolores and len(yolores) == 3 and len(yolores[1]) > 0:
boxes_arr = yolores[0]
cls_ids = yolores[1]
confs = yolores[2]
res_list = []
has_person = False
for i in range(len(cls_ids)):
cls_id = int(cls_ids[i])
conf = float(confs[i])
box_np = boxes_arr[i]
x1 = float(box_np[0])
y1 = float(box_np[1])
w = float(box_np[2])
h = float(box_np[3])
x2 = x1 + w
y2 = y1 + h
label = labels[cls_id] if cls_id < len(labels) else "unknown"
res_list.append(f"{label}:{conf:.2f},box=[{x1:.1f},{y1:.1f},{x2:.1f},{y2:.1f}]")
if label == "cable_tie":
has_person = True
res_str = ";".join(res_list)
tcp_lock.acquire()
has_client = client_conn is not None
tcp_lock.release()
if has_client:
try:
client_conn.send(f"YOLO_DETECT:{res_str}\n".encode())
except Exception:
tcp_lock.acquire()
client_conn = None
tcp_lock.release()
else:
yolores = []
if ocr_det_rec is not None:
ocr_frame_cnt += 1
if ocr_frame_cnt >= OCR_RUN_INTERVAL:
ocr_frame_cnt = 0
ocr_in_img = cv2.resize(img_np.copy(), tuple(OCR_RGB_SIZE), interpolation=cv2.INTER_LINEAR)
ocr_img_chw = hwc2chw(ocr_in_img)
boxes, texts = ocr_det_rec.run(ocr_img_chw)
ocr_lock.acquire()
ocr_boxes = boxes
ocr_texts = texts
ocr_lock.release()
global person_continuous_frame
if has_person:
person_continuous_frame += 1
if person_continuous_frame >= PERSON_TRIGGER_FRAME and not beep_busy:
_thread.start_new_thread(beep_alert_thread, ())
else:
person_continuous_frame = 0
gc.collect()
time.sleep_ms(5)
except Exception as e:
print(f"YOLO/OCR线程异常: {e}")
time.sleep_ms(100)
def record_mp4_thread(save_path, width, height):
global recording, stop_record, video_index
venc_chn = VENC_CHN_ID_2
width_align = ALIGN_UP(width, 16)
video_payload = K_MP4_CODEC_ID_H264
while not stop_record:
ts = get_time_str()
mp4_filepath = f"{save_path}/video_{ts}.mp4"
thumb_filepath = f"{save_path}/video_{ts}.thumb.jpg"
print(f"开始录制: {mp4_filepath}")
mp4 = mp4_muxer_init(mp4_filepath, True)
track = mp4_muxer_create_video_track(mp4, width_align, height, video_payload)
cnt = 0
thumb_saved = False
while cnt < 200 and not stop_record:
try:
# UVC模式：软件把UVC取出图像送入编码器
sensor_lock.acquire()
snap_img = uvc_snapshot_sim(CAM_CHN_ID_2)
sensor_lock.release()
if snap_img is None:
time.sleep_ms(10)
continue
encoder.SendFrame(venc_chn, snap_img)
encoder.GetStream(venc_chn, streamData)
pts = streamData.pts[0]
data = streamData.data[0]
size = streamData.data_size[0]
fd = k_mp4_frame_data_s()
fd.data = data
fd.data_length = size
fd.time_stamp = pts
kd_mp4_write_frame(mp4, track, fd)
encoder.ReleaseStream(venc_chn, streamData)
if not thumb_saved and cnt == 20:
try:
snap_np = snap_img.to_numpy_ref().copy()
src_h, src_w = snap_np.shape[0], snap_np.shape[1]
scale = min(GALLERY_THUMB_W / src_w, GALLERY_THUMB_H / src_h)
dst_w = int(src_w * scale)
dst_h = int(src_h * scale)
resize_img = cv2.resize(snap_np.copy(), (dst_w, dst_h), interpolation=cv2.INTER_LINEAR)
resize_img_bgr = cv2.cvtColor(resize_img, cv2.COLOR_RGB2BGR)
cv2.imwrite(thumb_filepath, resize_img_bgr)
thumb_saved = True
print(f"mp4配套缩略图已保存 {thumb_filepath} 尺寸:{dst_w}x{dst_h}")
except Exception as e:
print("保存mp4缩略图失败:", e)
cnt +=1
except Exception as e:
print(f"record loop except {e}")
break
kd_mp4_destroy_tracks(mp4)
kd_mp4_destroy(mp4)
video_index +=1
record_lock.acquire()
recording = False
stop_record = False
record_lock.release()
def mp4_muxer_init(fn, fmp4):
cfg = k_mp4_config_s()
cfg.config_type = K_MP4_CONFIG_MUXER
cfg.muxer_config.file_name[:] = bytes(fn, 'utf-8')
cfg.muxer_config.fmp4_flag = fmp4
h = k_u64_ptr()
ret = kd_mp4_create(h, cfg)
if ret:
raise OSError("kd_mp4_create failed.")
return h.value
def mp4_muxer_create_video_track(h, w, hh, pt):
info = k_mp4_track_info_s()
info.track_type = K_MP4_STREAM_VIDEO
info.time_scale = 1000
info.video_info.width = w
info.video_info.height = hh
info.video_info.codec_id = pt
th = k_u64_ptr()
kd_mp4_create_track(h, th, info)
return th.value
if __name__ == "__main__":
_thread.start_new_thread(tcp_server_thread, ())
_thread.start_new_thread(key_detect_thread, ())
time.sleep_ms(2000)
Display.init(Display.ST7701, width=DISPLAY_WIDTH, height=DISPLAY_HEIGHT, osd_num=2,to_ide=True)
# LAYER_OSD0 底层：RGB565，UVC视频画面
Display.config_layer(
rect=(0, 0, 800, 480),
pix_format=image.RGB565,
layer=Display.LAYER_OSD1,
alpha=255,
flag=0
)
# LAYER_OSD2 上层：ARGB8888，UI按钮文字，硬件alpha叠加
Display.config_layer(
rect=(0, 0, 800, 480),
pix_format=image.ARGB8888,
layer=Display.LAYER_OSD0,
alpha=255,
flag=0
)
print("方案B：双OSD硬件图层配置完成 OSD0(RGB565)+OSD2(ARGB8888)")
print("等待UVC摄像头插入...")
while True:
plugin, dev = UVC.probe()
if plugin:
print(f"detect USB Camera {dev}")
break
mode = UVC.video_mode(640, 480, UVC.FOURCC_MJPEG, 30)
succ, mode = UVC.select_video_mode(mode)
print(f"select mode success: {succ}, mode: {mode}")
UVC.start(cvt=False)
touch = TOUCH(0)
key = Pin(2, Pin.IN, pull=Pin.PULL_UP)
# OSD上层 ARGB8888 UI绘制层
show_buf = image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.ARGB8888)
# UVC底层背景画布 RGB565，给到 LAYER_VIDEO0
uvc_bg_buf = image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.RGB565)
# 图库看图专用RGB565画布
media_buf = image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.RGB565)
# ========= 编码器初始化，软件SendFrame送入RGB图像 =========
encoder = Encoder()
venc_chn = VENC_CHN_ID_2
encoder.SetOutBufs(venc_chn,VIDEO_WIDTH,VIDEO_HEIGHT)
MediaManager.init()
chnAttr = ChnAttrStr(encoder.PAYLOAD_TYPE_H264, encoder.H264_PROFILE_MAIN, VIDEO_WIDTH, VIDEO_HEIGHT)
streamData = StreamData()
encoder.Create(chnAttr)
encoder.Start()
_thread.start_new_thread(yolo_detect_thread, ())
FONT = 24
last_touch = False
page_file_items = []
try:
while True:
os.exitpoint()
# =========渲染分支=========
if image_view_mode:
# 图片查看，关闭UI层OSD2
Display.config_layer((0,0,0,0), image.ARGB8888, Display.LAYER_OSD2, alpha=0)
draw_image_view_ui(media_buf)
Display.show_image(media_buf, layer=Display.LAYER_OSD0)
elif gallery_mode:
# 图库模式，关闭UI层OSD2
Display.config_layer((0,0,0,0), image.ARGB8888, Display.LAYER_OSD2, alpha=0)
page_file_items = draw_gallery_ui(media_buf)
Display.show_image(media_buf, layer=Display.LAYER_OSD0)
elif video_view_mode:
# 视频详情，关闭UI层OSD2
Display.config_layer((0,0,0,0), image.ARGB8888, Display.LAYER_OSD2, alpha=0)
draw_video_view_ui(media_buf)
Display.show_image(media_buf, layer=Display.LAYER_OSD0)
else:
# 切回相机，打开上层UI OSD2
Display.config_layer((0, 0, DISPLAY_WIDTH, DISPLAY_HEIGHT), Display.PIX_FMT_ARGB8888, Display.LAYER_OSD2, alpha=255)
osd_buf_lock.acquire()
try:
show_buf.clear()
yolo_lock.acquire()
running = yolo_running
yolo_lock.release()
if running and yolores:
yolo.draw_result(yolores, show_buf)
if running and ocr_det_rec is not None:
ocr_lock.acquire()
local_box = ocr_boxes
local_txt = ocr_texts
ocr_lock.release()
ocr_det_rec.draw_result_to_img(show_buf, local_box, local_txt)
# 全部颜色补全4元ARGB (R,G,B,A)，A=255不透明
show_buf.draw_string_advanced(10,10, FONT, f"录制:{'中' if recording else '空闲'}", (255,0,0,255) if recording else (0,255,0,255))
tcp_lock.acquire()
tcp_str = "客户端:已连接" if client_conn else "客户端:未连接"
tcp_lock.release()
show_buf.draw_string_advanced(10,40, FONT, tcp_str, (0,0,255,255))
show_buf.draw_string_advanced(10,70, FONT, f"YOLO+OCR:{'运行中' if running else '已停止'}", (0,255,120,255) if running else (255,100,0,255))
show_buf.draw_string_advanced(10,100, FONT, f"ADC:{adcdianya}", (0,255,0,255))
show_buf.draw_string_advanced(10,130,FONT, djshow_status, djstatus_color + (255,))
ip_str = f"IP:{wifi_ip}" if wifi_ip else "IP:未连接"
show_buf.draw_string_advanced(600,10, FONT, ip_str, (255,255,0,255))
dianyanum +=1
if dianyanum>10:
adccaiyang = adc.read_u16()
adcdianya = adc.read_uv()
dianyanum=0
yolo_btn_text = "停止检测" if running else "开始检测"
buttons = [
[0, 400, 200, 80, "结束录制" if recording else "开始录制", "rec"],
[400, 400, 200, 80, "图库", "tuku"],
[600, 400, 200, 80, yolo_btn_text, "yolo_btn"],
[200, 400, 200, 80, "拍照", "pic"],
]
for x,y,w,h,t,bid in buttons:
show_buf.draw_rectangle(x,y,w,h,(220,220,220,255),fill=True)
show_buf.draw_rectangle(x,y,w,h,(180,180,180,255),2)
show_buf.draw_string_advanced(x+40,y+30,18,t,(0,0,0,255))
finally:
osd_buf_lock.release()
sensor_lock.acquire()
cam_frame = uvc_snapshot_sim(CAM_CHN_ID_2)
sensor_lock.release()
if cam_frame is not None:
cam_frame_rgb565 = cam_frame.to_rgb565()
uvc_bg_buf.clear()
# 软件缩放UVC 640x480 → 800x480，绘制到底层画布
uvc_bg_buf.draw_image(cam_frame_rgb565, 0,0,
x_scale=DISPLAY_WIDTH / 640.0,
y_scale=DISPLAY_HEIGHT /480.0
)
del cam_frame
del cam_frame_rgb565
else:
uvc_bg_buf.clear((0,0,0))
# =====【关键】分别输出到两个OSD硬件图层，VO硬件做叠加，没有软件draw_image混合！=====
Display.show_image(uvc_bg_buf, layer=Display.LAYER_OSD0)
Display.show_image(show_buf, layer=Display.LAYER_OSD2)
# =====================触摸检测逻辑完全不变=====================
tp = touch.read()
down = False
tx,ty=0,0
for p in tp:
if p.event == TOUCH.EVENT_DOWN:
down=True; tx=p.x; ty=p.y
if down and not last_touch:
current_time = time.ticks_ms()
if time.ticks_diff(current_time, last_click_time) > CLICK_COOLDOWN:
last_click_time = current_time
# 图片查看模式
if image_view_mode:
btn_h = 60
btn_y = DISPLAY_HEIGHT - btn_h - 10
if 320 <= tx <= 480 and btn_y <= ty <= btn_y+btn_h:
image_view_mode = False
gallery_refresh_flag = True
# 图库模式
elif gallery_mode:
total_page = get_gallery_total_page()
btn_h = 70
btn_y = DISPLAY_HEIGHT - btn_h - 10
if 20 <= tx <= 180 and btn_y <= ty <= btn_y+btn_h:
gallery_mode = False
gallery_page = 0
gallery_refresh_flag = True
thumb_cache.clear()
gc.collect()
elif 200 <= tx <= 360 and btn_y <= ty <= btn_y+btn_h:
if gallery_page > 0:
thumb_cache.clear()
gc.collect()
gallery_page -= 1
elif 380 <= tx <= 540 and btn_y <= ty <= btn_y+btn_h:
if gallery_page < total_page -1:
thumb_cache.clear()
gc.collect()
gallery_page += 1
elif 560 <= tx <= 760 and btn_y <= ty <= btn_y+btn_h:
if gallery_selected_idx != -1 and gallery_selected_idx < len(gallery_file_list):
del_path = gallery_file_list[gallery_selected_idx][2]
try:
os.remove(del_path)
print(f"已删除文件: {del_path}")
if del_path.endswith(".mp4"):
thumb_del = del_path.replace(".mp4", ".thumb.jpg")
try:
os.remove(thumb_del)
except OSError:
pass
except Exception as e:
print("删除失败:", e)
gallery_selected_idx = -1
gallery_refresh_flag = True
thumb_cache.clear()
gc.collect()
else:
for idx, item in enumerate(page_file_items):
_, fname, fpath, ftype, x, y, w, h = item
if x <= tx <= x+w and y <= ty <= y+h:
gallery_selected_idx = gallery_page * gallery_page_size + idx
current_media_path = fpath
current_media_name = fname
if ftype == "jpg":
image_view_mode = True
else:
video_view_mode = True
break
elif video_view_mode:
btn_h = 60
btn_y = DISPLAY_HEIGHT - btn_h - 10
if 320 <= tx <= 480 and btn_y <= ty <= btn_y+btn_h:
video_view_mode = False
gallery_refresh_flag = True
else:
for x,y,w,h,t,bid in buttons:
if x<=tx<=x+w and y<=ty<=y+h:
if bid=="rec":
record_lock.acquire()
if not recording:
recording=True; stop_record=False
_thread.start_new_thread(record_mp4_thread,("/data/video",VIDEO_WIDTH,VIDEO_HEIGHT))
else:
stop_record=True; recording=False
record_lock.release()
elif bid=="yolo_btn":
yolo_lock.acquire()
yolo_running = not yolo_running
yolo_lock.release()
print(f"YOLO+OCR: {'启动' if yolo_running else '停止'}")
elif bid=="tuku":
gallery_mode = True
gallery_page = 0
gallery_selected_idx = -1
gallery_refresh_flag = True
elif bid=="pic":
ts = get_time_str()
save_path = f"{SAVE_DIR}/pic-touch-{ts}.jpg"
photo_lock.acquire()
taking_photo = True
photo_lock.release()
sensor_lock.acquire()
try:
img = uvc_snapshot_sim(chn=CAM_CHN_ID_2)
if img is not None:
img_np = img.to_numpy_ref().copy()
resize_img = cv2.resize(img_np.copy(), (640, 480), interpolation=cv2.INTER_LINEAR)
resize_img_bgr = cv2.cvtColor(resize_img, cv2.COLOR_RGB2BGR)
cv2.imwrite(save_path, resize_img_bgr)
print(f"拍照完成：{save_path}")
gallery_refresh_flag = True
except Exception as e:
print("触摸拍照失败：", e)
finally:
sensor_lock.release()
photo_lock.acquire()
taking_photo = False
photo_lock.release()
last_touch = down
except Exception as e:
print("main loop exception:",e)
import sys
sys.print_exception(e)
finally:
# 退出先关闭两个OSD图层
Display.config_layer((0,0,0,0), image.RGB565, Display.LAYER_OSD0, alpha=0)
Display.config_layer((0,0,0,0), image.ARGB8888, Display.LAYER_OSD2, alpha=0)
UVC.stop()
encoder.Stop()
MediaManager.deinit()
Display.deinit()
del uvc_bg_buf
del show_buf
del media_buf
tcp_lock.acquire()
if client_conn:
try:
client_conn.close()
except:pass
tcp_lock.release()
print("系统退出，资源全部释放完成")
