Add additional explanations

Author: Tom-Willemsen

src/ibex_bluesky_core/callbacks/__init__.py

@@ -16,12 +16,14 @@
 from event_model import RunStart
 from matplotlib.axes import Axes
 
+from ibex_bluesky_core.callbacks._centre_of_mass import (
+    CentreOfMass,
+)
 from ibex_bluesky_core.callbacks._document_logger import DocLoggingCallback
 from ibex_bluesky_core.callbacks._file_logger import (
     HumanReadableFileCallback,
 )
 from ibex_bluesky_core.callbacks._fitting import (
-    CentreOfMass,
     ChainedLiveFit,
     LiveFit,
     LiveFitLogger,

src/ibex_bluesky_core/callbacks/_centre_of_mass.py

@@ -0,0 +1,59 @@
+import logging
+
+import numpy as np
+from bluesky.callbacks import CollectThenCompute
+
+from ibex_bluesky_core.utils import center_of_mass_of_area_under_curve
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["CentreOfMass"]
+
+
+class CentreOfMass(CollectThenCompute):
+    """Compute centre of mass after a run finishes.
+
+    Calculates the CoM of the 2D region bounded by min(y), min(x), max(x),
+    and straight-line segments joining (x, y) data points with their nearest
+    neighbours along the x axis.
+    """
+
+    def __init__(self, x: str, y: str) -> None:
+        """Initialise the callback.
+
+        Args:
+            x: Name of independent variable in event data
+            y: Name of dependent variable in event data
+
+        """
+        super().__init__()
+        self.x: str = x
+        self.y: str = y
+        self._result: float | None = None
+
+    @property
+    def result(self) -> float | None:
+        """The centre-of-mass calculated by this callback."""
+        return self._result
+
+    def compute(self) -> None:
+        """Calculate statistics at the end of the run."""
+        x_values = []
+        y_values = []
+
+        for event in self._events:
+            if self.x not in event["data"]:
+                raise ValueError(f"{self.x} is not in event document.")
+
+            if self.y not in event["data"]:
+                raise ValueError(f"{self.y} is not in event document.")
+
+            x_values.append(event["data"][self.x])
+            y_values.append(event["data"][self.y])
+
+        if not x_values:
+            return
+
+        x_data = np.array(x_values, dtype=np.float64)
+        y_data = np.array(y_values, dtype=np.float64)
+        (self._result, _) = center_of_mass_of_area_under_curve(x_data, y_data)

src/ibex_bluesky_core/callbacks/_fitting.py

@@ -11,7 +11,7 @@
 import numpy as np
 from bluesky.callbacks import CallbackBase, LiveFitPlot
 from bluesky.callbacks import LiveFit as _DefaultLiveFit
-from bluesky.callbacks.core import CollectThenCompute, make_class_safe
+from bluesky.callbacks.core import make_class_safe
 from event_model import Event, EventDescriptor, RunStart, RunStop
 from lmfit import Parameter
 from matplotlib.axes import Axes
@@ -26,11 +26,10 @@
     get_instrument,
 )
 from ibex_bluesky_core.fitting import FitMethod
-from ibex_bluesky_core.utils import center_of_mass_of_area_under_curve
 
 logger = logging.getLogger(__name__)
 
-__all__ = ["CentreOfMass", "ChainedLiveFit", "LiveFit", "LiveFitLogger"]
+__all__ = ["ChainedLiveFit", "LiveFit", "LiveFitLogger"]
 
 
 @make_class_safe(logger=logger)  # pyright: ignore (pyright doesn't understand this decorator)
@@ -256,54 +255,6 @@ def write_fields_table_uncertainty(self) -> None:
         self.csvwriter.writerows(rows)
 
 
-class CentreOfMass(CollectThenCompute):
-    """Compute centre of mass after a run finishes.
-
-    Calculates the CoM of the 2D region bounded by min(y), min(x), max(x),
-    and straight-line segments joining (x, y) data points with their nearest
-    neighbours along the x axis.
-    """
-
-    def __init__(self, x: str, y: str) -> None:
-        """Initialise the callback.
-
-        Args:
-            x: Name of independent variable in event data
-            y: Name of dependent variable in event data
-
-        """
-        super().__init__()
-        self.x: str = x
-        self.y: str = y
-        self._result: float | None = None
-
-    @property
-    def result(self) -> float | None:
-        return self._result
-
-    def compute(self) -> None:
-        """Calculate statistics at the end of the run."""
-        x_values = []
-        y_values = []
-
-        for event in self._events:
-            if self.x not in event["data"]:
-                raise OSError(f"{self.x} is not in event document.")
-
-            if self.y not in event["data"]:
-                raise OSError(f"{self.y} is not in event document.")
-
-            x_values.append(event["data"][self.x])
-            y_values.append(event["data"][self.y])
-
-        if not x_values:
-            return
-
-        x_data = np.array(x_values, dtype=np.float64)
-        y_data = np.array(y_values, dtype=np.float64)
-        (self._result, _) = center_of_mass_of_area_under_curve(x_data, y_data)
-
-
 class ChainedLiveFit(CallbackBase):
     """Processes multiple LiveFits, each fit's results inform the next, with optional plotting.
 

src/ibex_bluesky_core/utils.py

@@ -73,17 +73,35 @@ def center_of_mass_of_area_under_curve(
     Returns a tuple of the centre of mass and the total area under the curve.
 
     """
+    # Sorting here avoids special-cases with disordered points, which may occur
+    # from a there-and-back scan, or from an adaptive scan.
     sort_indices = np.argsort(x, kind="stable")
     x = np.take_along_axis(x, sort_indices, axis=None)
     y = np.take_along_axis(y - np.min(y), sort_indices, axis=None)
+
+    # If the data points are "fence-posts", this calculates the x width of
+    # each "fence panel".
     widths = np.diff(x)
 
     # Area under the curve for two adjacent points is a right trapezoid.
     # Split that trapezoid into a rectangular region, plus a right triangle.
     # Find area and effective X CoM for each.
+
+    # We want the area of the rectangular part of the right trapezoid.
+    # This is width * [height of either left or right point, whichever is lowest]
     rect_areas = widths * np.minimum(y[:-1], y[1:])
+    # CoM of a rectangle in x is simply the average x.
     rect_x_com = (x[:-1] + x[1:]) / 2.0
+
+    # Now the area of the triangular part of the right trapezoid - this is
+    # width * height / 2, where height is the absolute difference between the
+    # two y values.
     triangle_areas = widths * np.abs(y[:-1] - y[1:]) / 2.0
+    # CoM of a right triangle is 1/3 along the base, from the right angle
+    # y[:-1] > y[1:] is true if y_[n] > y_[n+1], (i.e. if the right angle is on the
+    # left-hand side of the triangle).
+    # If that's true, the CoM lies 1/3 of the way along the x axis
+    # Otherwise, the CoM lies 2/3 of the way along the x axis (1/3 from the right angle)
     triangle_x_com = np.where(
         y[:-1] > y[1:], x[:-1] + (widths / 3.0), x[:-1] + (2.0 * widths / 3.0)
     )

tests/callbacks/fitting/test_centre_of_mass.py

@@ -94,7 +94,7 @@ def test_error_thrown_if_no_x_data_in_event():
         }
     )
 
-    with pytest.raises(OSError, match=r"motor is not in event document."):
+    with pytest.raises(ValueError, match=r"motor is not in event document."):
         com.compute()
 
 
@@ -108,5 +108,5 @@ def test_error_thrown_if_no_y_data_in_event():
         }
     )
 
-    with pytest.raises(OSError, match=r"invariant is not in event document."):
+    with pytest.raises(ValueError, match=r"invariant is not in event document."):
         com.compute()

tests/callbacks/test_isis_callbacks.py

@@ -31,7 +31,7 @@ def test_peak_stats_without_peak_stats_callback_raises():
         ).peak_stats
 
 
-def test_centre_of_mass_centre_of_mass_callback_raises():
+def test_GIVEN_centre_of_mass_callback_not_added_WHEN_getting_com_from_isiscallbacks_THEN_raises():
     with pytest.raises(
         ValueError,
         match=r"centre of mass was not added as a callback.",