Move unit tests to their own sub-dir in tests

tests/unit_tests/utilities/__init__.py

@@ -0,0 +1,26 @@
+"""Utilities for testing and debugging.
+
+At the top level are some basic testing functions. More specific testing utilities are
+provided by modules inside the package.
+"""
+
+from collections.abc import Hashable
+from typing import Iterable, Protocol
+
+
+class SizedIterableHashable(Iterable[Hashable], Protocol):
+    """A protocol for sized iterable of hashable objects."""
+
+    def __len__(self) -> int:
+        """Add a len function to protocol so Python knows the object is sized."""
+
+
+def assert_unique_of_length(data: SizedIterableHashable, length: int) -> None:
+    """Assert that a list (or other iterable) has unique contents of a given length.
+
+    :param data: A list or other sized iterable of hashable objects. To be checked
+    for unique contents and length.
+    :param length: The expected length of data
+    """
+    assert len(data) == len(set(data))
+    assert len(data) == length

tests/unit_tests/utilities/lt_test_utils.py

@@ -0,0 +1,225 @@
+"""Test utilities for interacting with the labthings Server."""
+
+import tempfile
+import time
+from types import TracebackType
+from typing import Any, Mapping, Optional, Self, TypeVar
+
+import requests
+from fastapi.testclient import TestClient
+
+import labthings_fastapi as lt
+
+ThingSubclass = TypeVar("ThingSubclass", bound=lt.Thing)
+
+ACTION_RUNNING_KEYWORDS = ["idle", "pending", "running"]
+
+
+class LabThingsTestEnv:
+    """A Labthings Server running in a FastAPI Test Client.
+
+    This can be used to create ThingClients for testing, to to enable direct access to
+    the server, or things on the server. It also provides high level functions to
+    start actions without blocking the test thread.
+
+    Use this as a context manager:
+
+    .. code-block:: python
+
+        with LabThingsTestEnv(things={"counter", CountingThing}) as env:
+            ...
+    """
+
+    def __init__(
+        self,
+        things: Mapping[str, lt.Thing | str],
+        settings_folder: Optional[str] = None,
+    ) -> None:
+        """Initialise the test environment.
+
+        The server and FastAPI TestClient are server are not created until this is used
+        as a context manager:
+
+        :param things: The thing configuration dictionary used to initialise the server.
+        :param settings_folder: The settings folder to use.
+        """
+        self._server: Optional[lt.ThingServer]
+        self._test_client: Optional[TestClient]
+        self._things_config = things
+        self._settings_folder = settings_folder
+        self._tmp_dir_obj: Optional[tempfile.TemporaryDirectory] = None
+
+    def __enter__(self) -> Self:
+        """Create server and run it in the TestClient."""
+        if self._settings_folder is None:
+            self._tmp_dir_obj = tempfile.TemporaryDirectory()
+            self._settings_folder = self._tmp_dir_obj.name
+        self._server = lt.ThingServer(
+            things=self._things_config, settings_folder=self._settings_folder
+        )
+        self._test_client = TestClient(self._server.app)
+        self._test_client.__enter__()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException],
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        """Close the TestClient and cleanup."""
+        self._test_client.__exit__(exc_type, exc_value, traceback)
+        if self._tmp_dir_obj is not None:
+            self._tmp_dir_obj.cleanup()
+
+    @property
+    def server(self) -> lt.ThingServer:
+        """The LabThings Server."""
+        if self._server is None:
+            raise RuntimeError("No server was started.")
+        return self._server
+
+    @property
+    def client(self) -> TestClient:
+        """The FastAPI TestClient running the server."""
+        if self._test_client is None:
+            raise RuntimeError("No server was started.")
+        return self._test_client
+
+    def check_thing_exists(self, thing_name: str):
+        """Raise a ValueError if no thing of with a matching name exists."""
+        if thing_name not in self.server.things:
+            raise ValueError(f"No Thing named {thing_name}")
+
+    def get_thing_by_name(self, thing_name: str) -> lt.Thing:
+        """Get a Thing from the server by name.
+
+        :param thing_name: The name of the thing to on the server.
+
+        :return: The Thing with the specified name.
+        """
+        self.check_thing_exists(thing_name)
+        return self.server.things[thing_name]
+
+    def get_thing_by_type(self, thing_class: type[ThingSubclass]) -> ThingSubclass:
+        """Get a thing by type.
+
+        :param thing_class: The subclass of thing to match.
+
+        :return: The Thing that matches the subclass.
+
+        :raises RuntimeError: If there are multiple things of the same type, or no
+            matching thing. If there are multiple things of this type use
+            ``get_thing_by_name`` or ``get_all_things_by_type``.
+        """
+        matching = self.get_all_things_by_type(thing_class)
+        n_things = len(matching)
+        if n_things == 0:
+            raise RuntimeError(f"No Thing of type {thing_class} on this server.")
+        if n_things > 1:
+            raise RuntimeError(
+                f"Cannot get Thing by type as there are {n_things} of type "
+                f"{thing_class} on this server."
+            )
+        return next(iter(matching.values()))
+
+    def get_all_things_by_type(
+        self, thing_class: type[ThingSubclass]
+    ) -> Mapping[str, ThingSubclass]:
+        """Get a dictionary of all things by matching a type.
+
+        :param thing_class: The subclass of thing to match.
+
+        :return: A dictionary of Things that match the subclass. If none match this
+            will be and empty dictionary.
+        """
+        matching = {}
+        for thing_name, thing in self.server.things.items():
+            if isinstance(thing, thing_class):
+                matching[thing_name] = thing
+        return matching
+
+    def get_thing_client(self, thing_name: str) -> lt.ThingClient:
+        """Get a ThingClient for a Thing by name.
+
+        :param thing_name: The name of the thing to on the server.
+
+        :return: A LabThings ThingClient for the Thing with the specified name.
+        """
+        self.check_thing_exists(thing_name)
+        thing = self.server.things[thing_name]
+        return lt.ThingClient.from_url(thing.path, self.client)
+
+    def start_action(
+        self,
+        thing_name: str,
+        action_name: str,
+        action_kwargs: Optional[Mapping[str, Any]] = None,
+    ) -> requests.Response:
+        """Start an action and return the server response.
+
+        For most purposes the best way to run an action is to use ``get_thing_client``
+        to create a ThingClient. At this point any actions can be run with a similar
+        Python API to calling directly. However, using ThingClient blocks the test
+        thread.
+
+        This function provides an alternative way to start actions without blocking the
+        test thread. It will return the HTTP response, this response can be used to
+        poll or cancel the action. Use this method if you:
+
+        * Want to test cancelling an action.
+        * Want to inspect the actions logs exactly as they would come to a web client
+        * Direcltly interact with the HTTP API
+
+        :param thing_name: The name of the Thing on the server.
+        :param action_name: The name of the action to start.
+        :action_kwargs: The keyword inputs to the action.
+        :return: A Response object with the HTTP response.
+        """
+        self.check_thing_exists(thing_name)
+        url = f"/{thing_name}/{action_name}"
+        json_payload = {} if action_kwargs is None else action_kwargs
+        response = self.client.post(url, json=json_payload)
+        response.raise_for_status()
+        return response
+
+    def poll_action(
+        self, response: requests.Response, interval: float = 0.01
+    ) -> Mapping[str, Any]:
+        """Poll an action until it completes and return the final response data.
+
+        :param response: The response from starting this action with ``start_action``.
+        """
+        invocation_data = response.json()
+
+        if "status" not in invocation_data:
+            raise ValueError(f"Response json has no status: {invocation_data}")
+        first_run = True
+        while invocation_data["status"] in ACTION_RUNNING_KEYWORDS:
+            if first_run:
+                first_run = False
+            else:
+                time.sleep(interval)
+            response = self.client.get(_invocation_href(invocation_data))
+            response.raise_for_status()
+            invocation_data = response.json()
+        return invocation_data
+
+    def cancel_action(self, response: requests.Response) -> None:
+        """Cancel an ongoing action.
+
+        :param response: The response from starting this action with ``start_action``.
+        """
+        invocation_data = response.json()
+        response = self.client.delete(_invocation_href(invocation_data))
+        response.raise_for_status()
+
+
+def _get_link(obj: Mapping[str, Any], rel: str) -> Mapping[str, Any]:
+    """Retrieve a link from an object's `links` list, by its `rel` attribute."""
+    return next(link for link in obj["links"] if link["rel"] == rel)
+
+
+def _invocation_href(invocation_data: Mapping[str, Any]) -> str:
+    """Get the invocation href from the invocation response data."""
+    return _get_link(invocation_data, "self")["href"]

tests/unit_tests/utilities/scan_test_helpers.py

@@ -0,0 +1,270 @@
+#! /usr/bin/env python3
+
+"""Utility functions for testing scan planners.
+
+These including fake sample creation, scan path visualisation, and
+persistent storage of expected scan paths for samples.
+"""
+
+import argparse
+import os
+import pickle
+
+import numpy as np
+from matplotlib import pyplot as plt
+from matplotlib.figure import Figure
+from matplotlib.patches import PathPatch
+from matplotlib.path import Path as MatPath
+from scipy import interpolate
+
+from openflexure_microscope_server import scan_planners
+
+THIS_DIR = os.path.dirname(os.path.realpath(__file__))
+
+ALL_SAMPLE_NAMES = ("lobed", "regular", "core")
+
+
+class FakeSample:
+    """A fake sample to test scan algorithms.
+
+    The sample is able to return whether a given position is sample, there is no image
+    associated with the sample
+    """
+
+    def __init__(self, xy_points: list[tuple[int, int]]):
+        """Create the sample from a spline interpolation around the given points."""
+        self._sample_perimeter = interp_closed_path(xy_points, 500)
+
+    def is_sample(self, pos: tuple[int, int], im_size: tuple[int, int]) -> bool:
+        """Return True if an image at a given location is on the sample.
+
+        The image size is specified to check if it overlaps the sample. It doesn't
+        check the entire image field as this is designed to be used where the fake
+        sample is much larger than the image and has smooth edges. It just checks the
+        4 corners.
+        """
+        img_corners = [
+            (pos[0] + im_size[0], pos[1] + im_size[1]),
+            (pos[0] + im_size[0], pos[1] - im_size[1]),
+            (pos[0] - im_size[0], pos[1] + im_size[1]),
+            (pos[0] - im_size[0], pos[1] - im_size[1]),
+        ]
+        return any(
+            self._sample_perimeter.contains_point(corner) for corner in img_corners
+        )
+
+    @property
+    def patch(self) -> PathPatch:
+        """The sample as a matplotlib patch for plotting."""
+        patch = PathPatch(self._sample_perimeter)
+        patch.set(color=(1.0, 0.8, 1.0, 1.0))
+        return patch
+
+
+def visualise_scan(sample: FakeSample, planner: scan_planners.ScanPlanner) -> Figure:
+    """For a given sample and scanner object return a matplotlib figure of the scan."""
+    fig, ax = plt.subplots(figsize=(8, 8))
+    ax.add_artist(sample.patch)
+    xh, yh = zip(*planner.path_history, strict=True)
+    xi, yi, _zi = zip(*planner.imaged_locations, strict=True)
+    if planner.secondary_locations:
+        xs, ys, _zs = zip(*planner.secondary_locations, strict=True)
+    else:
+        xs, ys, _zs = [], [], []
+
+    # convert history to numpy array so can calculate quiver arrows
+    xh = np.array(xh)
+    yh = np.array(yh)
+
+    plt.quiver(
+        xh[:-1],
+        yh[:-1],
+        xh[1:] - xh[:-1],
+        yh[1:] - yh[:-1],
+        scale_units="xy",
+        angles="xy",
+        scale=1,
+    )
+    plt.plot(xh, yh, "r.")
+    plt.plot(xi, yi, "g*")
+    plt.plot(xs, ys, "o", mfc="none", mec="blue")
+    ax.axis("equal")
+    return fig
+
+
+def interp_closed_path(xy_points: list[tuple[int, int]], n_points: int) -> MatPath:
+    """Interpolate an n_point closed curve from a lists of xy_points.
+
+    This can be used to create an arbitrary sample shape for testing a scan planner.
+
+    Modified from:
+    https://stackoverflow.com/questions/33962717/interpolating-a-closed-curve-using-scipy
+    """
+    # Use zip to separate x and y points into tuples
+    x, y = zip(*xy_points, strict=True)
+
+    # Append first point and convert to array
+    x = np.array(x + (x[0],))
+    y = np.array(y + (y[0],))
+
+    # fit splines to x=f(u) and y=g(u), treating both as periodic. also note that s=0
+    # is needed in order to force the spline fit to pass through all the input points.
+    spline_data, *_unused = interpolate.splprep([x, y], s=0, per=True)
+
+    # evaluate the spline
+    xi, yi = interpolate.splev(np.linspace(0, 1, n_points), spline_data)
+
+    # Convert to a matplotlib closed path
+    path_points = [[xp, yp] for xp, yp in (zip(xi, yi, strict=True))]
+    return MatPath(path_points, closed=True)
+
+
+def example_smart_spiral(
+    sample_name: str = "lobed",
+) -> tuple[FakeSample, scan_planners.ScanPlanner]:
+    """Run an example scan.
+
+    :returns: The sample scanned and the planner object after scan is complete.
+    """
+    xy_sample_points = load_sample_points(sample_name)
+    sample = FakeSample(xy_sample_points)
+    img_size = (1000, 1000)
+    initial_position = (0, 0)
+    planner_settings = {"dx": 1200, "dy": 800, "max_dist": 100000}
+    planner = scan_planners.SmartSpiral(
+        initial_position=initial_position, planner_settings=planner_settings
+    )
+
+    while not planner.scan_complete:
+        xy_pos, _ = planner.get_next_location_and_z_estimate()
+        xyz_pos = (xy_pos[0], xy_pos[1], 0)
+        imaged = sample.is_sample(xy_pos, img_size)
+        planner.mark_location_visited(xyz_pos, imaged=imaged, focused=imaged)
+    return sample, planner
+
+
+def profile_example_smart_spiral():
+    """Profile running an example scan and print the cumulative profile stats."""
+    import cProfile
+    import pstats
+
+    profiler = cProfile.Profile()
+    profiler.runcall(example_smart_spiral)
+
+    stats_fname = os.path.join(THIS_DIR, "scan_example_stats.pstats")
+    profiler.dump_stats(stats_fname)
+
+    run_stats = pstats.Stats(stats_fname)
+    run_stats.strip_dirs()
+    run_stats.sort_stats("cumulative")
+    run_stats.print_stats("scan_planners.py")
+
+
+def plot_all_examples():
+    """Plot all examples as png files."""
+    for sample_name in ALL_SAMPLE_NAMES:
+        sample, planner = example_smart_spiral(sample_name)
+        png_fname = os.path.join(THIS_DIR, f"scan_example_plot_{sample_name}.png")
+        fig = visualise_scan(sample, planner)
+        fig.savefig(png_fname, dpi=200)
+
+
+def update_example_smart_spiral_pickles():
+    """Pickle the ScanPlanner for the example_smart_spiral().
+
+    This is done so the history can be compared by testing to check
+    the algorithm is unchanged.
+
+    If the algorithm is purposefully changed then this will need to be
+    run to update the pickle for the test to pass.
+
+    Takes sample, the sample type we have generated, so we can make a
+    pickle for each sample type
+    """
+    for sample_name in ALL_SAMPLE_NAMES:
+        pkl_fname = os.path.join(THIS_DIR, f"example_smart_spiral_{sample_name}.pkl")
+        _, planner = example_smart_spiral(sample_name)
+        with open(pkl_fname, "wb") as pkl_file_obj:
+            pickle.dump(planner, pkl_file_obj, pickle.HIGHEST_PROTOCOL)
+
+
+def get_expected_result_for_example_smart_spiral(
+    sample_name: str,
+) -> scan_planners.ScanPlanner:
+    """Return the expected ScanPlanner object for the example_smart_spiral().
+
+    This is loaded from a pickle so that the object can be committed to the repo.
+    """
+    pkl_fname = os.path.join(THIS_DIR, f"example_smart_spiral_{sample_name}.pkl")
+    with open(pkl_fname, "rb") as pkl_file_obj:
+        return pickle.load(pkl_file_obj)
+
+
+def load_sample_points(sample_name: str):
+    """Return the points to generate the FakeSample corresponding to the given input name.
+
+    Options are "lobed", "regular", and "core".
+    """
+    sample_options = {
+        "lobed": [
+            (-5000, -5000),
+            (-2000, 16000),
+            (1000, 2000),
+            (6000, 7000),
+            (9000, 2000),
+        ],
+        "regular": [
+            (-5000, -5000),
+            (-5000, 5000),
+            (5000, 5000),
+            (5000, -5000),
+        ],
+        "core": [
+            (-12000, 2000),
+            (-12000, 1000),
+            (0, -2000),
+            (10000, -2000),
+            (10000, -1000),
+            (1000, 0),
+        ],
+    }
+    if sample_name not in sample_options:
+        all_samples = ", ".join(sample_options.keys())
+        raise ValueError(
+            f"{sample_name} is not a valid sample name. Valid names : {all_samples}"
+        )
+    return sample_options[sample_name]
+
+
+def main():
+    """Run the profiler, the plotting, or update the pickles based on command line input.
+
+    This only runs if run as a command line script.
+    """
+    parser = argparse.ArgumentParser(description="Simulated scan-planning utility")
+
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    # profile
+    subparsers.add_parser("profile", help="Run simulation under cProfile")
+
+    # plot
+    subparsers.add_parser("plot", help="Run simulation and produce plots")
+
+    # update
+    subparsers.add_parser("update", help="Update stored reference pickles")
+
+    args = parser.parse_args()
+
+    if args.command == "profile":
+        profile_example_smart_spiral()
+    elif args.command == "plot":
+        plot_all_examples()
+    elif args.command == "update":
+        update_example_smart_spiral_pickles()
+    else:
+        parser.error(f"Unknown command: {args.command}")
+
+
+if __name__ == "__main__":
+    main()