working on the numpy.recarray

Author: alexlib

openptv_python/correspondences.py

@@ -102,14 +102,14 @@ def __del__(self):
 Correspond_dtype = np.dtype([
     ('p1', np.int32), # PT_UNUSED
     ('n', np.int32), # 0
-    ('p2', (np.float64, MAXCAND)), # np.zeros
+    ('p2', (np.int32, MAXCAND)), # np.zeros
     ('corr', (np.float64, MAXCAND)), # np.zeros
     ('dist', (np.float64, MAXCAND)) # np.zeros
 ])
 
 def safely_allocate_target_usage_marks(
     num_cams: int, nmax: int = NMAX
-) -> List[List[int]]:
+) -> np.ndarray: # num_cams x nmax instead of List[List[int]]:
     """Allocate space for per-camera arrays marking whether a certain target was used.
 
     If some allocation failed, it cleans up memory and returns NULL. Allocated arrays are zeroed
@@ -123,39 +123,48 @@ def safely_allocate_target_usage_marks(
     -------
         A list of lists of integers, or `None` if an allocation failed.
     """
-    tusage = []
-    for cam in range(num_cams):
-        tusage.append([0] * nmax)  # Initialize the array to all zeros.
+    # tusage = []
+    # for cam in range(num_cams):
+    #     tusage.append([0] * nmax)  # Initialize the array to all zeros.
 
-    # Check if any of the allocations failed.
-    for cam in range(num_cams):
-        if tusage[cam] is None:
-            return []  # was None
+    # # Check if any of the allocations failed.
+    # for cam in range(num_cams):
+    #     if tusage[cam] is None:
+    #         return []  # was None
 
-    return tusage
+    return np.zeros((num_cams, nmax), dtype=np.int32)
 
 def safely_allocate_adjacency_lists(
     num_cams: int, target_counts: List[int]
-) -> List[List[List[np.recarray]]]:
+) -> np.recarray:
     """Allocate space for the adjacency lists."""
-    one_element = np.array(
-        [(PT_UNUSED, 0, np.zeros(MAXCAND), np.zeros(MAXCAND), np.zeros(MAXCAND))],
-        dtype=Correspond_dtype).view(np.recarray)
+    # one_element = np.array(
+    #     [(PT_UNUSED, 0, np.zeros(MAXCAND), np.zeros(MAXCAND), np.zeros(MAXCAND))],
+    #     dtype=Correspond_dtype).view(np.recarray)
 
     try:
-        lists = [
-            [[one_element for _ in range(target_counts[c1])] for _ in range(num_cams)]
-            for c1 in range(num_cams)
-        ]
+        # lists = [
+        #     [[one_element for _ in range(target_counts[c1])] for _ in range(num_cams)]
+        #     for c1 in range(num_cams)
+        # ]
+
+        lists = np.recarray((num_cams, num_cams, max(target_counts)),dtype=Correspond_dtype)
+
+    except MemoryError as exc:
+        raise MemoryError("Failed to allocate adjacency lists.") from exc
+        # lists = [[[one_element]]]
+
+    lists.p1 = PT_UNUSED
+    lists.n = 0
+    lists.p2 = np.zeros(MAXCAND)
+    lists.corr = np.zeros(MAXCAND)
+    lists.dist = np.zeros(MAXCAND)
 
-    except MemoryError:
-        print("Memory allocation failed.")
-        lists = [[[one_element]]]
 
     return lists
 
 def four_camera_matching(
-    corr_list: List[List[List[np.recarray]]],
+    corr_list: np.recarray,
     base_target_count,
     accept_corr,
     scratch,
@@ -231,7 +240,7 @@ def four_camera_matching(
 
 
 def three_camera_matching(
-    corr_list: List[List[List[np.recarray]]],
+    corr_list: np.recarray, # num_cam, num_cam, num_targets
     num_cams,
     target_counts,
     accept_corr,
@@ -310,13 +319,13 @@ def three_camera_matching(
 
 
 def consistent_pair_matching(
-    corr_list: List[List[List[np.recarray]]],
+    corr_list: np.recarray,
     num_cams: int,
     target_counts: List[int],
     accept_corr: float,
-    scratch,
+    scratch: np.recarray,
     scratch_size: int,
-    tusage: List[List[int]],
+    tusage: np.ndarray,
 ) -> int:
     """Find consistent pairs of correspondences."""
     matched = 0
@@ -356,8 +365,8 @@ def consistent_pair_matching(
 
 
 def match_pairs(
-    corr_lists: List[List[List[np.recarray]]],
-    corrected: np.ndarray, # List[List[Coord2d]],
+    corr_lists: np.recarray, # num_cam, num_cam, num_targets
+    corrected: np.recarray, # List[List[Coord2d]],
     frm: Frame,
     vpar: VolumePar,
     cpar: ControlPar,
@@ -457,7 +466,7 @@ def match_pairs(
 
 
 def take_best_candidates(
-    src: np.recarray, dst: np.recarray, num_cams: int, tusage: List[List[int]] #List[n_tupel]
+    src: np.recarray, dst: np.recarray, num_cams: int, tusage: np.ndarray #List[n_tupel]
 ):
     """
     Take the best candidates from the candidate list based on their correlation measure.
@@ -532,7 +541,7 @@ def take_best_candidates(
 
 def py_correspondences(
     img_pts: List[List[Target]],  # num_cams * num_targets[cam]
-    flat_coords: List[List[np.recarray]],
+    flat_coords: np.recarray,
     calib: List[Calibration],
     vparam: VolumePar,
     cparam: ControlPar,
@@ -648,12 +657,12 @@ def py_correspondences(
 
 def correspondences(
     frm: Frame,
-    corrected: List[List[np.recarray]],  # List[List[Coord2d]],
+    corrected: np.recarray,  # List[List[Coord2d]],
     vpar: VolumePar,
     cpar: ControlPar,
     calib: List[Calibration],
     match_counts: List[int],
-) -> List[np.ndarray]: # n_tupel
+) -> np.recarray: # n_tupel_dtype
     """Find correspondences between cameras.
 
     /*  correspondences() generates a list of tuple target numbers (one for each
@@ -690,8 +699,8 @@ def correspondences(
     nmax = NMAX
 
     # Allocation of scratch buffers for internal tasks and return-value space
-    con0 = np.empty((nmax * cpar.num_cams,), dtype=n_tupel_dtype)
-    con = np.empty((nmax * cpar.num_cams,), dtype=n_tupel_dtype)
+    con0 = np.recarray((nmax * cpar.num_cams), dtype=n_tupel_dtype)
+    con = np.recarray((nmax * cpar.num_cams), dtype=n_tupel_dtype)
 
     tim = safely_allocate_target_usage_marks(cpar.num_cams, nmax)
 

openptv_python/find_candidate.py

@@ -6,14 +6,14 @@
 
 from .calibration import Calibration
 from .constants import MAXCAND
-from .epi import Candidate, Coord2d
+from .epi import Candidate_dtype
 from .parameters import ControlPar, VolumePar
 from .tracking_frame_buf import Target
 from .trafo import correct_brown_affine
 
 
 def find_candidate(
-    crd: List[Coord2d],
+    crd: np.recarray,
     pix: List[Target],
     num: int,
     xa: float,
@@ -27,7 +27,7 @@ def find_candidate(
     vpar: VolumePar,
     cpar: ControlPar,
     cal: Calibration,
-) -> List[Candidate]:
+) -> np.recarray:
     """Search in the image space of the image all the candidates around the epipolar line.
 
     originating from another camera. It is a binary search in an x-sorted coord-set,
@@ -57,7 +57,8 @@ def find_candidate(
     -------
         cand: list of candidates, or empty list if nothing is found
     """
-    cand: List[Candidate] = []
+    cand: List[np.recarray] = []
+    # cand: np.recarray = np.recarray((0,), dtype=Candidate_dtype)
 
     # The image space is the image plane of the camera. The image space is
     # given in millimeters of sensor size and the origin is in the center of the sensor.
@@ -89,7 +90,8 @@ def find_candidate(
 
     # If epipolar line out of sensor area, give up.
     if xb <= xmin or xa >= xmax or yb <= ymin or ya >= ymax:
-        return cand
+        # return np.array(cand).view(np.recarray)
+        return np.recarray((0,), dtype=Candidate_dtype)
 
     j0 = find_start_point(crd, num, xa, vpar)
 
@@ -99,7 +101,8 @@ def find_candidate(
         # Since the list is x-sorted, an out of x-bound candidate is after the
         # last possible candidate, so stop.
         if crd[j].x > xb + vpar.eps0:
-            return cand
+            out = np.array(cand).view(np.recarray).flatten()
+            return out  # type: ignore
 
         # Candidate should at the very least be in the epipolar search window
         # to be considered.
@@ -128,8 +131,9 @@ def find_candidate(
         if qn < vpar.cn or qnx < vpar.cnx or qny < vpar.cny or qsumg <= vpar.csumg:
             continue
         if len(cand) >= MAXCAND:
-            print(f"More candidates than {MAXCAND}: {len(cand)}\n")
-            return cand
+            print(f"Increase maximum number of candidates {len(cand)}\n")
+            out = np.array(cand).view(np.recarray).flatten()
+            return out # type: ignore
 
         # Empirical correlation coefficient from shape and brightness
         # parameters.
@@ -138,11 +142,13 @@ def find_candidate(
         # Prefer matches with brighter targets.
         corr *= sumg + pix[p2].sumg
 
-        cand.append(Candidate(pnr=j, tol=d, corr=corr))
+        # cand.append(Candidate(pnr=j, tol=d, corr=corr))
+        cand.append(np.array([(j, d, corr)], dtype=Candidate_dtype)) # type: ignore
 
+        out = np.array(cand).view(np.recarray).flatten()
         # print(f"appended: {cand[-1]}")
 
-    return cand
+    return out # type: ignore
 
 
 # def quality_ratio(a, b):
@@ -156,7 +162,7 @@ def quality_ratio(a: float, b: float) -> float:
         return 0
     return min(a, b) / max(a, b)
 
-def find_start_point(crd: List[Coord2d], num: int, xa: float, vpar: VolumePar) -> int:
+def find_start_point(crd: np.recarray, num: int, xa: float, vpar: VolumePar) -> int:
     """Find the start point of the candidate search.
 
     Args:
@@ -171,9 +177,8 @@ def find_start_point(crd: List[Coord2d], num: int, xa: float, vpar: VolumePar) -
     -------
         The start point of the candidate search.
     """
-    x = np.array([_.x for _ in crd])
-    eps0 = vpar.eps0
-    out = find_start_point_binary(x, num, xa, eps0)
+    # x = np.array([_.x for _ in crd])
+    out = find_start_point_binary(crd.x, num, xa, vpar.eps0)
     return out
 
 @njit(int32(float64[:], int32, float64, float64))

openptv_python/tracking_frame_buf.py

@@ -883,7 +883,7 @@ def match_coords(
     cal: Calibration,
     tol: float = 1e-5,
     reset_numbers: bool = False,
-) -> np.ndarray:
+) -> np.recarray:
     """Match coordinates from all cameras into a single block.
 
     replaces MatchedCoords class in Cython
@@ -898,7 +898,7 @@ def match_coords(
     is a low priority.
 
     """
-    matched_coords = np.empty(len(targs), dtype=Coord2d_dtype)
+    matched_coords = np.recarray(len(targs), dtype=Coord2d_dtype)
 
     for tnum, targ in enumerate(targs):
         # targ = targs[tnum]
@@ -915,7 +915,7 @@ def match_coords(
 
 
 def matched_coords_as_arrays(
-    matched_coords: List[np.recarray], #Coord2d
+    matched_coords: np.recarray, #Coord2d
 ) -> Tuple[np.ndarray, np.ndarray]:
     """
     Return the data associated with the object (the matched coordinates.
@@ -927,7 +927,7 @@ def matched_coords_as_arrays(
     pos - (n,2) array, the (x,y) flat-coordinates position of n targets.
     pnr - n-length array, the corresponding target number for each point.
     """
-    num_pts = len(matched_coords)
+    num_pts = matched_coords.shape[0]
     pos = np.empty((num_pts, 2))
     pnr = np.empty(num_pts, dtype=np.int_)