namedtuple to rescue numba jit

Author: alexlib

openptv_python/parameters.py

@@ -1,4 +1,5 @@
 """Parameters for OpenPTV-Python."""
+from collections import namedtuple
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import List, Tuple
@@ -7,6 +8,21 @@
 
 from openptv_python.constants import TR_MAX_CAMS
 
+TrackParTuple = namedtuple('TrackParTuple',
+                           ['dvxmin',
+                            'dvxmax',
+                            'dvymin',
+                            'dvymax',
+                            'dvzmin',
+                            'dvzmax',
+                            'dangle',
+                            'dacc',
+                            'add',
+                            'dsumg',
+                            'dn',
+                            'dnx',
+                            'dny'])
+
 
 @dataclass
 class Parameters:
@@ -193,24 +209,6 @@ class TrackPar(Parameters):
     dny: float = 0.0
 
 
-    # def to_dict(self):
-    #     """Convert TrackPar instance to a dictionary."""
-    #     return {
-    #         'dvxmax': self.dvxmax,
-    #         'dvxmin': self.dvxmin,
-    #         'dvymax': self.dvymax,
-    #         'dvymin': self.dvymin,
-    #         'dvzmax': self.dvzmax,
-    #         'dvzmin': self.dvzmin,
-    #         'dangle': self.dangle,
-    #         'dacc': self.dacc,
-    #         'add': self.add,
-    #         'dsumg': self.dsumg,
-    #         'dn': self.dn,
-    #         'dnx': self.dnx,
-    #         'dny': self.dny,
-    #     }
-
     @classmethod
     def from_file(cls, filename: Path):
         """Read tracking parameters from file and return TrackPar object.
@@ -315,6 +313,24 @@ def compare_track_par(t1: TrackPar, t2: TrackPar) -> bool:
     return all(getattr(t1, field) == getattr(t2, field) for field in t1.__annotations__)
 
 
+def convert_track_par_to_tuple(track_par: TrackPar) -> TrackParTuple:
+    """Convert TrackPar object to TrackParTuple object."""
+    return TrackParTuple(track_par.dvxmin,
+                         track_par.dvxmax,
+                         track_par.dvymin,
+                         track_par.dvymax,
+                         track_par.dvzmin,
+                         track_par.dvzmax,
+                         track_par.dangle,
+                         track_par.dacc,
+                         track_par.add,
+                         track_par.dsumg,
+                         track_par.dn,
+                         track_par.dnx,
+                         track_par.dny)
+
+
+
 @dataclass
 class VolumePar(Parameters):
     """Volume parameters."""

openptv_python/track.py

@@ -21,7 +21,7 @@
 )
 from .imgcoord import img_coord
 from .orientation import point_position
-from .parameters import ControlPar, TrackPar
+from .parameters import ControlPar, TrackParTuple, convert_track_par_to_tuple
 from .tracking_frame_buf import Frame, Pathinfo, Target
 from .tracking_run import TrackingRun
 from .trafo import dist_to_flat, metric_to_pixel, pixel_to_metric
@@ -188,8 +188,8 @@ def predict(prev_pos, curr_pos, output):
     output[0] = 2 * curr_pos[0] - prev_pos[0]
     output[1] = 2 * curr_pos[1] - prev_pos[1]
 
-
-def pos3d_in_bounds(pos: np.ndarray, bounds: TrackPar) -> bool:
+@njit(cache=True, fastmath=True, nogil=True)
+def pos3d_in_bounds(pos: np.ndarray, bounds: TrackParTuple) -> bool:
     """Check that all components of a pos3d are in their respective bounds.
 
     taken from a track_par object.
@@ -211,44 +211,6 @@ def pos3d_in_bounds(pos: np.ndarray, bounds: TrackPar) -> bool:
     )
 
 
-# def angle_acc(
-#     start: np.ndarray, pred: np.ndarray, cand: np.ndarray
-# ) -> Tuple[float, float]:
-#     """Calculate the angle between the (1st order) numerical velocity vectors.
-
-#     to the predicted next_frame position and to the candidate actual position. The
-#     angle is calculated in [gon], see [1]. The predicted position is the
-#     position if the particle continued at current velocity.
-
-#     Arguments:
-#     ---------
-#     start -- vec3d, the particle start position
-#     pred -- vec3d, predicted position
-#     cand -- vec3d, possible actual position
-
-#     Returns:
-#     -------
-#     angle -- float, the angle between the two velocity vectors, [gon]
-#     acc -- float, the 1st-order numerical acceleration embodied in the deviation from prediction.
-#     """
-#     v0 = pred - start
-#     v1 = cand - start
-
-#     acc = math.dist(v0, v1)
-#     # acc = np.linalg.norm(v0 - v1)
-
-#     if np.all(v0 == -v1):
-#         angle = 200
-#     elif np.all(v0 == v1):
-#         angle = 0
-#     else:
-#         angle = float((200.0 / math.pi) * math.acos(
-#             math.fsum([v0[i] * v1[i] for i in range(3)])
-#             / (math.dist(start, pred) * math.dist(start, cand)))
-#         )
-
-#     return angle, acc
-
 @njit(float64[:](float64[:], float64[:], float64[:]), cache=True, fastmath=True, nogil=True, parallel=True)
 def angle_acc(
     start: np.ndarray, pred: np.ndarray, cand: np.ndarray
@@ -441,7 +403,7 @@ def candsearch_in_pix_rest(
 
 
 def searchquader(
-    point: np.ndarray, tpar: TrackPar, cpar: ControlPar, cal: List[Calibration]
+    point: np.ndarray, tpar: TrackParTuple, cpar: ControlPar, cal: List[Calibration]
 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
     """Calculate the search volume in image space."""
     mins = np.array([tpar.dvxmin, tpar.dvymin, tpar.dvzmin])
@@ -819,7 +781,7 @@ def trackcorr_c_loop(run_info, step):
 
     fb = run_info.fb
     cal = run_info.cal
-    tpar = run_info.tpar
+    tpar = convert_track_par_to_tuple(run_info.tpar)
     vpar = run_info.vpar
     cpar = run_info.cpar
     curr_targets = fb.buf[1].targets

openptv_python/tracking_run.py

@@ -11,8 +11,9 @@
 from .parameters import (
     ControlPar,
     SequencePar,
-    TrackPar,
+    TrackParTuple,
     VolumePar,
+    convert_track_par_to_tuple,
     read_control_par,
     read_sequence_par,
     read_track_par,
@@ -26,7 +27,7 @@ class TrackingRun:
 
     fb: FrameBuf
     seq_par: SequencePar
-    tpar: TrackPar
+    tpar: TrackParTuple
     vpar: VolumePar
     cpar: ControlPar
     cal: List[Calibration]
@@ -40,7 +41,7 @@ class TrackingRun:
     def __init__(
         self,
         seq_par: SequencePar,
-        tpar: TrackPar,
+        tpar: TrackParTuple,
         vpar: VolumePar,
         cpar: ControlPar,
         buf_len: int,
@@ -113,7 +114,7 @@ def tr_new(
     """Create a new tracking run from legacy files."""
     cpar = read_control_par(cpar_fname)
     seq_par = read_sequence_par(seq_par_fname, cpar.num_cams)
-    tpar = read_track_par(tpar_fname)
+    tpar = convert_track_par_to_tuple(read_track_par(tpar_fname))
     vpar = read_volume_par(vpar_fname)
 
     tr = TrackingRun(

profile

@@ -138,7 +138,8 @@ def test_burgers(self):
             10000.0,
         )
 
-        run.tpar.add = 1
+        # run.tpar = run.tpar._replace(add=1)
+        run.tpar = run.tpar._replace(add=1)
         print("changed add particle to", run.tpar.add)
 
         track_forward_start(run)

tests/test_burgers.py

@@ -0,0 +1,26 @@
+from collections import namedtuple
+
+import numpy as np
+from numba import njit
+
+# Define a namedtuple
+Point = namedtuple('Point', ['x', 'y'])
+
+# Create a list of points
+points = [Point(x, y) for x in range(1000) for y in range(1000)]
+
+# Calculate the distance between each point and the origin
+@njit
+def distance_to_origin(point):
+    """Calculate the distance between a point and the origin."""
+    return np.sqrt(point.x**2 + point.y**2)
+
+
+# Calculate the distance between each point and the origin using the function
+distances = np.array([distance_to_origin(point) for point in points])
+
+# Calculate the distance between each point and the origin manually
+expected_distances = np.array([np.sqrt(point.x**2 + point.y**2) for point in points])
+
+# Compare the results
+np.testing.assert_allclose(distances, expected_distances)

tests/test_numba_namedtuple.py

@@ -20,6 +20,7 @@
 from openptv_python.parameters import (
     ControlPar,
     TrackPar,
+    convert_track_par_to_tuple,
 )
 from openptv_python.track import (
     Foundpix_dtype,
@@ -122,7 +123,8 @@ def test_pos3d_in_bounds(self):
         inside = np.array([1.0, -1.0, 0.0])
         outside = np.array([2.0, -0.8, 2.1])
 
-        bounds = TrackPar(
+        bounds = convert_track_par_to_tuple(
+            TrackPar(
             -2.0,
             2.0,
             -2.0,
@@ -137,6 +139,7 @@ def test_pos3d_in_bounds(self):
             0.0,
             0.0,
         )
+        )
 
         result = pos3d_in_bounds(inside, bounds)
 
@@ -361,9 +364,11 @@ def test_searchquader(self):
 
         #  print(f"cpar = {self.cpar}")
 
-        tpar = TrackPar(
+        tpar = convert_track_par_to_tuple(
+            TrackPar(
             0.2, -0.2, 0.1, -0.1, 0.1, -0.1, 120, 0.4, 1, 0.0, 0.0, 0.0, 0.0
         )
+        )
         xr, xl, yd, yu = searchquader(point, tpar, self.cpar, self.calib)
 
         # print(f"xr = {xr}, xl = {xl}, yd = {yd}, yu = {yu}")
@@ -379,9 +384,11 @@ def test_searchquader(self):
 
         # Let's test with just one camera to check borders
         self.cpar.num_cams = 1
-        tpar1 = TrackPar(
+        tpar1 = convert_track_par_to_tuple(
+            TrackPar(
             0.0, -0.0, 0.0, -0.0, 0.0, -0.0, 120, 0.4, 1, 0.0, 0.0, 0.0, 0.0
         )
+        )
         xr, xl, yd, yu = searchquader(point, tpar1, self.cpar, self.calib)
 
         # print(f"xr = {xr}, xl = {xl}, yd = {yd}, yu = {yu}")
@@ -391,7 +398,8 @@ def test_searchquader(self):
         )
 
         # Test with infinitely large values of tpar that should return about half the image size
-        tpar2 = TrackPar(
+        tpar2 = convert_track_par_to_tuple(
+            TrackPar(
             1000.0,
             -1000.0,
             1000.0,
@@ -406,6 +414,7 @@ def test_searchquader(self):
             0.0,
             0.0,
         )
+        )
 
         xr, xl, yd, yu = searchquader(point, tpar2, self.cpar, self.calib)
 

tests/test_tracking.py

@@ -125,7 +125,8 @@ def test_trackcorr_no_add(self):
             calib,
             10000.0,
         )
-        run.tpar.add = 0
+
+        run.tpar = run.tpar._replace(add = 0)
         print(f"run.seq_par.first = {run.seq_par.first} run.seq_par.last = {run.seq_par.last}")
 
         track_forward_start(run)
@@ -198,7 +199,7 @@ def test_trackcorr_add(self):
 
         run.seq_par.first = 10240
         run.seq_par.last = 10250
-        run.tpar.add = 1
+        run.tpar = run.tpar._replace(add=1)
 
         track_forward_start(run)
         trackcorr_c_loop(run, run.seq_par.first)
@@ -274,7 +275,7 @@ def test_trackback(self):
 
         run.seq_par.first = 10240
         run.seq_par.last = 10250
-        run.tpar.add = 1
+        run.tpar = run.tpar._replace(add=1)
 
         track_forward_start(run)
         trackcorr_c_loop(run, run.seq_par.first)
@@ -285,8 +286,14 @@ def test_trackback(self):
         trackcorr_c_finish(run, run.seq_par.last)
 
 
-        run.tpar.dvxmin = run.tpar.dvymin = run.tpar.dvzmin = -50.0
-        run.tpar.dvxmax = run.tpar.dvymax = run.tpar.dvzmax = 50.0
+        run.tpar = run.tpar._replace(
+            dvxmin = -50,
+            dvymin = -50,
+            dvzmin = -50.0,
+            dvxmax = 50.0,
+            dvymax = 50.0,
+            dvzmax = 50.0,
+        )
 
 
         run.lmax = vec_norm(
@@ -356,7 +363,7 @@ def test_new_particle(self):
             0.1,
         )
 
-        run.tpar.add = 0
+        run.tpar = run.tpar._replace(add = 0)
 
         track_forward_start(run)
         trackcorr_c_loop(run, 10001)
@@ -381,7 +388,7 @@ def test_new_particle(self):
         #     calib,
         #     0.1,
         # )
-        run.tpar.add = 1
+        run.tpar = run.tpar._replace(add=1)
         track_forward_start(run)
         trackcorr_c_loop(run, 10001)
         trackcorr_c_loop(run, 10002)

tests/test_tracking_run.py

@@ -131,7 +131,7 @@ class TestCavity(unittest.TestCase):
             10000.0,
         )
 
-        run.tpar.add = 1
+        run.tpar = run.tpar._replace(add=1)
         print("changed add particle to", run.tpar.add)
 
         track_forward_start(run)