MicroPythonOS/internal_filesystem/apps/com.micropythonos.breakout/assets/breakout.py

# 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