82 spot analysis for peak intensity correction

contrib/app/SpotAnalysis/PeakFlux.py

@@ -0,0 +1,116 @@
+import os
+import re
+
+
+import opencsp.common.lib.cv.SpotAnalysis as sa
+from opencsp.common.lib.cv.spot_analysis.SpotAnalysisImagesStream import ImageType
+import opencsp.common.lib.cv.spot_analysis.SpotAnalysisOperableAttributeParser as saoap
+from opencsp.common.lib.cv.spot_analysis.image_processor import *
+import opencsp.common.lib.tool.file_tools as ft
+import opencsp.common.lib.tool.log_tools as lt
+import opencsp.common.lib.tool.time_date_tools as tdt
+
+
+class PeakFlux:
+    """
+    A class to process images from heliostat sweeps across a target, to find the spot of maximum flux from the
+    heliostat.
+
+    The input includes::
+
+        - A series of images with the heliostat on target, and with the target in ambient light conditions. These images
+          should be clearly labeled with the name of the heliostat under test, and whether the target is under ambient
+          light or heliostat reflected light.
+        - The pixel intensity to flux correction mapping.
+
+    The generated output includes::
+
+        - Over/under exposure warnings
+        - Per-heliostat heatmap visualizations
+        - Per-heliostat peak flux identification
+    """
+
+    def __init__(self, indir: str, outdir: str, experiment_name: str, settings_path_name_ext: str):
+        self.indir = indir
+        self.outdir = outdir
+        self.experiment_name = experiment_name
+        self.settings_path_name_ext = settings_path_name_ext
+
+        settings_path, settings_name, settings_ext = ft.path_components(self.settings_path_name_ext)
+        settings_dict = ft.read_json("PeakFlux settings", settings_path, settings_name + settings_ext)
+        self.crop_box: list[int] = settings_dict['crop_box']
+        self.bcs_pixel: list[int] = settings_dict['bcs_pixel_location']
+        self.heliostate_name_pattern = re.compile(settings_dict['heliostat_name_pattern'])
+
+        group_assigner = AverageByGroupImageProcessor.group_by_name(re.compile(r"(_off)?( Raw)"))
+        group_trigger = AverageByGroupImageProcessor.group_trigger_on_change()
+        supporting_images_map = {
+            ImageType.PRIMARY: lambda operable, operables: "off" not in operable.primary_image_source_path,
+            ImageType.NULL: lambda operable, operables: "off" in operable.primary_image_source_path,
+        }
+
+        self.image_processors: list[AbstractSpotAnalysisImagesProcessor] = [
+            CroppingImageProcessor(*self.crop_box),
+            AverageByGroupImageProcessor(group_assigner, group_trigger),
+            EchoImageProcessor(),
+            SupportingImagesCollectorImageProcessor(supporting_images_map),
+            NullImageSubtractionImageProcessor(),
+            ConvolutionImageProcessor(kernel="box", diameter=3),
+            BcsLocatorImageProcessor(),
+            PopulationStatisticsImageProcessor(initial_min=0, initial_max=255),
+            FalseColorImageProcessor(),
+            AnnotationImageProcessor(),
+        ]
+        self.spot_analysis = sa.SpotAnalysis(
+            experiment_name, self.image_processors, save_dir=outdir, save_overwrite=True
+        )
+
+        filenames = ft.files_in_directory_by_extension(self.indir, [".jpg"])[".jpg"]
+        source_path_name_exts = [os.path.join(self.indir, filename) for filename in filenames]
+        self.spot_analysis.set_primary_images(source_path_name_exts)
+
+    def run(self):
+        # process all images from indir
+        for result in self.spot_analysis:
+            # save the processed image
+            save_path = self.spot_analysis.save_image(
+                result, self.outdir, save_ext="png", also_save_supporting_images=False, also_save_attributes_file=True
+            )
+            if save_path is None:
+                lt.warn(
+                    f"Warning in PeakFlux.run(): failed to save image. "
+                    + "Maybe SpotAnalaysis.save_overwrite is False? ({self.spot_analysis.save_overwrite=})"
+                )
+            else:
+                lt.info(f"Saved image to {save_path}")
+
+            # Get the attributes of the processed image, to save the results we're most interested in into a single
+            # condensed csv file.
+            parser = saoap.SpotAnalysisOperableAttributeParser(result, self.spot_analysis)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        prog=__file__.rstrip(".py"), description='Processes images to find the point of peak flux.'
+    )
+    parser.add_argument('indir', type=str, help="Directory with images to be processed.")
+    parser.add_argument('outdir', type=str, help="Directory for where to put processed images and computed results.")
+    parser.add_argument('experiment_name', type=str, help="A description of the current data collection.")
+    parser.add_argument('settings_file', type=str, help="Path to the settings JSON file for this PeakFlux evaluation.")
+    args = parser.parse_args()
+
+    # create the output directory
+    ft.create_directories_if_necessary(args.outdir)
+    ft.delete_files_in_directory(args.outdir, "*")
+
+    # create the log file
+    log_path_name_ext = os.path.join(args.outdir, "PeakFlux_" + tdt.current_date_time_string_forfile() + ".log")
+    lt.logger(log_path_name_ext)
+
+    # validate the rest of the inputs
+    if not ft.directory_exists(args.indir):
+        lt.error_and_raise(FileNotFoundError, f"Error in PeakFlux.py: input directory '{args.indir}' does not exist!")
+
+    PeakFlux(args.indir, args.outdir, args.experiment_name, args.settings_file).run()

opencsp/common/lib/cv/AbstractFiducials.py

@@ -0,0 +1,179 @@
+from abc import ABC, abstractmethod
+from typing import Callable
+
+import matplotlib.axes
+import matplotlib.pyplot as plt
+import numpy as np
+import scipy.spatial
+
+import opencsp.common.lib.geometry.Pxy as p2
+import opencsp.common.lib.geometry.RegionXY as reg
+import opencsp.common.lib.geometry.Vxyz as v3
+import opencsp.common.lib.render.figure_management as fm
+import opencsp.common.lib.render_control.RenderControlPointSeq as rcps
+import opencsp.common.lib.tool.log_tools as lt
+
+
+class AbstractFiducials(ABC):
+    """
+    A collection of markers (such as an ArUco board) that is used to orient the camera relative to observed objects
+    in the scene. It is suggested that each implementing class be paired with a complementary locator method or
+    SpotAnalysisImageProcessor.
+    """
+
+    def __init__(self, style=None, pixels_to_meters: Callable[[p2.Pxy], v3.Vxyz] = None):
+        """
+        Parameters
+        ----------
+        style : RenderControlPointSeq, optional
+            How to render this fiducial when using the defaul render_to_plot() method. By default rcps.default().
+        pixels_to_meters : Callable[[p2.Pxy], v3.Vxyz], optional
+            Conversion function to get the physical point in space for the given x/y position information. Used in the
+            default self.scale implementation. A good implementation of this function will correct for many factors such
+            as relative camera position and camera distortion. For extreme accuracy, this will also account for
+            non-uniformity in the target surface. Defaults to a simple 1 meter per pixel model.
+        """
+        self.style = style if style is not None else rcps.default()
+        self.pixels_to_meters = pixels_to_meters
+
+    @abstractmethod
+    def get_bounding_box(self, index=0) -> reg.RegionXY:
+        """The X/Y bounding box(es) of this instance, in pixels."""
+
+    @property
+    @abstractmethod
+    def origin(self) -> p2.Pxy:
+        """The origin point(s) of this instance, in pixels."""
+
+    @property
+    @abstractmethod
+    def rotation(self) -> scipy.spatial.transform.Rotation:
+        """
+        The pointing of the normal vector(s) of this instance.
+        This is relative to the camera's reference frame, where x is positive
+        to the right, y is positive down, and z is positive in (away from the
+        camera).
+
+        This can be used to describe the forward transformation from the
+        camera's perspective. For example, an aruco marker whose origin is in
+        the center of the image and is facing towards the camera could have the
+        rotation::
+
+            Rotation.from_euler('y', np.pi)
+
+        If that same aruco marker was also placed upside down, then it's
+        rotation could be::
+
+            Rotation.from_euler(
+                'yz',
+                [ [np.pi, 0],
+                  [0,     np.pi] ]
+            )
+
+        Not that this just describes rotation, and not the translation. We call
+        the rotation and translation together the orientation.
+        """
+
+    @property
+    @abstractmethod
+    def size(self) -> list[float]:
+        """The scale(s) of this fiducial, in pixels, relative to its longest axis.
+        For example, if the fiducial is a square QR-code and is oriented tangent
+        to the camera, then the scale will be the number of pixels from one
+        corner to the other."""  # TODO is this a good definition?
+
+    @property
+    def scale(self) -> list[float]:
+        """
+        The scale(s) of this fiducial, in meters, relative to its longest axis.
+        This can be used to determine the distance and rotation of the
+        fiducial relative to the camera.
+        """
+        ret = []
+
+        for i in range(len(self.origin)):
+            bb = self.get_bounding_box(i)
+            left_px, right_px, bottom_px, top_px = bb.loops[0].axis_aligned_bounding_box()
+            top_left_m = self.pixels_to_meters(p2.Pxy([left_px, top_px]))
+            bottom_right_m = self.pixels_to_meters(p2.Pxy([right_px, bottom_px]))
+            scale = (bottom_right_m - top_left_m).magnitude()[0]
+            ret.append(scale)
+
+        return ret
+
+    def _render(self, axes: matplotlib.axes.Axes):
+        """
+        Called from render(). The parameters are always guaranteed to be set.
+        """
+        axes.scatter(
+            self.origin.x,
+            self.origin.y,
+            linewidth=self.style.linewidth,
+            marker=self.style.marker,
+            s=self.style.markersize,
+            c=self.style.markerfacecolor,
+            edgecolor=self.style.markeredgecolor,
+        )
+
+    def render(self, axes: matplotlib.axes.Axes = None):
+        """
+        Renders this fiducial to the active matplotlib.pyplot plot.
+
+        The default implementation uses plt.scatter().
+
+        Parameters
+        ----------
+        axes: matplotlib.axes.Axes, optional
+            The plot to render to. Uses the active plot if None. Default is None.
+        """
+        if axes is None:
+            axes = plt.gca()
+        self._render(axes)
+
+    def render_to_image(self, image: np.ndarray) -> np.ndarray:
+        """
+        Renders this fiducial to the a new image on top of the given image.
+
+        The default implementation creates a new matplotlib plot, and then renders to it with self.render_to_plot().
+        """
+        # Create the figure to plot to
+        dpi = 300
+        width = image.shape[1]
+        height = image.shape[0]
+        fig = fm.mpl_pyplot_figure(figsize=(width / dpi, height / dpi), dpi=dpi)
+
+        try:
+            # A portion of this code is from:
+            # https://stackoverflow.com/questions/35355930/figure-to-image-as-a-numpy-array
+
+            # Get the axis and canvas
+            axes = fig.gca()
+            canvas = fig.canvas
+
+            # Image from plot
+            axes.axis('off')
+            fig.tight_layout(pad=0)
+
+            # To remove the huge white borders
+            axes.margins(0)
+
+            # Prepare the image and the feature points
+            axes.imshow(image)
+            self.render(axes)
+
+            # Render
+            canvas.draw()
+
+            # Convert back to a numpy array
+            new_image = np.asarray(canvas.buffer_rgba())
+            new_image = new_image.astype(image.dtype)
+
+            # Return the updated image
+            return new_image
+
+        except Exception as ex:
+            lt.error("Error in AnnotationImageProcessor.render_points(): " + repr(ex))
+            raise
+
+        finally:
+            plt.close(fig)