unify visualization processor interactivity with the VisualizationCoo…

…rdinator, add hotpsot cross section visualization to PeakFlux.py

contrib/app/SpotAnalysis/PeakFlux.py

@@ -6,6 +6,7 @@
 from opencsp.common.lib.cv.AbstractFiducials import AbstractFiducials
 import opencsp.common.lib.cv.SpotAnalysis as sa
 from opencsp.common.lib.cv.fiducials.BcsFiducial import BcsFiducial
+from opencsp.common.lib.cv.fiducials.HotspotFiducial import HotspotFiducial
 from opencsp.common.lib.cv.spot_analysis.SpotAnalysisImagesStream import ImageType
 from opencsp.common.lib.cv.spot_analysis.SpotAnalysisOperable import SpotAnalysisOperable
 import opencsp.common.lib.cv.spot_analysis.SpotAnalysisOperableAttributeParser as saoap
@@ -66,6 +67,9 @@ def __init__(self, indir: str, outdir: str, experiment_name: str, settings_path_
             ViewCrossSectionImageProcessor(
                 self.get_bcs_origin, 'BCS', single_plot=False, crop_to_threshold=20, interactive=True
             ),
+            ViewCrossSectionImageProcessor(
+                self.get_peak_origin, 'Hotspot', single_plot=False, crop_to_threshold=20, interactive=True
+            ),
             PopulationStatisticsImageProcessor(initial_min=0, initial_max=255),
             FalseColorImageProcessor(),
             AnnotationImageProcessor(),
@@ -104,6 +108,13 @@ def get_bcs_origin(self, operable: SpotAnalysisOperable):
         origin_ix, origin_iy = int(np.round(origin_fx)), int(np.round(origin_fy))
         return origin_ix, origin_iy
 
+    def get_peak_origin(self, operable: SpotAnalysisOperable):
+        fiducials = operable.get_fiducials_by_type(HotspotFiducial)
+        fiducial = fiducials[0]
+        origin_fx, origin_fy = fiducial.origin.astuple()
+        origin_ix, origin_iy = int(np.round(origin_fx)), int(np.round(origin_fy))
+        return origin_ix, origin_iy
+
 
 if __name__ == "__main__":
     import argparse

opencsp/common/lib/cv/spot_analysis/VisualizationCoordinator.py

@@ -1,8 +1,14 @@
+import weakref
+
+import matplotlib
+import matplotlib.backend_bases
 import numpy as np
 
 from opencsp.common.lib.cv.spot_analysis.SpotAnalysisOperable import SpotAnalysisOperable
 from opencsp.common.lib.cv.spot_analysis.image_processor import *
 import opencsp.common.lib.render_control.RenderControlFigure as rcf
+import opencsp.common.lib.render_control.RenderControlFigureRecord as rcfr
+import opencsp.common.lib.tool.log_tools as lt
 
 
 class VisualizationCoordinator:
@@ -19,11 +25,56 @@ class VisualizationCoordinator:
     def __init__(self):
         self.visualization_processors: list[AbstractVisualizationImageProcessor] = []
         self.render_control_fig: rcf.RenderControlFigure = None
+        self.figures: list[weakref.ref[rcfr.RenderControlFigureRecord]] = []
+
+        # used to ensure a valid internal state
+        self.has_registered_visualization_processors = False
+
+        # user interaction
+        self.shift_down = False
+        self.enter_pressed = False
+        self.enter_shift_pressed = False
+        self.closed = False
 
     def clear(self):
         self.visualization_processors.clear()
+        self.render_control_fig = None
+        self.figures.clear()
+
+        self.has_registered_visualization_processors = False
+        self.enter_shift_pressed = False
+
+    def on_key_release(self, event: matplotlib.backend_bases.KeyEvent):
+        shift_down = self.shift_down
+        key = event.key
+
+        if "shift+" in key:
+            shift_down = True
+            key = key.replace("shift+", "")
+
+        if key == "enter" or key == "return":
+            if shift_down:
+                self.enter_shift_pressed = True
+            else:
+                self.enter_pressed = True
+        elif key == "shift":
+            self.shift_down = False
+
+    def on_key_press(self, event: matplotlib.backend_bases.KeyEvent):
+        if event.key == "shift":
+            self.shift_down = True
+
+    def on_close(self, event: matplotlib.backend_bases.CloseEvent):
+        self.closed = True
 
     def register_visualization_processors(self, all_processors: list[AbstractSpotAnalysisImagesProcessor]):
+        # this method is not safe to be called multiple times
+        if self.has_registered_visualization_processors:
+            lt.warning("Warning in VisualizationCoordinator.register_visualization_processors(): " +
+                       "attempting to register processors again without calling 'clear()' first.")
+            return
+        self.has_registered_visualization_processors = True
+
         # find and register all visualization processors
         for processor in all_processors:
             if isinstance(processor, AbstractVisualizationImageProcessor):
@@ -61,15 +112,32 @@ def register_visualization_processors(self, all_processors: list[AbstractSpotAna
 
         # initialize the visualizers
         for processor in self.visualization_processors:
-            processor._init_figure_records(self.render_control_fig)
+            processor_figures = processor._init_figure_records(self.render_control_fig)
+            for fig_record in processor_figures:
+                fig_record.figure.canvas.mpl_connect('close_event', self.on_close)
+                fig_record.figure.canvas.mpl_connect('key_release_event', self.on_key_release)
+                fig_record.figure.canvas.mpl_connect('key_press_event', self.on_key_press)
+                self.figures.append(weakref.ref(fig_record))
+
+    def _get_figures(self) -> list[rcfr.RenderControlFigureRecord]:
+        figures = [(ref, ref()) for ref in self.figures]
+        alive = [fr for ref, fr in filter(lambda ref, fr: fr is not None, figures)]
+        dead = [ref for ref, fr in filter(lambda ref, fr: fr is None, figures)]
+
+        for ref in dead:
+            self.figures.remove(ref)
+
+        return alive
 
     def is_visualize(
         self,
         visualization_processor: AbstractVisualizationImageProcessor,
         operable: SpotAnalysisOperable,
         is_last: bool,
     ) -> bool:
-        if visualization_processor == self.visualization_processors[-1]:
+        if self.closed:
+            return False
+        elif visualization_processor == self.visualization_processors[-1]:
             return True
         return False
 
@@ -81,3 +149,43 @@ def visualize(
     ):
         for processor in self.visualization_processors:
             processor._visualize_operable(operable, is_last)
+
+        # if interactive, then block until the user presses "enter" or closes one or more visualizations
+        interactive = False
+        for processor in self.visualization_processors:
+            if isinstance(processor.interactive, bool):
+                interactive |= processor.interactive
+            else:
+                interactive |= processor.interactive(operable)
+        if interactive:
+            self.enter_pressed = False
+            self.closed = False
+
+            first_iteration = True
+            while True:
+                # if shift+enter was ever pressed, that will disable interactive mode
+                if self.enter_shift_pressed:
+                    break
+
+                # wait for up to total_wait_time for the user to interact with the visualizations
+                old_raise = matplotlib.rcParams["figure.raise_window"]
+                matplotlib.rcParams["figure.raise_window"] = first_iteration
+                figures = list(filter(lambda fr: fr is not None, [ref() for ref in self.figures]))
+                total_wait_time = 0.1  # seconds
+                per_record_wait_time = total_wait_time / len(figures)
+                for fig_record in figures:
+                    if fig_record.figure.waitforbuttonpress(per_record_wait_time) is not None:
+                        break
+                matplotlib.rcParams["figure.raise_window"] = old_raise
+
+                # check for interaction
+                if self.enter_pressed:
+                    break
+                if self.closed:
+                    # UI design decision: it feels more natural to me (Ben) for
+                    # the plot to not be shown again when it has been closed
+                    # instead of being reinitialized and popping back up.
+                    for processor in self.visualization_processors:
+                        processor._close_figures()
+
+                first_iteration = False

opencsp/common/lib/cv/spot_analysis/image_processor/AbstractVisualizationImageProcessor.py

@@ -1,19 +1,25 @@
 from abc import ABC, abstractmethod
+from typing import Callable
 
 from opencsp.common.lib.cv.spot_analysis.SpotAnalysisOperable import SpotAnalysisOperable
 from opencsp.common.lib.cv.spot_analysis.image_processor.AbstractSpotAnalysisImageProcessor import (
     AbstractSpotAnalysisImagesProcessor,
 )
 import opencsp.common.lib.render_control.RenderControlFigure as rcf
+import opencsp.common.lib.render_control.RenderControlFigureRecord as rcfr
 
 
 class AbstractVisualizationImageProcessor(AbstractSpotAnalysisImagesProcessor, ABC):
-    def __init__(self, name: str):
+    def __init__(self, name: str, interactive: bool | Callable[[SpotAnalysisOperable], bool]):
         # import here to avoid circular dependencies
         from opencsp.common.lib.cv.spot_analysis.VisualizationCoordinator import VisualizationCoordinator
 
         super().__init__(name)
 
+        # register arguments
+        self.interactive = interactive
+
+        # internal values
         self.visualization_coordinator: VisualizationCoordinator = None
         self.initialized_figure_records = False
 
@@ -23,13 +29,17 @@ def num_figures(self) -> int:
         pass
 
     @abstractmethod
-    def _init_figure_records(self, render_control_fig: rcf.RenderControlFigure) -> None:
+    def _init_figure_records(self, render_control_fig: rcf.RenderControlFigure) -> list[rcfr.RenderControlFigureRecord]:
         pass
 
     @abstractmethod
     def _visualize_operable(self, operable: SpotAnalysisOperable, is_last: bool) -> None:
         pass
 
+    @abstractmethod
+    def _close_figures(self):
+        pass
+
     @property
     def has_visualization_coordinator(self) -> bool:
         return self.visualization_coordinator is not None

opencsp/common/lib/cv/spot_analysis/image_processor/View3dImageProcessor.py

@@ -45,9 +45,8 @@ def __init__(
             Crops the image on the x and y axis to the first/last value >= the given threshold. None to not crop the
             image. Useful when trying to inspect hot spots on images with very concentrated values. By default None.
         """
-        super().__init__(self.__class__.__name__)
+        super().__init__(self.__class__.__name__, interactive)
 
-        self.interactive = interactive
         self.max_resolution = max_resolution
         self.crop_to_threshold = crop_to_threshold
 
@@ -57,8 +56,6 @@ def __init__(
         else:
             self.rca = label
         self.rcs = rcs.RenderControlSurface(alpha=1.0, color=None, contour='xyz')
-        self.enter_pressed = False
-        self.closed = False
 
         # declare future values
         self.fig_record: rcfr.RenderControlFigureRecord
@@ -69,7 +66,7 @@ def __init__(
     def num_figures(self) -> int:
         return 1
 
-    def _init_figure_records(self, render_control_fig: rcf.RenderControlFigure):
+    def _init_figure_records(self, render_control_fig: rcf.RenderControlFigure) -> list[rcfr.RenderControlFigureRecord]:
         self.fig_record = fm.setup_figure_for_3d_data(
             render_control_fig,
             self.rca,
@@ -81,28 +78,11 @@ def _init_figure_records(self, render_control_fig: rcf.RenderControlFigure):
         self.view = self.fig_record.view
         self.axes = self.fig_record.figure.gca()
 
-        self.enter_pressed = False
-        self.closed = False
-        self.fig_record.figure.canvas.mpl_connect('close_event', self.on_close)
-        self.fig_record.figure.canvas.mpl_connect('key_release_event', self.on_key_release)
-
-    def on_key_release(self, event: matplotlib.backend_bases.KeyEvent):
-        if event.key == "enter" or event.key == "return":
-            self.enter_pressed = True
-
-    def on_close(self, event: matplotlib.backend_bases.CloseEvent):
-        self.closed = True
+        return [self.fig_record]
 
     def _visualize_operable(self, operable: SpotAnalysisOperable, is_last: bool):
         image = operable.primary_image.nparray
 
-        # check if the view has been closed
-        if self.closed:
-            # UI design decision: it feels more natural to me (Ben) for the plot to not be shown again when it has
-            # been closed instead of being reinitialized and popping back up.
-            return
-            # self._init_figure_record()
-
         # reduce data based on threshold
         y_start, y_end, x_start, x_end = 0, image.shape[0], 0, image.shape[1]
         if self.crop_to_threshold is not None:
@@ -135,11 +115,9 @@ def _visualize_operable(self, operable: SpotAnalysisOperable, is_last: bool):
         # draw
         self.view.show(block=False)
 
-        # wait for the user to press enter
-        wait_for_enter_key = self.interactive if isinstance(self.interactive, bool) else self.interactive(operable)
-        if wait_for_enter_key:
-            self.enter_pressed = False
-            while True:
-                if self.enter_pressed or self.closed:
-                    break
-                self.fig_record.figure.waitforbuttonpress(0.1)
+    def _close_figures(self):
+        self.view.close()
+
+        self.fig_record = None
+        self.view = None
+        self.axes = None

opencsp/common/lib/cv/spot_analysis/image_processor/ViewCrossSectionImageProcessor.py

@@ -56,14 +56,11 @@ def __init__(
             inspect hot spots on images with very concentrated values. By
             default None.
         """
-        super().__init__(self.__class__.__name__)
+        super().__init__(self.__class__.__name__, interactive)
 
         self.cross_section_location = cross_section_location
         self.label = label
         self.single_plot = single_plot
-        self.interactive = interactive
-        self.enter_pressed = False
-        self.closed = False
         self.crop_to_threshold = crop_to_threshold
 
         # initialize certain visualization values
@@ -85,10 +82,7 @@ def num_figures(self) -> int:
         else:
             return 2
 
-    def _init_figure_records(self, render_control_figure: rcf.RenderControlFigure):
-        self.enter_pressed = False
-        self.closed = False
-
+    def _init_figure_records(self, render_control_figure: rcf.RenderControlFigure) -> list[rcfr.RenderControlFigureRecord]:
         self.view_specs = []
         self.rc_axises = []
         self.fig_records = []
@@ -126,8 +120,6 @@ def _init_figure_records(self, render_control_figure: rcf.RenderControlFigure):
             )
             view = fig_record.view
             axes = fig_record.figure.gca()
-            fig_record.figure.canvas.mpl_connect('close_event', self.on_close)
-            fig_record.figure.canvas.mpl_connect('key_release_event', self.on_key_release)
 
             self.view_specs.append(view_spec)
             self.rc_axises.append(rc_axis)
@@ -136,23 +128,11 @@ def _init_figure_records(self, render_control_figure: rcf.RenderControlFigure):
             self.axes.append(axes)
             self.plot_titles.append(plot_title)
 
-    def on_key_release(self, event: matplotlib.backend_bases.KeyEvent):
-        if event.key == "enter" or event.key == "return":
-            self.enter_pressed = True
-
-    def on_close(self, event: matplotlib.backend_bases.CloseEvent):
-        self.closed = True
+        return self.fig_records
 
     def _visualize_operable(self, operable: SpotAnalysisOperable, is_last: bool):
         image = operable.primary_image.nparray
 
-        # check if the view has been closed
-        if self.closed:
-            # UI design decision: it feels more natural to me (Ben) for the plot to not be shown again when it has
-            # been closed instead of being reinitialized and popping back up.
-            return
-            # self._init_figure_record()
-
         # get the cross section pixel location
         if isinstance(self.cross_section_location, tuple):
             cs_loc_x, cs_loc_y = self.cross_section_location
@@ -211,14 +191,16 @@ def _visualize_operable(self, operable: SpotAnalysisOperable, is_last: bool):
         for view in self.views:
             view.show(block=False, legend=self.single_plot)
 
-        # wait for the user to press enter
-        wait_for_enter_key = self.interactive if isinstance(self.interactive, bool) else self.interactive(operable)
-        if wait_for_enter_key:
-            self.enter_pressed = False
-            while True:
-                if self.enter_pressed or self.closed:
-                    break
-                self.fig_records[-1].figure.waitforbuttonpress(0.1)
+    def _close_figures(self):
+        for view in self.views:
+            view.close()
+
+        self.view_specs.clear()
+        self.rc_axises.clear()
+        self.fig_records.clear()
+        self.views.clear()
+        self.axes.clear()
+        self.plot_titles.clear()
 
 
 if __name__ == "__main__":