""" Minimal LVGL single-frame capture application. Sensor id 2 channel 0 outputs RGB888 at the requested capture resolution. The media thread repeatedly calls snapshot(); only the IDE virtual preview is fitted to the 800x480 LVGL canvas. CAPTURE stops requesting new frames, SAVE stores the original-resolution frame, and DISCARD resumes preview. LVGL uses the top OSD1 layer for controls only. """ import _thread import gc import os import sys import time import image import lvgl as lv import uctypes from machine import TOUCH from media.display import * from media.media import * from media.sensor import * DISPLAY_WIDTH = ALIGN_UP(800, 16) DISPLAY_HEIGHT = 480 DISPLAY_TYPE = Display.ST7701 SENSOR_ID = 2 CAPTURE_CHANNEL = CAM_CHN_ID_0 DEFAULT_WIDTH = 800 DEFAULT_HEIGHT = 480 IMAGE_DIR = "/data/images" JPEG_QUALITY = 95 STATE_CONFIG = 0 STATE_STARTING = 1 STATE_LIVE = 2 STATE_CAPTURING = 3 STATE_FROZEN = 4 STATE_SAVING = 5 STATE_ERROR = 6 CMD_NONE = 0 CMD_CAPTURE = 1 CMD_SAVE = 2 CMD_DISCARD = 3 disp_images = None width_input = None height_input = None keyboard = None main_button = None main_button_label = None save_button = None discard_button = None status_label = None resolution_label = None app_running = True worker_started = False worker_done = True media_command = CMD_NONE app_state = STATE_CONFIG status_text = "Set resolution and start preview" ui_dirty = True capture_index = 1 display_lock = _thread.allocate_lock() def print_error(err): printer = getattr(sys, "print_exception", None) if printer: printer(err) else: print("exception: %s" % err) def ensure_directory(path): current = "" for part in path.strip("/").split("/"): current += "/" + part try: os.mkdir(current) except OSError: os.stat(current) def find_next_index(): next_index = 1 try: for name in os.listdir(IMAGE_DIR): if not name.startswith("capture_") or not name.endswith(".jpg"): continue try: index = int(name.split("_")[1]) if index >= next_index: next_index = index + 1 except (ValueError, IndexError): pass except OSError: pass return next_index class IdeVirtualTouch: def __init__(self): self.touch = TOUCH( TOUCH.DEV_IDE, range_x=DISPLAY_WIDTH, range_y=DISPLAY_HEIGHT, ) self.indev = lv.indev_create() self.indev.set_type(lv.INDEV_TYPE.POINTER) self.indev.set_read_cb(self.read) def read(self, driver, data): try: points = self.touch.read(1) if points: point = points[0] data.point.x = point.x data.point.y = point.y if point.event in (TOUCH.EVENT_DOWN, TOUCH.EVENT_MOVE): data.state = lv.INDEV_STATE.PRESSED else: data.state = lv.INDEV_STATE.RELEASED else: data.state = lv.INDEV_STATE.RELEASED except Exception: data.state = lv.INDEV_STATE.RELEASED def deinit(self): try: self.touch.deinit() except Exception: pass def lvgl_flush_cb(disp_drv, area, color): if disp_drv.flush_is_last(): ptr = uctypes.addressof(color.__dereference__()) frame = disp_images[0] if frame.virtaddr() != ptr: frame = disp_images[1] display_lock.acquire() try: Display.show_image(frame, layer=Display.LAYER_OSD1) finally: display_lock.release() disp_drv.flush_ready() def lvgl_setup(): global disp_images lv.init() disp_images = [ image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.BGRA8888), image.Image(DISPLAY_WIDTH, DISPLAY_HEIGHT, image.BGRA8888), ] disp_images[0].clear() disp_images[1].clear() disp_drv = lv.disp_create(DISPLAY_WIDTH, DISPLAY_HEIGHT) disp_drv.set_color_format(lv.COLOR_FORMAT.ARGB8888) disp_drv.set_draw_buffers( disp_images[0].bytearray(), disp_images[1].bytearray(), disp_images[0].size(), lv.DISP_RENDER_MODE.FULL, ) disp_drv.set_flush_cb(lvgl_flush_cb) def set_state(state, text): global app_state, status_text, ui_dirty app_state = state status_text = text ui_dirty = True def set_hidden(obj, hidden): if hidden: obj.add_flag(lv.obj.FLAG.HIDDEN) else: obj.clear_flag(lv.obj.FLAG.HIDDEN) def refresh_ui(): global ui_dirty if not ui_dirty or status_label is None: return ui_dirty = False status_label.set_text(status_text) set_hidden(resolution_label, app_state == STATE_FROZEN) config_enabled = app_state in (STATE_CONFIG, STATE_ERROR) if config_enabled: width_input.clear_state(lv.STATE.DISABLED) height_input.clear_state(lv.STATE.DISABLED) else: width_input.add_state(lv.STATE.DISABLED) height_input.add_state(lv.STATE.DISABLED) if app_state in (STATE_CONFIG, STATE_ERROR): main_button_label.set_text("START PREVIEW") main_button.clear_state(lv.STATE.DISABLED) set_hidden(main_button, False) set_hidden(save_button, True) set_hidden(discard_button, True) elif app_state == STATE_LIVE: main_button_label.set_text("CAPTURE") main_button.clear_state(lv.STATE.DISABLED) set_hidden(main_button, False) set_hidden(save_button, True) set_hidden(discard_button, True) elif app_state == STATE_FROZEN: set_hidden(main_button, True) set_hidden(save_button, False) set_hidden(discard_button, False) else: main_button.add_state(lv.STATE.DISABLED) set_hidden(main_button, False) set_hidden(save_button, True) set_hidden(discard_button, True) def input_focused(event): if app_state not in (STATE_CONFIG, STATE_ERROR): return keyboard.set_textarea(event.get_target()) keyboard.clear_flag(lv.obj.FLAG.HIDDEN) def input_ready(event): keyboard.add_flag(lv.obj.FLAG.HIDDEN) def parse_resolution(): width_text = width_input.get_text().strip() height_text = height_input.get_text().strip() if not width_text or not height_text: raise ValueError("Width and height are required") width = int(width_text) height = int(height_text) if width < 64 or height < 64: raise ValueError("Minimum output size is 64 x 64") if width % 16: raise ValueError("Width must be a multiple of 16") return width, height def main_button_clicked(event): global worker_started, worker_done, media_command keyboard.add_flag(lv.obj.FLAG.HIDDEN) if app_state in (STATE_CONFIG, STATE_ERROR): if worker_started: return try: width, height = parse_resolution() except Exception as err: set_state(STATE_ERROR, "Invalid resolution: %s" % err) return worker_started = True worker_done = False set_state(STATE_STARTING, "Starting sensor at %d x %d..." % (width, height)) _thread.start_new_thread(media_worker, (width, height)) elif app_state == STATE_LIVE: # Publish state before the command so the worker cannot consume the # command while still observing STATE_LIVE. set_state(STATE_CAPTURING, "Capturing one frame...") media_command = CMD_CAPTURE def save_clicked(event): global media_command if app_state == STATE_FROZEN: set_state(STATE_SAVING, "Saving JPEG...") media_command = CMD_SAVE def discard_clicked(event): global media_command if app_state == STATE_FROZEN: set_state(STATE_STARTING, "Discarding and resuming preview...") media_command = CMD_DISCARD def create_button(parent, text, x, color, callback): button = lv.btn(parent) button.set_size(140, 42) button.set_pos(x, 28) button.set_style_bg_color(lv.color_hex(color), 0) button.set_style_radius(8, 0) button.add_event(callback, lv.EVENT.CLICKED, None) label = lv.label(button) label.set_text(text) label.center() return button, label def make_input(parent, x, title_text, default_text): title = lv.label(parent) title.set_text(title_text) title.set_style_text_color(lv.color_hex(0xDCE6F2), 0) title.set_pos(x, 38) field = lv.textarea(parent) field.set_one_line(True) field.set_accepted_chars("0123456789") field.set_max_length(4) field.set_text(default_text) field.set_size(115, 48) field.set_pos(x + 58, 22) field.set_style_bg_color(lv.color_hex(0x18212B), 0) field.set_style_text_color(lv.color_hex(0xFFFFFF), 0) field.set_style_border_color(lv.color_hex(0x3C91E6), 0) field.set_style_border_width(2, 0) field.set_style_radius(8, 0) field.add_event(input_focused, lv.EVENT.FOCUSED, None) field.add_event(input_ready, lv.EVENT.READY, None) return field def create_ui(): global width_input, height_input, keyboard global main_button, main_button_label, save_button, discard_button global status_label, resolution_label screen = lv.scr_act() screen.set_style_bg_opa(lv.OPA.TRANSP, 0) toolbar = lv.obj(screen) toolbar.set_size(DISPLAY_WIDTH, 96) toolbar.set_pos(0, 0) toolbar.set_style_bg_color(lv.color_hex(0x101820), 0) toolbar.set_style_bg_opa(lv.OPA._80, 0) toolbar.set_style_border_width(0, 0) toolbar.set_style_radius(0, 0) width_input = make_input(toolbar, 15, "Width", str(DEFAULT_WIDTH)) height_input = make_input(toolbar, 205, "Height", str(DEFAULT_HEIGHT)) main_button, main_button_label = create_button( toolbar, "START PREVIEW", 410, 0x2878D0, main_button_clicked ) save_button, _ = create_button(toolbar, "SAVE", 450, 0x239B56, save_clicked) discard_button, _ = create_button(toolbar, "DISCARD", 610, 0xB23A48, discard_clicked) resolution_label = lv.label(toolbar) resolution_label.set_text("/data/images") resolution_label.set_style_text_color(lv.color_hex(0xAFC4D8), 0) resolution_label.set_pos(625, 38) status_panel = lv.obj(screen) status_panel.set_size(DISPLAY_WIDTH, 48) status_panel.align(lv.ALIGN.BOTTOM_MID, 0, 0) status_panel.set_style_bg_color(lv.color_hex(0x101820), 0) status_panel.set_style_bg_opa(lv.OPA._80, 0) status_panel.set_style_border_width(0, 0) status_panel.set_style_radius(0, 0) status_label = lv.label(status_panel) status_label.set_width(DISPLAY_WIDTH - 30) status_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0) status_label.set_style_text_color(lv.color_hex(0xD8F6FF), 0) status_label.center() keyboard = lv.keyboard(screen) keyboard.set_mode(lv.keyboard.MODE.NUMBER) keyboard.set_size(DISPLAY_WIDTH, 150) keyboard.align(lv.ALIGN.BOTTOM_MID, 0, 0) keyboard.add_flag(lv.obj.FLAG.HIDDEN) refresh_ui() def media_worker(width, height): global worker_started, worker_done, media_command, capture_index sensor = None current_frame = None try: sensor = Sensor(id=SENSOR_ID) sensor.reset() sensor.set_framesize( width=width, height=height, chn=CAPTURE_CHANNEL, ) sensor.set_pixformat(Sensor.RGB888, chn=CAPTURE_CHANNEL) sensor.run() set_state(STATE_LIVE, "Live preview - press CAPTURE for %d x %d" % (width, height)) while app_running: command = media_command if command == CMD_CAPTURE: media_command = CMD_NONE if current_frame is None: set_state(STATE_LIVE, "Waiting for first camera frame") else: # Do not snapshot again. OSD0 keeps showing current_frame. set_state(STATE_FROZEN, "Frame frozen - choose SAVE or DISCARD") continue if command in (CMD_SAVE, CMD_DISCARD): media_command = CMD_NONE if current_frame is None: set_state(STATE_LIVE, "No frame available - preview resumed") continue saved_path = None if command == CMD_SAVE: filename = "capture_%06d_%dx%d.jpg" % ( capture_index, width, height, ) saved_path = IMAGE_DIR + "/" + filename jpeg = current_frame.to_jpeg() jpeg.save(saved_path, quality=JPEG_QUALITY) del jpeg capture_index += 1 print("saved:", saved_path) if saved_path: set_state(STATE_LIVE, "Saved %s - preview resumed" % saved_path) else: set_state(STATE_LIVE, "Frame discarded - preview resumed") continue if app_state == STATE_LIVE: frame = sensor.snapshot(chn=CAPTURE_CHANNEL) if width == DISPLAY_WIDTH and height == DISPLAY_HEIGHT: preview_frame = frame else: # Capture keeps its requested resolution; only the IDE # virtual preview is fitted to the 800x480 UI canvas. preview_frame = frame.to_rgb888( x_size=DISPLAY_WIDTH, y_size=DISPLAY_HEIGHT, ) display_lock.acquire() try: Display.show_image(preview_frame, layer=Display.LAYER_OSD0) finally: display_lock.release() current_frame = frame time.sleep_ms(5) else: time.sleep_ms(10) except BaseException as err: print_error(err) set_state(STATE_ERROR, "Media error: %s" % err) finally: current_frame = None display_lock.acquire() try: try: Display.disable_layer(Display.LAYER_OSD0) except Exception: pass finally: display_lock.release() if sensor is not None: try: sensor.stop() except Exception as err: print("sensor cleanup:", err) gc.collect() worker_started = False worker_done = True def main(): global app_running, capture_index touch = None os.exitpoint(os.EXITPOINT_ENABLE) ensure_directory(IMAGE_DIR) capture_index = find_next_index() Display.init( DISPLAY_TYPE, width=DISPLAY_WIDTH, height=DISPLAY_HEIGHT, fps=60, to_ide=True, osd_num=2, ) try: lvgl_setup() touch = IdeVirtualTouch() create_ui() print("Set resolution, then press START PREVIEW in the IDE Preview.") while True: os.exitpoint() refresh_ui() delay = lv.task_handler() if delay is None or delay < 5: delay = 5 time.sleep_ms(delay) except KeyboardInterrupt: print("user stop") except BaseException as err: print_error(err) finally: app_running = False for _ in range(300): if worker_done: break time.sleep_ms(10) if touch: touch.deinit() lv.deinit() Display.deinit() gc.collect() os.exitpoint(os.EXITPOINT_ENABLE_SLEEP) time.sleep_ms(100) if __name__ == "__main__": main()