LEDMatrix/test/test_offscreen_rendering.py

"""Per-thread off-screen rendering (DisplayManager.offscreen) and its Vegas use.

The display canvas used to be one shared object, so any plugin that drew on it
could only be rendered on the render thread, stalling the scroll for 40-600ms
each. offscreen() gives the calling thread a canvas of its own. These tests pin
the property that makes that safe: another thread's drawing never reaches what
the render loop sees, presents or paces by.

Runs against RGBMatrixEmulator, exercising the real DisplayManager.
"""

import os
import sys
import threading
from types import SimpleNamespace

os.environ["EMULATOR"] = "true"

import pytest
from PIL import Image

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from src.vegas_mode.config import VegasModeConfig  # noqa: E402
from src.vegas_mode.plugin_adapter import PluginAdapter  # noqa: E402
from src.vegas_mode.stream_manager import StreamManager  # noqa: E402

WIDTH, HEIGHT = 128, 32
YELLOW = (255, 255, 0)


@pytest.fixture(scope="module")
def dm(tmp_path_factory):
    from src.display_manager import DisplayManager
    DisplayManager._instance = None
    DisplayManager._initialized = False
    manager = DisplayManager({
        "display": {
            "hardware": {"rows": 32, "cols": 64, "chain_length": 2,
                         "parallel": 1, "brightness": 90},
            "runtime": {"gpio_slowdown": 0},
        },
    }, suppress_test_pattern=True)
    manager._snapshot_path = str(
        tmp_path_factory.mktemp("offscreen") / "led_matrix_preview.png")
    if manager.matrix is None:
        pytest.fail("DisplayManager fell back to matrix=None; see the "
                    "'Failed to initialize RGB Matrix' log line above.")
    yield manager
    DisplayManager._instance = None
    DisplayManager._initialized = False


@pytest.fixture
def fresh(dm):
    """A known shared canvas, and scroll state reset around each test."""
    dm.image = Image.new('RGB', (WIDTH, HEIGHT))
    from PIL import ImageDraw
    dm.draw = ImageDraw.Draw(dm.image)
    dm.set_scrolling_state(False)
    yield dm
    dm.set_scrolling_state(False)


class _Spy:
    """Counts calls to one method of one object, and still calls it."""

    def __init__(self, obj, name):
        self.obj, self.name, self.count = obj, name, 0
        self._orig = getattr(obj, name)

    def __enter__(self):
        def counting(*args, **kwargs):
            self.count += 1
            return self._orig(*args, **kwargs)
        setattr(self.obj, self.name, counting)
        return self

    def __exit__(self, *exc):
        setattr(self.obj, self.name, self._orig)


def _in_thread(fn):
    """Run fn on another thread and return its result, re-raising its error."""
    box = {}

    def run():
        try:
            box["value"] = fn()
        except BaseException as exc:  # surfaced below
            box["error"] = exc

    thread = threading.Thread(target=run)
    thread.start()
    thread.join(10)
    assert not thread.is_alive(), "worker hung"
    if "error" in box:
        raise box["error"]
    return box.get("value")


class TestIsolation:
    def test_another_threads_drawing_is_invisible_here(self, fresh):
        dm = fresh
        shared = dm.image
        inside, release = threading.Event(), threading.Event()
        seen = {}

        def worker():
            with dm.offscreen(40) as surface:
                dm.draw.rectangle([0, 0, 39, 31], fill=YELLOW)
                seen.update(width=dm.width, matrix_width=dm.matrix.width,
                            own_image=dm.image is surface.image)
                inside.set()
                release.wait(5)
            seen["restored"] = dm.image is shared

        thread = threading.Thread(target=worker)
        thread.start()
        try:
            assert inside.wait(5)
            # While the worker is mid-draw, this thread sees the real canvas.
            assert dm.image is shared
            assert (dm.width, dm.matrix.width) == (WIDTH, WIDTH)
            assert shared.getpixel((0, 0)) == (0, 0, 0)
        finally:
            release.set()
            thread.join(5)
        assert seen == {"width": 40, "matrix_width": 40, "own_image": True,
                        "restored": True}

    def test_the_render_loop_keeps_presenting_while_another_thread_draws(self, fresh):
        dm = fresh
        inside, release = threading.Event(), threading.Event()

        def worker():
            with dm.offscreen():
                dm.draw.rectangle([0, 0, 10, 10], fill=YELLOW)
                dm.update_display()          # must not reach the panel
                inside.set()
                release.wait(5)

        with _Spy(dm.matrix, "SwapOnVSync") as swaps:
            thread = threading.Thread(target=worker)
            thread.start()
            try:
                assert inside.wait(5)
                dm.draw.point((5, 5), fill=(0, 0, 255))
                dm.update_display()          # the render thread's own frame
            finally:
                release.set()
                thread.join(5)
        assert swaps.count == 1

    def test_assignments_go_to_the_callers_canvas(self, fresh):
        dm = fresh
        shared = dm.image
        replacement = Image.new('RGB', (WIDTH, HEIGHT), (1, 2, 3))

        def worker():
            with dm.offscreen():
                dm.image = replacement       # e.g. clear() inside display()
                return dm.image is replacement

        assert _in_thread(worker) is True
        assert dm.image is shared

    def test_render_size_narrows_only_the_calling_thread(self, fresh):
        dm = fresh
        inside, release = threading.Event(), threading.Event()
        seen = {}

        def worker():
            with dm.render_size(32):
                seen["width"] = dm.width
                inside.set()
                release.wait(5)

        thread = threading.Thread(target=worker)
        thread.start()
        try:
            assert inside.wait(5)
            assert dm.width == WIDTH
        finally:
            release.set()
            thread.join(5)
        assert seen["width"] == 32


class TestNothingReachesThePanel:
    def test_update_display_clear_and_pacing_are_inert_inside(self, fresh):
        dm = fresh
        real_matrix = dm.matrix
        hold_before = dm._frame_hold
        with _Spy(real_matrix, "SwapOnVSync") as swaps, \
                _Spy(dm.offscreen_canvas, "Clear") as clears:
            with dm.offscreen() as surface:
                dm.draw.rectangle([0, 0, 5, 5], fill=YELLOW)
                dm.update_display()
                drawn = dm.image
                dm.clear()
                # clear() replaced the surface's image, not the shared one.
                assert dm.image is surface.image and dm.image is not drawn
                assert dm.image.getpixel((0, 0)) == (0, 0, 0)
                dm.set_scrolling_state(True, 3)
                dm.set_frame_hold(4)
                dm.matrix.SwapOnVSync(object())       # inert through the proxy too
        assert swaps.count == 0
        assert clears.count == 0
        assert dm._frame_hold == hold_before
        assert dm._scrolling_state['is_scrolling'] is False

    def test_capture_mode_inside_offscreen_does_not_end_suppression(self, fresh):
        dm = fresh
        real_matrix = dm.matrix
        with _Spy(real_matrix, "SwapOnVSync") as swaps:
            with dm.offscreen():
                with dm.capture_mode():
                    pass
                dm.draw.point((1, 1), fill=YELLOW)
                dm.update_display()
        assert swaps.count == 0

    def test_nesting_and_exceptions_restore_state(self, fresh):
        dm = fresh
        shared = dm.image
        with pytest.raises(RuntimeError):
            with dm.offscreen(100):
                with dm.offscreen(50):
                    assert dm.width == 50
                    raise RuntimeError("plugin failed mid-draw")
        assert dm.image is shared
        assert dm.width == WIDTH
        assert not dm._writes_suppressed()

        with dm.offscreen(100) as outer:
            with dm.offscreen(50):
                pass
            assert dm.image is outer.image
            assert dm.width == 100


# --- Vegas: the adapter draws plugins off the render thread ------------------

class CapturePlugin:
    """A plugin with neither get_vegas_content nor a scroll helper: captured."""

    def __init__(self, display_manager):
        self.display_manager = display_manager
        self.calls = 0

    def display(self, force_clear=False):
        self.calls += 1
        self.display_manager.draw.rectangle([0, 0, 30, 20], fill=YELLOW)
        self.display_manager.update_display()


class ScrollingPlugin:
    """A ticker whose scroll image only exists once display() has run."""

    def __init__(self, display_manager):
        from src.common.scroll_helper import ScrollHelper
        self.display_manager = display_manager
        self.scroll_helper = ScrollHelper(WIDTH, HEIGHT)

    def display(self, force_clear=False):
        item = Image.new('RGB', (300, HEIGHT), YELLOW)
        self.scroll_helper.create_scrolling_image([item], item_gap=0, element_gap=0)
        self.display_manager.set_scrolling_state(True, 5)
        self.display_manager.update_display()


def _adapter(dm, **config):
    return PluginAdapter(dm, VegasModeConfig(**config))


def _has_yellow(images):
    return any(YELLOW in {img.getpixel((x, y)) for x in range(min(img.width, 40))
                          for y in range(img.height)} for img in images)


class TestAdapterOffTheRenderThread:
    def test_display_capture_runs_on_a_background_thread(self, fresh):
        dm = fresh
        shared = dm.image
        before = shared.tobytes()
        plugin = CapturePlugin(dm)
        adapter = _adapter(dm)

        with _Spy(dm.matrix, "SwapOnVSync") as swaps:
            images = _in_thread(
                lambda: adapter.get_content(plugin, "capture-bg", offscreen_only=True))

        assert images and _has_yellow(images)
        assert dm.image is shared and shared.tobytes() == before
        assert swaps.count == 0

    def test_scroll_content_is_generated_on_a_background_thread(self, fresh):
        dm = fresh
        plugin = ScrollingPlugin(dm)
        adapter = _adapter(dm)
        hold_before = dm._frame_hold

        images = _in_thread(
            lambda: adapter.get_content(plugin, "scroll-bg", offscreen_only=True))

        assert images and _has_yellow(images)
        # The plugin's own set_scrolling_state(True, 5) did not re-pace Vegas.
        assert dm._frame_hold == hold_before
        assert dm._scrolling_state['is_scrolling'] is False

    def test_with_the_switch_off_background_capture_is_left_for_the_render_thread(self, fresh):
        dm = fresh
        plugin = CapturePlugin(dm)
        adapter = _adapter(dm, offscreen_prefetch=False)

        images = _in_thread(
            lambda: adapter.get_content(plugin, "capture-legacy", offscreen_only=True))

        assert images is None
        assert plugin.calls == 0


class TestPluginLock:
    def test_a_background_fetch_waits_for_update_then_skips(self, fresh):
        dm = fresh
        lock = threading.Lock()
        plugin = CapturePlugin(dm)
        adapter = PluginAdapter(dm, VegasModeConfig(),
                                plugin_manager=SimpleNamespace(
                                    get_plugin_lock=lambda plugin_id: lock))
        adapter.PLUGIN_LOCK_TIMEOUT = 0.05

        with lock:   # update() in progress
            skipped = _in_thread(
                lambda: adapter.get_content(plugin, "locked", offscreen_only=True))
        assert skipped is None
        assert plugin.calls == 0

        served = _in_thread(
            lambda: adapter.get_content(plugin, "locked", offscreen_only=True))
        assert served and plugin.calls == 1
        assert not lock.locked()


class TestStreamDoesNotDeferEmptyResults:
    def _group(self, offscreen_prefetch):
        config = VegasModeConfig(offscreen_prefetch=offscreen_prefetch)
        adapter = SimpleNamespace(get_content=lambda *a, **k: None)
        stream = StreamManager(config, SimpleNamespace(plugins={"p": object()}),
                               adapter)
        stream.refresh = lambda: None
        stream._ordered_plugins = ["p"]
        return stream.take_next_group(count=1, offscreen_only=True)

    def test_nothing_to_show_is_not_sent_to_the_render_thread(self):
        assert self._group(offscreen_prefetch=True) == [("p", [])]

    def test_the_old_contract_still_defers_with_the_switch_off(self):
        assert self._group(offscreen_prefetch=False) == [("p", None)]