add queue to astar algorithm to show points being looked at

.github/workflows/test.yml

@@ -30,6 +30,6 @@ jobs:
         pip install '.[test]'
     - name: Test and generate coverage report
       run: |
-        pytest --cov=./tests --cov-report=xml --ignore=./brightest_path_lib/sandbox.py
+        pytest --cov=./tests --cov-report=xml --ignore=./brightest_path_lib/sandbox.py,./brightest_path_lib/sandbox2.py
     - name: Upload coverage to Codecov
       uses: codecov/codecov-action@v3

brightest_path_lib/algorithm/astar.py

@@ -1,8 +1,8 @@
 from collections import defaultdict
 import math
 import numpy as np
-from queue import PriorityQueue
-from typing import List
+from queue import PriorityQueue, Queue
+from typing import List, Tuple
 from brightest_path_lib.cost import Reciprocal
 from brightest_path_lib.heuristic import Euclidean
 from brightest_path_lib.image import ImageStats
@@ -19,10 +19,16 @@ class AStarSearch:
         the 2D/3D image on which we will run an A star search
     start_point : numpy ndarray
         the 2D/3D coordinates of the starting point (could be a pixel or a voxel)
+        For 2D images, the coordinates are of the form (y, x)
+        For 2D images, the coordinates are of the form (z, x, y)
     goal_point : numpy ndarray
         the 2D/3D coordinates of the goal point (could be a pixel or a voxel)
-    scale : float
-        the scale of the image; defaults to 1.0 if image is not zoomed in/out
+        For 2D images, the coordinates are of the form (y, x)
+        For 2D images, the coordinates are of the form (z, x, y)
+    scale : Tuple
+        the scale of the image; defaults to (1.0, 1.0), i.e. image is not zoomed in/out
+        For 2D images, the scale is of the form (x, y)
+        For 2D images, the scale is of the form (x, y, z)
     cost_function : Enum CostFunction
         this enum value specifies the cost function to be used for computing 
         the cost of moving to a new point
@@ -41,8 +47,8 @@ class AStarSearch:
         the coordinates of the start point
     goal_point : numpy ndarray
         the coordinates of the goal point
-    scale : float
-        the scale of the image; defaults to 1.0 if image is not zoomed in/out
+    scale : Tuple
+        the scale of the image; defaults to (1.0, 1.0), i.e. image is not zoomed in/out
     cost_function : Cost
         the cost function to be used for computing the cost of moving 
         to a new point
@@ -62,9 +68,10 @@ def __init__(
         image: np.ndarray,
         start_point: np.ndarray,
         goal_point: np.ndarray,
-        scale: np.ndarray = np.array([1.0, 1.0]),
+        scale: Tuple = (1.0, 1.0),
         cost_function: CostFunction = CostFunction.RECIPROCAL,
-        heuristic_function: HeuristicFunction = HeuristicFunction.EUCLIDEAN
+        heuristic_function: HeuristicFunction = HeuristicFunction.EUCLIDEAN,
+        open_nodes: Queue = None
     ):
 
         self._validate_inputs(image, start_point, goal_point)
@@ -74,6 +81,7 @@ def __init__(
         self.start_point = start_point
         self.goal_point = goal_point
         self.scale = scale
+        self.open_nodes = open_nodes
 
         if cost_function == CostFunction.RECIPROCAL:
             self.cost_function = Reciprocal(
@@ -83,6 +91,8 @@ def __init__(
         if heuristic_function == HeuristicFunction.EUCLIDEAN:
             self.heuristic_function = Euclidean(scale=self.scale)
 
+        self.is_canceled = False
+        self.found_path = False
         self.result = []
 
     def _validate_inputs(
@@ -97,6 +107,26 @@ def _validate_inputs(
         if len(image) == 0 or len(start_point) == 0 or len(goal_point) == 0:
             raise ValueError
 
+    @property
+    def found_path(self) -> bool:
+        return self._found_path
+
+    @found_path.setter
+    def found_path(self, value: bool):
+        if value is None:
+            raise TypeError
+        self._found_path = value
+
+    @property
+    def is_canceled(self) -> bool:
+        return self._is_canceled
+
+    @is_canceled.setter
+    def is_canceled(self, value: bool):
+        if value is None:
+            raise TypeError
+        self._is_canceled = value
+
     def search(self) -> List[np.ndarray]:
         """Function that performs A star search
 
@@ -122,6 +152,8 @@ def search(self) -> List[np.ndarray]:
         f_scores[tuple(self.start_point)] = start_node.f_score
 
         while not open_set.empty():
+            if self.is_canceled:
+                break
             current_node = open_set.get()[2]
             current_coordinates = tuple(current_node.point)
             if current_coordinates in close_set_hash:
@@ -132,6 +164,7 @@ def search(self) -> List[np.ndarray]:
             if self._found_goal(current_node.point):
                 print("Found goal!")
                 self._construct_path_from(current_node)
+                self.found_path = True
                 break
 
             neighbors = self._find_neighbors_of(current_node)
@@ -144,6 +177,10 @@ def search(self) -> List[np.ndarray]:
                     count += 1
                     open_set.put((neighbor.f_score, count, neighbor))
                     open_set_hash.add(neighbor_coordinates)
+                    if self.open_nodes is not None:
+                        # add to our queue
+                        # can be monitored from caller to update plots
+                        self.open_nodes.put(neighbor_coordinates)
                 else:
                     if neighbor.f_score < f_scores[neighbor_coordinates]:
                         f_scores[neighbor_coordinates] = neighbor.f_score
@@ -203,7 +240,7 @@ def _find_2D_neighbors_of(self, node: Node) -> List[Node]:
         diagonal neighbors: top-left, top-right, bottom-left, bottom-right
         - Of course, we need to check for invalid cases where we can't move
         in these directions
-        - 2D coordinates are of the type (x, y)
+        - 2D coordinates are of the type (y, x)
         """
         neighbors = []
         steps = [-1, 0, 1]
@@ -212,19 +249,23 @@ def _find_2D_neighbors_of(self, node: Node) -> List[Node]:
                 if xdiff == ydiff == 0:
                     continue
 
-                new_x = node.point[0] + xdiff
+                new_x = node.point[1] + xdiff
+                # new_x = node.point[0] + xdiff
                 if new_x < self.image_stats.x_min or new_x > self.image_stats.x_max:
                     continue
 
-                new_y = node.point[1] + ydiff
+                new_y = node.point[0] + ydiff
+                # new_y = node.point[1] + ydiff
                 if new_y < self.image_stats.y_min or new_y > self.image_stats.y_max:
                     continue
 
-                new_point = np.array([new_x, new_y])
+                # new_point = np.array([new_x, new_y])
+                new_point = np.array([new_y, new_x])
 
                 h_for_new_point = self._estimate_cost_to_goal(new_point)
 
-                intensity_at_new_point = self.image[new_x, new_y]
+                intensity_at_new_point = self.image[new_y, new_x]
+
                 cost_of_moving_to_new_point = self.cost_function.cost_of_moving_to(intensity_at_new_point)
                 if cost_of_moving_to_new_point < self.cost_function.minimum_step_cost():
                     cost_of_moving_to_new_point = self.cost_function.minimum_step_cost()
@@ -235,7 +276,7 @@ def _find_2D_neighbors_of(self, node: Node) -> List[Node]:
                     g_score=g_for_new_point,
                     h_score=h_for_new_point,
                     predecessor=node
-                ) 
+                )
 
                 neighbors.append(neighbor)
 

brightest_path_lib/heuristic/euclidean.py

@@ -1,16 +1,17 @@
 from brightest_path_lib.heuristic import Heuristic
 import math
 import numpy as np
+from typing import Tuple
 
 class Euclidean(Heuristic):
     """heuristic cost estimation using Euclidean distance from current point to goal point
 
     Parameters
     ----------
-    scale : numpy ndarray
-        the scale of the image's axes. For example [1.0 1.0] for a 2D image.
-        - for 2D points, the order of coordinates is: (x, y)
-        - for 3D points, the order of coordinates is: (x, y, z)
+    scale : Tuple
+        the scale of the image's axes. For example (1.0 1.0) for a 2D image.
+        - for 2D points, the order of scale is: (x, y)
+        - for 3D points, the order of scale is: (x, y, z)
 
     Attributes
     ----------
@@ -23,7 +24,7 @@ class Euclidean(Heuristic):
 
     """
 
-    def __init__(self, scale: np.ndarray):
+    def __init__(self, scale: Tuple):
         if scale is None:
             raise TypeError
         if len(scale) == 0:
@@ -59,16 +60,18 @@ def estimate_cost_to_goal(self, current_point: np.ndarray, goal_point: np.ndarra
         By including the scale in the calculation of distance to the goal we
         can get an accurate cost.
 
-        - for 2D points, the order of coordinates is: (x, y)
+        - for 2D points, the order of coordinates is: (y, x)
         - for 3D points, the order of coordinates is: (z, x, y)
         """
         if current_point is None or goal_point is None:
             raise TypeError
         if (len(current_point) == 0 or len(goal_point) == 0) or (len(current_point) != len(goal_point)):
             raise ValueError
-
-        current_x, current_y, current_z = current_point[0], current_point[1], 0
-        goal_x, goal_y, goal_z = goal_point[0], goal_point[1], 0
+
+        # current_x, current_y, current_z = current_point[0], current_point[1], 0
+        current_x, current_y, current_z = current_point[1], current_point[0], 0
+        # goal_x, goal_y, goal_z = goal_point[0], goal_point[1], 0
+        goal_x, goal_y, goal_z = goal_point[1], goal_point[0], 0
 
         if len(current_point) == len(goal_point) == 3:
             current_z, current_x, current_y = current_point[0], current_point[1], current_point[2]

brightest_path_lib/image/stats.py

@@ -46,11 +46,10 @@ def __init__(self, image: np.ndarray):
             self.y_max = image.shape[1] - 1
             self.x_max = image.shape[2] - 1
         elif len(image.shape) == 2:
+            # will be in the form (y, x)
             self.z_max = 0
             self.y_max = image.shape[0] - 1
             self.x_max = image.shape[1] - 1
-            # self.x_max = len(image[0]) - 1
-            # self.y_max = len(image) - 1
 
     @property
     def min_intensity(self) -> float:

brightest_path_lib/sandbox2.py

@@ -0,0 +1,104 @@
+import sys
+sys.path.append("/Users/vasudhajha/Documents/mapmanager/brightest-path-lib")
+
+from queue import Empty, Queue
+from threading import Thread
+import tifffile
+from typing import List
+
+from brightest_path_lib.algorithm import AStarSearch
+
+from skimage import data
+import numpy as np
+import matplotlib.pyplot as plt
+
+
+class AStarThread(Thread):
+    def __init__(self,
+        image : np.ndarray,
+        start_point : np.ndarray,
+        goal_point : np.ndarray,
+        queue = None):
+        super().__init__(daemon=True)
+        self.queue = queue
+        self.search_algorithm = AStarSearch(image, start_point=start_point, goal_point=goal_point, open_nodes=queue)
+
+    def cancel(self):
+        self.search_algorithm.is_canceled = True
+
+    def run(self):
+        """
+        run A* tracing algorithm
+        """
+        print("Searching...")
+        self.search_algorithm.search()
+        print("Done")
+
+
+def _plot_image(image: np.ndarray, start: np.ndarray, end: np.ndarray):
+    plt.imshow(image, cmap='gray')
+    plt.plot(start[1], start[0],'og')
+    plt.plot(end[1], end[0], 'or')
+    plt.pause(0.001)
+
+
+def _plot_points(points: List[np.ndarray], color, size, alpha=1.0):
+    """Plot points
+
+    Args:
+        points: [(y,x)]
+    """
+    yPlot = [point[0] for point in points]
+    xPlot = [point[1] for point in points]
+
+    plt.scatter(xPlot, yPlot, c=color, s=size, alpha=alpha)
+    plt.pause(0.0001)
+
+
+def plot_brightest_path():
+    # image = data.cells3d()[30, 0]
+    # start_point = np.array([0,192]) # [y, x]
+    # goal_point = np.array([198,9])
+
+    image = tifffile.imread('a-star-image.tif')
+    start_point = np.array([188, 71]) # (y,x)
+    goal_point = np.array([126, 701])
+
+    _plot_image(image, start_point, goal_point)
+
+    queue = Queue()
+    search_thread = AStarThread(image, start_point, goal_point, queue)
+    search_thread.start()  # start the thread, internally Python calls tt.run()
+
+    _updateInterval = 100  # wait for this number of results and update plot
+    plotItems = []
+    while search_thread.is_alive() or not queue.empty(): # polling the queue
+        if search_thread.search_algorithm.found_path:
+            break
+
+        try:
+            item = queue.get(False)
+            # update a matplotlib/pyqtgraph/napari interface
+            plotItems.append(item)
+            if len(plotItems) > _updateInterval:
+                _plot_points(plotItems, 'c', 8, 0.3)
+                plotItems = []
+
+        except Empty:
+            # Handle empty queue here
+            pass
+
+
+    if search_thread.search_algorithm.found_path:
+        plt.clf()
+
+        _plot_image(image, start_point, goal_point)
+
+        _plot_points(search_thread.search_algorithm.result, 'y', 4, 0.5)
+
+
+    # keep the plot up
+    plt.show()
+
+if __name__ == "__main__":
+    plot_brightest_path()

tests/test_astar.py

@@ -10,7 +10,7 @@
        [ 5745,  7845, 11113,  7820,  3551]])
 two_dim_start_point = np.array([0,0])
 two_dim_goal_point = np.array([4,4])
-two_dim_scale = np.array([1.0, 1.0])
+two_dim_scale = (1.0, 1.0)
 two_dim_result = np.array([np.array([0, 0]), np.array([0, 1]), np.array([1, 2]), np.array([2, 3]), np.array([3, 3]), np.array([4, 3]), np.array([4, 4])])
 
 three_dim_image = np.array([[[ 4496,  5212,  6863, 10113,  7055],
@@ -26,7 +26,7 @@
         [ 6117,  6022,  7160,  7113,  7066]]])
 three_dim_start_point = np.array([0,0,0])
 three_dim_goal_point = np.array([0,4,4])
-three_dim_scale = np.array([1.0, 1.0, 1.0])
+three_dim_scale = (1.0, 1.0, 1.0)
 three_dim_result = np.array([np.array([0, 0, 0]), np.array([1, 1, 0]), np.array([1, 2, 1]), np.array([0, 3, 2]), np.array([0, 4, 3]), np.array([0, 4, 4])])
 
 @pytest.mark.parametrize("image, start_point, goal_point, scale", [