mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
266 lines
9.4 KiB
Python
266 lines
9.4 KiB
Python
# Breakout app UI/driver glue. This app renders into a framebuffer that may be
|
|
# smaller than the full display (partial framebuffer). The draw loop is more
|
|
# complex because it slices the screen into chunks, renders each slice in C,
|
|
# and flushes them sequentially using a flush-ready IRQ callback. A scheduled
|
|
# (non-IRQ) handler advances chunks so it can work on larger-than-320x230
|
|
# displays without requiring a full-size framebuffer.
|
|
import lvgl as lv
|
|
import mpos.ui
|
|
from mpos import Activity, DisplayMetrics, InputManager
|
|
|
|
import sys
|
|
if sys.platform == "esp32":
|
|
import breakout_xtensawin as breakout
|
|
else:
|
|
import breakout_x64 as breakout
|
|
|
|
class Breakout(Activity):
|
|
|
|
hor_res = 0
|
|
ver_res = 0
|
|
paddle_move_step = None
|
|
layer = None
|
|
buffer = None
|
|
touch_active = False
|
|
touch_last_x = None
|
|
last_fps = 0
|
|
average_fps = 0
|
|
|
|
old_callback = None
|
|
|
|
render_next = True
|
|
flush_ready = False
|
|
chunk_in_progress = False
|
|
chunk_waiting = False
|
|
chunk_rows_per = 0
|
|
chunk_total = 0
|
|
chunk_index = 0
|
|
|
|
# Widgets:
|
|
screen = None
|
|
canvas = None
|
|
leftbutton = None
|
|
playbutton = None
|
|
rightbutton = None
|
|
|
|
def onCreate(self):
|
|
self.screen = lv.obj()
|
|
self.screen.add_flag(lv.obj.FLAG.CLICKABLE)
|
|
self.screen.add_event_cb(self.touch_cb, lv.EVENT.ALL, None)
|
|
|
|
d = lv.display_get_default()
|
|
self.hor_res = d.get_horizontal_resolution()
|
|
self.paddle_move_step = round(self.hor_res/10)
|
|
self.ver_res = d.get_vertical_resolution()
|
|
|
|
self.leftbutton = lv.button(self.screen)
|
|
self.leftbutton.align(lv.ALIGN.BOTTOM_LEFT, 0, 0)
|
|
leftlabel = lv.label(self.leftbutton)
|
|
leftlabel.set_text("<")
|
|
self.leftbutton.add_event_cb(lambda e: self.move_left_unfocus(),lv.EVENT.FOCUSED,None)
|
|
self.leftbutton.add_event_cb(lambda e: self.move_left(),lv.EVENT.CLICKED,None)
|
|
|
|
# Invisible button, just for defocusing the left and right buttons:
|
|
self.play_button = lv.button(self.screen)
|
|
self.play_button.align(lv.ALIGN.BOTTOM_MID,0,0)
|
|
self.play_button.set_size(1,1)
|
|
self.play_button.set_style_opa(lv.OPA.TRANSP, lv.PART.MAIN)
|
|
|
|
self.rightbutton = lv.button(self.screen)
|
|
self.rightbutton.align(lv.ALIGN.BOTTOM_RIGHT, 0, 0)
|
|
rightlabel = lv.label(self.rightbutton)
|
|
rightlabel.set_text(">")
|
|
self.rightbutton.add_event_cb(lambda e: self.move_right_unfocus(),lv.EVENT.FOCUSED,None)
|
|
self.rightbutton.add_event_cb(lambda e: self.move_right(),lv.EVENT.CLICKED,None)
|
|
|
|
self.setContentView(self.screen)
|
|
|
|
def onResume(self, screen):
|
|
lv.log_register_print_cb(self.log_callback)
|
|
breakout.init(mpos.ui.main_display._frame_buffer1, self.hor_res, self.ver_res)
|
|
mpos.ui.main_display._data_bus.register_callback(self.flush_ready_cb)
|
|
mpos.ui.task_handler.add_event_cb(self.drawframe, mpos.ui.task_handler.TASK_HANDLER_FINISHED)
|
|
|
|
def onPause(self, screen):
|
|
lv.log_register_print_cb(None)
|
|
mpos.ui.main_display._data_bus.register_callback(mpos.ui.main_display._flush_ready_cb)
|
|
|
|
def move_left(self):
|
|
breakout.move_paddle(-self.paddle_move_step)
|
|
|
|
def move_right(self):
|
|
breakout.move_paddle(self.paddle_move_step)
|
|
|
|
def move_left_unfocus(self):
|
|
self.unfocus()
|
|
breakout.move_paddle(-self.paddle_move_step)
|
|
|
|
def move_right_unfocus(self):
|
|
self.unfocus()
|
|
breakout.move_paddle(self.paddle_move_step)
|
|
|
|
# This only works with the PREV/pageup and NEXT/pagedown buttons,
|
|
# because the focus_direction handling of the arrow keys uses a trick to move focus (focus_next)
|
|
# which conflicts with the focus_next below...
|
|
def unfocus(self):
|
|
focusgroup = lv.group_get_default()
|
|
if not focusgroup:
|
|
print("WARNING: imageview.py could not get default focus group")
|
|
return
|
|
focused = focusgroup.get_focused()
|
|
if focused:
|
|
if focused == self.rightbutton:
|
|
focusgroup.focus_prev()
|
|
elif focused == self.leftbutton:
|
|
focusgroup.focus_next()
|
|
else:
|
|
print("focus isn't on next or previous, leaving it...")
|
|
|
|
def flush_ready_cb(self, arg1=None, arg2=None):
|
|
# This is called in IRQ (interrupt) context so it can't allocate memory
|
|
# So no printf, no calling drawframe() directly, just setting variables or scheduling a function.
|
|
mpos.ui.main_display._disp_drv.flush_ready()
|
|
self.flush_ready = True
|
|
|
|
def drawframe(self, arg1=None, arg2=None):
|
|
if self.chunk_waiting:
|
|
if self.flush_ready:
|
|
self.flush_ready = False
|
|
self.chunk_waiting = False
|
|
self.chunk_index += 1
|
|
if self.chunk_index >= self.chunk_total:
|
|
self.chunk_in_progress = False
|
|
self.render_next = True
|
|
else:
|
|
self._render_and_send_chunk()
|
|
return
|
|
|
|
if self.chunk_in_progress or not self.render_next:
|
|
return
|
|
|
|
self.render_next = False
|
|
|
|
buffer_len = len(mpos.ui.main_display._frame_buffer1)
|
|
bytes_per_row = self.hor_res * 2
|
|
if bytes_per_row <= 0:
|
|
self.render_next = True
|
|
return
|
|
|
|
rows_per_chunk = buffer_len // bytes_per_row
|
|
if rows_per_chunk <= 0:
|
|
self.render_next = True
|
|
return
|
|
|
|
if rows_per_chunk >= self.ver_res:
|
|
self.chunk_rows_per = self.ver_res
|
|
self.chunk_index = 0
|
|
self.chunk_total = 1
|
|
else:
|
|
self.chunk_rows_per = rows_per_chunk
|
|
self.chunk_index = 0
|
|
self.chunk_total = (self.ver_res + rows_per_chunk - 1) // rows_per_chunk
|
|
|
|
self.chunk_in_progress = True
|
|
self.chunk_waiting = False
|
|
self.flush_ready = False
|
|
self._render_and_send_chunk()
|
|
|
|
def _render_and_send_chunk(self):
|
|
if not self.chunk_in_progress:
|
|
return
|
|
if self.chunk_waiting:
|
|
return
|
|
if self.chunk_index >= self.chunk_total:
|
|
self.chunk_in_progress = False
|
|
self.render_next = True
|
|
return
|
|
|
|
y_offset = self.chunk_index * self.chunk_rows_per
|
|
rows = min(self.chunk_rows_per, self.ver_res - y_offset)
|
|
advance = (self.chunk_index == 0)
|
|
is_last = (self.chunk_index + 1) == self.chunk_total
|
|
|
|
self.chunk_waiting = True
|
|
breakout.render(y_offset, rows, advance)
|
|
self.send_to_display(y_offset, rows, is_last)
|
|
|
|
def send_to_display(self, y_offset=0, rows=None, is_last=True):
|
|
x1 = 0
|
|
x2 = mpos.ui.main_display.get_horizontal_resolution() - 1
|
|
x2 = x2 + mpos.ui.main_display._offset_x
|
|
x1 = x1 + mpos.ui.main_display._offset_x
|
|
|
|
if rows is None:
|
|
rows = mpos.ui.main_display.get_vertical_resolution()
|
|
y1 = y_offset
|
|
y2 = y_offset + rows - 1
|
|
y1 = y1 + mpos.ui.main_display._offset_y
|
|
y2 = y2 + mpos.ui.main_display._offset_y
|
|
|
|
cmd = mpos.ui.main_display._set_memory_location(x1, y1, x2, y2)
|
|
bytes_needed = rows * mpos.ui.main_display.get_horizontal_resolution() * 2
|
|
data_view = memoryview(mpos.ui.main_display._frame_buffer1)[:bytes_needed]
|
|
|
|
tx_last = True
|
|
mpos.ui.main_display._data_bus.tx_color(
|
|
cmd,
|
|
data_view,
|
|
x1, y1, x2, y2,
|
|
mpos.ui.main_display._rotation,
|
|
tx_last,
|
|
)
|
|
|
|
def touch_cb(self, event):
|
|
event_code = event.get_code()
|
|
if event_code == lv.EVENT.PRESSED:
|
|
x, y = InputManager.pointer_xy()
|
|
self.touch_active = True
|
|
self.touch_last_x = x
|
|
return
|
|
|
|
if event_code == lv.EVENT.PRESSING:
|
|
if not self.touch_active:
|
|
x, y = InputManager.pointer_xy()
|
|
self.touch_active = True
|
|
self.touch_last_x = x
|
|
return
|
|
x, y = InputManager.pointer_xy()
|
|
if self.touch_last_x is not None:
|
|
delta = x - self.touch_last_x
|
|
if delta:
|
|
breakout.move_paddle(round(delta*1.3)) # amplify movement to avoid sides
|
|
self.touch_last_x = x
|
|
return
|
|
|
|
if event_code == lv.EVENT.RELEASED:
|
|
self.touch_active = False
|
|
self.touch_last_x = None
|
|
return
|
|
|
|
average_samples = 20
|
|
fps_buffer = [0.0] * average_samples
|
|
fps_index = 0
|
|
fps_sum = 0.0
|
|
fps_count = 0 # Number of valid samples (0 to average_samples)
|
|
def moving_average(self, value):
|
|
if self.fps_count == self.average_samples:
|
|
self.fps_sum -= self.fps_buffer[self.fps_index]
|
|
else:
|
|
self.fps_count += 1
|
|
self.fps_sum += value
|
|
self.fps_buffer[self.fps_index] = value
|
|
self.fps_index = (self.fps_index + 1) % self.average_samples
|
|
return self.fps_sum / self.fps_count
|
|
|
|
# Custom log callback to capture FPS
|
|
def log_callback(self, level, log_str):
|
|
log_str = log_str.decode() if isinstance(log_str, bytes) else log_str
|
|
if "sysmon:" in log_str and "FPS" in log_str:
|
|
try:
|
|
fps_part = log_str.split("FPS")[0].split("sysmon:")[1].strip()
|
|
self.last_fps = int(fps_part)
|
|
self.average_fps = self.moving_average(self.last_fps)
|
|
print(f"Current FPS: {self.last_fps} - Average FPS: {self.average_fps}")
|
|
except (IndexError, ValueError):
|
|
pass
|