Merge pull request #991 from kdmukai/image_entropy_2_preview_pool

[enhancement] Image Entropy: Require a full pool of distinct preview frames before capture

src/seedsigner/gui/screens/tools_screens.py

@@ -1,9 +1,10 @@
+import hashlib
 import time
 
 from dataclasses import dataclass
 from gettext import gettext as _
 from typing import Any
-from PIL.Image import Image
+from PIL import Image, ImageDraw
 from seedsigner.gui.renderer import Renderer
 from seedsigner.hardware.camera import Camera
 from seedsigner.gui.components import FontAwesomeIconConstants, Fonts, GUIConstants, IconTextLine, SeedSignerIconConstants, TextArea
@@ -17,6 +18,10 @@ from seedsigner.gui.keyboard import Keyboard
 
 @dataclass
 class ToolsImageEntropyLivePreviewScreen(BaseScreen):
+    # Set how many distinct, non-blank preview frames must be collected into the preview
+    # pool before the final image can be taken.
+    PREVIEW_POOL_SIZE = 50
+
     def __post_init__(self):
         super().__post_init__()
 
@@ -32,13 +37,23 @@ class ToolsImageEntropyLivePreviewScreen(BaseScreen):
     def _run(self):
         # save preview image frames to use as additional entropy below
         preview_images = []
-        max_entropy_frames = 50
         instructions_font = Fonts.get_font(GUIConstants.get_body_font_name(), GUIConstants.get_button_font_size())
 
+        # Pre-calculate how wide the frame counter display can be.
+        # TRANSLATOR_NOTE: Counts frames collected so far vs the total required (e.g. 12/50)
+        max_counter_text = _("{}/{}").format(self.PREVIEW_POOL_SIZE, self.PREVIEW_POOL_SIZE)
+        (left, top, right, bottom) = instructions_font.getbbox(max_counter_text)
+        counter_text_width = right - left
+
+        # The camera hands us its most recent frame on every loop pass, but the SAME frame
+        # can arrive more than once. Store each image's sha256 hash so we can recognize
+        # and skip the repeats. The set is intentionally never pruned. A frame identical
+        # to any previously admitted frame can never be added a second time.
+        preview_frame_hashes = set()
+
         # If the user continues holding the button that brought them into this flow, we
-        # have to ensure that it doesn't trigger the ANYCLICK check below, otherwise they
+        # have to ensure that it doesn't trigger the ANYCLICK check below, otherwise the
-        # will inadvertently skip all the preview frames and take the final image
+        # final image capture would fire on its own the instant the preview pool fills.
-        # immediately.
         is_maybe_still_holding = True
 
         while True:
@@ -49,7 +64,7 @@ class ToolsImageEntropyLivePreviewScreen(BaseScreen):
                 self.camera.stop_video_stream_mode()
                 return RET_CODE__BACK_BUTTON
 
-            frame: Image = self.camera.read_video_stream(as_image=True)
+            frame: Image.Image = self.camera.read_video_stream(as_image=True)
 
             if frame is None:
                 # Camera probably isn't ready yet
@@ -81,68 +96,160 @@ class ToolsImageEntropyLivePreviewScreen(BaseScreen):
 
                 self.renderer.canvas.paste(frame.crop(box=box))
 
-            # If the ANYCLICK buttons are detected as being all released (none of them
+            # Decide whether this frame can be added to the preview pool.
-            # cause check_for_low to return True), we can be sure that the user isn't
+            # Rule 1: the frame must not be a single flat color (e.g. an all-black frame
-            # still holding down the initial button press that brought them into this
+            # or an overexposed all-white one). getextrema() reports the lowest and
-            # flow. It is then safe to arm the loop to trigger the final image capture for
+            # highest value found in each color channel; if the lowest equals the highest
-            # whenever the *next* ANYCLICK button is pressed.
+            # in every channel, every pixel in the frame is identical and the frame is
-            if not self.hw_inputs.check_for_low(keys=HardwareButtonsConstants.KEYS__ANYCLICK):
+            # rejected.
-                # Confirmed that all ANYCLICK buttons are released. The next click can now
+            frame_has_variation = False
-                # trigger the final image capture.
+            for lowest_value, highest_value in frame.getextrema():
-                is_maybe_still_holding = False
+                if lowest_value != highest_value:
+                    frame_has_variation = True
 
-            elif not is_maybe_still_holding:
+            if frame_has_variation:
-                # We passed the above check; this is a fresh, explicit click. We can now
+                # Rule 2: the frame must be one we have never counted before
-                # capture the final image and exit this loop.
+                frame_hash = hashlib.sha256(frame.tobytes()).digest()
+                if frame_hash not in preview_frame_hashes:
+                    preview_frame_hashes.add(frame_hash)
+                    if len(preview_images) == self.PREVIEW_POOL_SIZE:
+                        # The preview pool is full. Dump the oldest and add the current
+                        # frame.
+                        preview_images.pop(0)
+                    preview_images.append(frame)
 
-                # Have to manually update last input time since we're not in a wait_for loop
+            # Can only proceed to the final image when the preview pool is full
-                self.hw_inputs.update_last_input_time()
+            if len(preview_images) == self.PREVIEW_POOL_SIZE:
-                self.camera.stop_video_stream_mode()
+                # If the ANYCLICK buttons are detected as being all released (none of
+                # them cause check_for_low to return True), we can be sure that the user
+                # isn't still holding down the initial button press that brought them
+                # into this flow. It is then safe to arm the loop to trigger the final
+                # image capture for whenever the *next* ANYCLICK button is pressed.
+                if not self.hw_inputs.check_for_low(keys=HardwareButtonsConstants.KEYS__ANYCLICK):
+                    # Confirmed that all ANYCLICK buttons are released. The next click
+                    # can now trigger the final image capture.
+                    is_maybe_still_holding = False
 
-                with self.renderer.lock:
+                elif not is_maybe_still_holding:
+                    # We passed the above check; this is a fresh, explicit click. We can
+                    # now capture the final image and exit this loop.
+
+                    # Have to manually update last input time since we're not in a wait_for loop
+                    self.hw_inputs.update_last_input_time()
+                    self.camera.stop_video_stream_mode()
+
+                    with self.renderer.lock:
+                        self.renderer.draw.text(
+                            xy=(
+                                int(self.renderer.canvas_width/2),
+                                self.renderer.canvas_height - GUIConstants.EDGE_PADDING
+                            ),
+                            text=_("Capturing image..."),
+                            fill=GUIConstants.ACCENT_COLOR,
+                            font=instructions_font,
+                            stroke_width=4,
+                            stroke_fill=GUIConstants.BACKGROUND_COLOR,
+                            anchor="ms"
+                        )
+                        self.renderer.show_image()
+
+                    return preview_images
+
+            # If we're still here, it's just another preview frame loop
+            with self.renderer.lock:
+                if len(preview_images) == self.PREVIEW_POOL_SIZE and not is_maybe_still_holding:
                     self.renderer.draw.text(
                         xy=(
                             int(self.renderer.canvas_width/2),
                             self.renderer.canvas_height - GUIConstants.EDGE_PADDING
                         ),
-                        text=_("Capturing image..."),
+                        text="< " + _("back") + "  |  " + _("click a button"),  # TODO: Render with UI elements instead of text
-                        fill=GUIConstants.ACCENT_COLOR,
+                        fill=GUIConstants.BODY_FONT_COLOR,
                         font=instructions_font,
                         stroke_width=4,
                         stroke_fill=GUIConstants.BACKGROUND_COLOR,
                         anchor="ms"
                     )
-                    self.renderer.show_image()
 
-                return preview_images
+                else:
+                    # Still collecting (or is_maybe_still_holding); report current
+                    # progress on number of preview pool frames.
+
+                    # TRANSLATOR_NOTE: Shown while the camera gathers the image frames a new seed requires
+                    collecting_text = _("Collecting entropy frames")
+                    self.renderer.draw.text(
+                        xy=(
+                            int(self.renderer.canvas_width/2),
+                            self.renderer.canvas_height - GUIConstants.EDGE_PADDING - GUIConstants.BUTTON_HEIGHT - GUIConstants.COMPONENT_PADDING
+                        ),
+                        text=collecting_text,
+                        fill=GUIConstants.BODY_FONT_COLOR,
+                        font=instructions_font,
+                        stroke_width=4,
+                        stroke_fill=GUIConstants.BACKGROUND_COLOR,
+                        anchor="ms"
+                    )
+
+                    # Render the frame counter progress bar; same visual design as the
+                    # animated QR scan progress bar.
+                    rectangle = Image.new('RGBA', (self.renderer.canvas_width - 2*GUIConstants.EDGE_PADDING, GUIConstants.BUTTON_HEIGHT), (0, 0, 0, 0))
+                    draw = ImageDraw.Draw(rectangle)
+
+                    # Start with a background rounded rectangle, same dims as the buttons
+                    overlay_color = (0, 0, 0, 191)  # opacity ranges from 0-255
+                    draw.rounded_rectangle(
+                        (
+                            (0, 0),
+                            (rectangle.width, rectangle.height)
+                        ),
+                        fill=overlay_color,
+                        radius=8,
+                        outline=overlay_color,
+                        width=2,
+                    )
+
+                    progress_bar_thickness = 4
+                    progress_bar_width = rectangle.width - 2*GUIConstants.EDGE_PADDING - counter_text_width - int(GUIConstants.EDGE_PADDING/2)
+                    progress_bar_xy = (
+                            (GUIConstants.EDGE_PADDING, int((rectangle.height - progress_bar_thickness) / 2)),
+                            (GUIConstants.EDGE_PADDING + progress_bar_width, int(rectangle.height + progress_bar_thickness) / 2)
+                        )
+                    draw.rounded_rectangle(
+                        progress_bar_xy,
+                        fill=GUIConstants.INACTIVE_COLOR,
+                        radius=8
+                    )
+
+                    if len(preview_images) > 0:
+                        draw.rounded_rectangle(
+                            (
+                                progress_bar_xy[0],
+                                (GUIConstants.EDGE_PADDING + int(len(preview_images) * progress_bar_width / self.PREVIEW_POOL_SIZE), progress_bar_xy[1][1])
+                            ),
+                            fill=GUIConstants.GREEN_INDICATOR_COLOR,
+                            radius=8
+                        )
+
+                    # TRANSLATOR_NOTE: Counts frames collected so far vs the total required (e.g. 12/50)
+                    counter_text = _("{}/{}").format(len(preview_images), self.PREVIEW_POOL_SIZE)
+
+                    draw.text(
+                        xy=(rectangle.width - GUIConstants.EDGE_PADDING, int(rectangle.height / 2)),
+                        text=counter_text,
+                        fill=GUIConstants.BODY_FONT_COLOR,
+                        font=instructions_font,
+                        anchor="rm",  # right-justified, middle
+                    )
+
+                    self.renderer.canvas.paste(rectangle, (GUIConstants.EDGE_PADDING, self.renderer.canvas_height - GUIConstants.EDGE_PADDING - rectangle.height), rectangle)
 
-            # If we're still here, it's just another preview frame loop
-            with self.renderer.lock:
-                self.renderer.draw.text(
-                    xy=(
-                        int(self.renderer.canvas_width/2),
-                        self.renderer.canvas_height - GUIConstants.EDGE_PADDING
-                    ),
-                    text="< " + _("back") + "  |  " + _("click a button"),  # TODO: Render with UI elements instead of text
-                    fill=GUIConstants.BODY_FONT_COLOR,
-                    font=instructions_font,
-                    stroke_width=4,
-                    stroke_fill=GUIConstants.BACKGROUND_COLOR,
-                    anchor="ms"
-                )
                 self.renderer.show_image()
 
-            if len(preview_images) == max_entropy_frames:
-                # Keep a moving window of the last n preview frames; pop the oldest
-                # before we add the currest frame.
-                preview_images.pop(0)
-            preview_images.append(frame)
-
 
 
 @dataclass
 class ToolsImageEntropyFinalImageScreen(BaseScreen):
-    final_image: Image = None
+    final_image: Image.Image = None
 
     def _run(self):
         instructions_font = Fonts.get_font(GUIConstants.get_body_font_name(), GUIConstants.get_button_font_size())

src/seedsigner/views/tools_views.py

@@ -60,6 +60,26 @@ class ToolsMenuView(View):
     Image entropy Views
 ****************************************************************************"""
 class ToolsImageEntropyLivePreviewView(View):
+    """
+    A fixed number of live preview frames are collected into a frame pool to provide an
+    additional source of entropy. These frames provide VOLUME for the final seed but are
+    not themselves individually assessed for entropy QUALITY. The quality heuristics are
+    only applied to the final image capture.
+
+    The preview pool enforces two rules:
+    1.) A frame that is a single flat color is rejected (e.g. completely black, completely
+    white, all just one shade of green).
+
+    2.) A frame identical to any previously admitted frame is rejected (de-duplicated via
+    sha256).
+
+    The final image cannot be taken until the full pool has arrived.
+
+    This mirrors the role that NIST SP 800-90B (sec. 4.2) assigns to continuous health
+    tests: detect gross noise-source failure -- a sensor stuck on one value, a stalled
+    or repeating camera -- without attempting to measure entropy.
+    """
+
     def run(self):
         from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
         self.controller.image_entropy_preview_frames = None
@@ -67,7 +87,14 @@ class ToolsImageEntropyLivePreviewView(View):
 
         if ret == RET_CODE__BACK_BUTTON:
             return Destination(BackStackView)
-        
+
+        # The live preview screen must return the required number of preview pool frames.
+        # Do not proceed if there is any mismatch.
+        if ret is None or len(ret) != ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE:
+            num_frames = 0 if ret is None else len(ret)
+            # TRANSLATOR_NOTE: Shown when the camera fails to collect enough frames for a new seed. "expected" and "actual" are the number of frames.
+            raise Exception(_("Entropy collection failed. Expected {expected} preview frames, got {actual}").format(expected=ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE, actual=num_frames))
+
         self.controller.image_entropy_preview_frames = ret
         return Destination(ToolsImageEntropyFinalImageView)
 

tests/test_flows_tools.py

@@ -279,3 +279,64 @@ class TestToolsFlows(FlowTest):
                 FlowStep(seed_views.SeedAddressVerificationView),
                 FlowStep(seed_views.SeedAddressVerificationSuccessView),
             ])
+
+
+class TestToolsImageEntropyFlows(FlowTest):
+
+    def test__image_entropy__incorrect_preview_frame_count_aborts(self):
+        """
+        If the live preview screen returns anything other than the required number of
+        entropy frames, the View must raise rather than continue on to seed creation.
+        """
+        from unittest.mock import Mock
+        from seedsigner.views.view import UnhandledExceptionView
+        from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
+
+        # Empty list (no frames)
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=[]),
+            FlowStep(UnhandledExceptionView),
+        ])
+
+        # Too few
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=[Mock()] * 10),
+            FlowStep(UnhandledExceptionView),
+        ])
+
+        # Too many
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=[Mock()] * (ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE + 5)),
+            FlowStep(UnhandledExceptionView),
+        ])
+
+        # Degenerate None
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=None),
+            FlowStep(UnhandledExceptionView),
+        ])
+
+
+
+    def test__image_entropy__screen_exception_does_not_advance(self):
+        """
+        There is no explicit handling for the live preview screen raising, but the flow
+        must never continue on to seed creation when it does.
+        """
+        from seedsigner.views.view import UnhandledExceptionView
+
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=Exception("Test exception")),
+            FlowStep(UnhandledExceptionView),
+        ])
+
+
+    def test__image_entropy__full_preview_frames_advances(self):
+        """ A full set of preview frames advances to the final image capture. """
+        from unittest.mock import Mock
+        from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
+
+        self.run_sequence([
+            FlowStep(tools_views.ToolsImageEntropyLivePreviewView, screen_return_value=[Mock()] * ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE),
+            FlowStep(tools_views.ToolsImageEntropyFinalImageView),
+        ])

tests/test_tools_screens.py

@@ -1,3 +1,4 @@
+import hashlib
 import os
 
 from unittest.mock import MagicMock, patch
@@ -30,6 +31,24 @@ def make_noise_frame(width: int = 240, height: int = 240) -> Image.Image:
     return Image.frombytes("RGBA", (width, height), os.urandom(width * height * 4))
 
 
+def make_blank_frame(color: tuple = (0, 0, 0, 255), width: int = 240, height: int = 240) -> Image.Image:
+    """ A frame of a single flat color: every pixel identical, as from a covered camera. """
+    return Image.new("RGBA", (width, height), color)
+
+
+# A blank frame is any frame of one uniform color: a covered lens reads full black, an
+# overexposed one reads full white, and a color cast reads as some other single color.
+BLANK_FRAME_COLORS = [
+    (0, 0, 0, 255),        # full black
+    (255, 255, 255, 255),  # full white
+    (128, 128, 128, 255),  # a flat mid grey
+    (255, 0, 0, 255),      # one saturated channel is still a single flat color
+    (0, 255, 0, 255),
+    (0, 0, 255, 255),
+    # ...and so is any color whose three channels each hold their own unchanging value
+] + [(value, (value * 7) % 256, (value * 13) % 256, 255) for value in range(1, 55)]
+
+
 def make_mock_camera(frames: list) -> MagicMock:
     """
     Returns a mocked Camera whose read_video_stream() plays back the given frames in
@@ -70,6 +89,14 @@ def make_mock_hw_inputs(left_script: list = None, anyclick_script: list = None)
     return hw_inputs
 
 
+def count_anyclick_checks(mock_hw_inputs: MagicMock) -> int:
+    """
+    How many times the screen polled the ANYCLICK group.
+    """
+    from seedsigner.hardware.buttons import HardwareButtonsConstants
+    return sum(1 for c in mock_hw_inputs.check_for_low.call_args_list if c.kwargs.get("keys") == HardwareButtonsConstants.KEYS__ANYCLICK)
+
+
 
 class ImageEntropyScreenTestBase(BaseTest):
 
@@ -102,42 +129,102 @@ class TestToolsImageEntropyLivePreviewScreen(ImageEntropyScreenTestBase):
         return screen
 
 
-    def test_button_held_from_start_never_captures_until_released(self):
+    def test_screen_returns_exactly_50_unique_frames(self):
         """
-        A button already held down when the screen starts must not trigger the final
+        The screen should return 50 unique frames from its preview pool.
-        image capture; only a fresh press after all buttons have been seen released
-        may capture.
         """
-        frames = [make_noise_frame() for i in range(5)]
+        from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
-        mock_camera = make_mock_camera(frames)
+        num_required = ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE
 
-        # Button is held for the first two loop passes, released on the third, then
+        frames = [make_noise_frame() for i in range(num_required)]
-        # pressed again on the fourth.
-        mock_hw_inputs = make_mock_hw_inputs(anyclick_script=[True, True, False, True])
-        screen = self.build_screen(mock_camera, mock_hw_inputs)
 
-        # The screen returns the live preview frames
+        # One extra read happens on the capture pass; feed it a duplicate of the last
-        result = screen._run()
+        # frame, which the dedupe check rejects (so the returned pool is untouched).
-
+        duplicate = frames[-1].copy()
-        # The held presses were ignored; the fresh press on the fourth pass captured.
+        mock_camera = make_mock_camera(frames + [duplicate])
-        # Three frames were collected before the capture pass.
-        assert result == frames[:3]
-        mock_camera.stop_video_stream_mode.assert_called_once()
-
-
-    def test_click_after_release_captures(self):
-        """ Normal use: no buttons pressed at first, then a click captures. """
-        frames = [make_noise_frame() for i in range(2)]
-        mock_camera = make_mock_camera(frames)
         mock_hw_inputs = make_mock_hw_inputs(anyclick_script=[False, True])
         screen = self.build_screen(mock_camera, mock_hw_inputs)
 
         result = screen._run()
 
-        assert result == frames[:1]
+        assert result == frames
         mock_camera.stop_video_stream_mode.assert_called_once()
 
 
+    def test_button_held_from_start_never_captures_until_released(self):
+        """
+        A button already held down when the screen starts must not trigger the final
+        image capture -- not even after the entropy pool has filled. Only a fresh
+        press after all buttons have been seen released may capture.
+        """
+        from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
+        num_required = ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE
+
+        # The pool fills at frame 50; the held button registers as still held for the next
+        # two loop iterations, released on the third (that's the go-ahead to arm the watch
+        # for the final click), then finally pressed on the fourth.
+        frames = [make_noise_frame() for i in range(num_required + 3)]
+        mock_camera = make_mock_camera(frames)
+        mock_hw_inputs = make_mock_hw_inputs(anyclick_script=[True, True, False, True])
+        screen = self.build_screen(mock_camera, mock_hw_inputs)
+
+        result = screen._run()
+
+        # The pool is a moving window of the LAST 50 admitted frames.
+        assert result == frames[3:]
+        assert len(result) == num_required
+
+        # The ANYCLICK checks should only start after the preview pool is filled so we
+        # should only see the four checks scripted above.
+        assert count_anyclick_checks(mock_hw_inputs) == 4
+        mock_camera.stop_video_stream_mode.assert_called_once()
+
+
+    def test_blank_frames_are_never_admitted(self):
+        """
+        Flat single-color frames (e.g. covered camera, all white, all turquoise, etc) are
+        never counted, the final capture click is never armed, and the back button still
+        exits.
+        """
+        # One frame per flat color, more of them than the pool needs, and all distinct.
+        blanks = [make_blank_frame(color) for color in BLANK_FRAME_COLORS]
+        mock_camera = make_mock_camera(blanks)
+        mock_hw_inputs = make_mock_hw_inputs(left_script=[False] * (len(blanks) - 1) + [True])
+        screen = self.build_screen(mock_camera, mock_hw_inputs)
+
+        result = screen._run()
+
+        assert result == RET_CODE__BACK_BUTTON
+
+        # The pool never filled, so the final capture click (ANYCLICK) was never checked.
+        assert count_anyclick_checks(mock_hw_inputs) == 0
+        mock_camera.stop_video_stream_mode.assert_called_once()
+
+
+    def test_duplicate_frames_are_only_counted_once(self):
+        """
+        A frame whose bytes exactly match an already-admitted frame cannot be included in
+        the preview pool again.
+        """
+        from seedsigner.gui.screens.tools_screens import ToolsImageEntropyLivePreviewScreen
+        num_required = ToolsImageEntropyLivePreviewScreen.PREVIEW_POOL_SIZE
+
+        test_frame = make_noise_frame()
+        fillers = [make_noise_frame() for i in range(num_required)]
+
+        feed = fillers[:10] + [test_frame] + fillers[10:20] + [test_frame] + fillers[20:]
+        mock_camera = make_mock_camera(feed)
+        mock_hw_inputs = make_mock_hw_inputs(anyclick_script=[False, True])
+        screen = self.build_screen(mock_camera, mock_hw_inputs)
+
+        result = screen._run()
+
+        assert test_frame in result
+        # PIL Images are unhashable, so uniqueness is checked on their bytes
+        assert len(set(frame.tobytes() for frame in result)) == num_required  # all unique
+
+
+
     def test_back_button_exits_immediately(self):
         """ KEY_LEFT backs out at any time, even before any frame is read. """
         mock_camera = make_mock_camera([])