diff --git a/openptv_python/correspondences.py b/openptv_python/correspondences.py index fcb2462..843ed6b 100644 --- a/openptv_python/correspondences.py +++ b/openptv_python/correspondences.py @@ -80,7 +80,7 @@ def safely_allocate_adjacency_lists( for c1 in range(num_cams - 1): for c2 in range(c1 + 1, num_cams): if error == 0: - lists[c1][c2] = [Correspond() for _ in range(target_counts[c1])] + lists[c1][c2] = [Correspond() for _ in range(target_counts[c1])] # type: ignore if lists[c1][c2] is None: error = 1 lists[c1][c2] = None diff --git a/openptv_python/parameters.py b/openptv_python/parameters.py index 44e4505..b2612cc 100644 --- a/openptv_python/parameters.py +++ b/openptv_python/parameters.py @@ -4,6 +4,8 @@ from pathlib import Path from typing import List, Tuple +from openptv_python.constants import TR_MAX_CAMS + @dataclass class MultimediaPar: @@ -546,16 +548,16 @@ def read_target_par(filename: str) -> TargetPar | None: ret = TargetPar() try: with open(filename, "r", encoding="utf-8") as file: - for i in range(4): # todo - make it no. cameras + for _ in range(TR_MAX_CAMS): # todo - make it no. cameras ret.gvthresh.append(int(file.readline())) ret.discont = int(file.readline()) - ret.nnmin = float(file.readline()) - ret.nnmax = float(file.readline()) - ret.nxmin = float(file.readline()) - ret.nxmax = float(file.readline()) - ret.nymin = float(file.readline()) - ret.nymax = float(file.readline()) + ret.nnmin = int(file.readline()) + ret.nnmax = int(file.readline()) + ret.nxmin = int(file.readline()) + ret.nxmax = int(file.readline()) + ret.nymin = int(file.readline()) + ret.nymax = int(file.readline()) ret.sumg_min = int(file.readline()) ret.cr_sz = int(file.readline()) return ret diff --git a/openptv_python/segmentation.py b/openptv_python/segmentation.py index 5640115..9ad6d9d 100644 --- a/openptv_python/segmentation.py +++ b/openptv_python/segmentation.py @@ -2,27 +2,28 @@ from typing import List import numpy as np +from scipy.ndimage import center_of_mass, gaussian_filter, label, maximum_filter -from .constants import CORRES_NONE, NMAX +from .constants import CORRES_NONE from .parameters import ControlPar, TargetPar -from .tracking_frame_buf import Target, TargetArray +from .tracking_frame_buf import Target @dataclass class Peak: """Peak dataclass.""" - pos: int | None = None - status: int | None = None - xmin: int | None = None - xmax: int | None = None - ymin: int | None = None - ymax: int | None = None - n: int | None = None - sumg: int | None = None - x: float | None = None - y: float | None = None - unr: int | None = None + pos: int = 0 + status: int = 0 + xmin: int = 0 + xmax: int = 0 + ymin: int = 0 + ymax: int = 0 + n: int = 0 + sumg: int = 0 + x: float = 0.0 + y: float = 0.0 + unr: int = 0 touch: list[int] = field(default_factory=list, repr=False) n_touch: int = 0 @@ -36,7 +37,7 @@ def targ_rec( ymax: int, cpar: ControlPar, num_cam, -) -> TargetArray: +) -> List[Target]: """Target recognition function.""" n = 0 n_wait = 0 @@ -141,6 +142,7 @@ def targ_rec( and (gvref + disco >= img[yn, xn - 1]) and (gvref + disco >= img[yn, xn + 1]) ): + print(f"gv = {gv} sumg = {sumg}") sumg += gv img0[yn, xn] = 0 if xn < xa: @@ -188,11 +190,7 @@ def targ_rec( and ny <= targ_par.nymax and sumg > targ_par.sumg_min ): - pix.append(Target()) - pix[n_targets].n = numpix - pix[n_targets].nx = nx - pix[n_targets].ny = ny - pix[n_targets].sumg = sumg + pix.append(Target(n=numpix, nx=nx, ny=ny, sumg=sumg)) sumg -= numpix * thres # finish the grey-value weighting: x /= sumg @@ -222,17 +220,15 @@ def peak_fit( ymax: int, cpar: ControlPar, num_cam: int, -) -> TargetArray: +) -> List[Target]: """Fit the peaks in the image to a gaussian.""" imx, imy = cpar.imx, cpar.imy n_peaks = 0 n_wait = 0 x8, y8 = [0, 1, 0, -1], [1, 0, -1, 0] p2 = 0 - thres = targ_par.gvthres[num_cam] + thres = targ_par.gvthresh[num_cam] disco = targ_par.discont - pnr, sumg, xn, yn = 0, 0, 0, 0 - n_target = 0 intx1, inty1 = 0, 0 unify = 0 unified = 0 @@ -241,12 +237,10 @@ def peak_fit( gv1, gv2 = 0, 0 x1, x2, y1, y2, s12 = 0.0, 0.0, 0.0, 0.0, 0.0 label_img = [0] * (imx * imy) - # nmax = 1024 - nmax = NMAX - peaks = [0] * (4 * nmax) - ptr_peak = Peak() + peaks = [] waitlist = [[]] pix = [] + n_target = 0 for i in range(ymin, ymax - 1): for j in range(xmin, xmax): @@ -275,21 +269,7 @@ def peak_fit( # label peak in label_img, initialize peak n_peaks += 1 label_img[n] = n_peaks - ptr_peak.pos = n - ptr_peak.status = 1 - ptr_peak.xmin = j - ptr_peak.xmax = j - ptr_peak.ymin = i - ptr_peak.ymax = i - ptr_peak.unr = 0 - ptr_peak.n = 0 - ptr_peak.sumg = 0 - ptr_peak.x = 0 - ptr_peak.y = 0 - ptr_peak.n_touch = 0 - for k in range(4): - ptr_peak.touch[k] = 0 - ptr_peak += 1 + peaks.append(Peak(pos=n, status=1, xmin=j, xmax=i, ymin=i, ymax=i)) waitlist[0][0] = j waitlist[0][1] = i @@ -498,10 +478,10 @@ def peak_fit( pix[n_target].pnr = n_target n_target += 1 - t = TargetArray() - t.num_targs = n_target - t.targs = pix - return t + # t = TargetArray(num_targets=n_target) + # t.num_targs = n_target + # t.append(pix) + return pix def check_touch(tpeak, p1, p2): @@ -544,12 +524,10 @@ def peak_fit_new( ------- List[Peak]: A list of Peak objects representing the detected peaks. """ - from scipy.ndimage import center_of_mass, gaussian_filter, label, maximum_filter - smoothed = gaussian_filter(image, sigma) - mask = smoothed > threshold * np.max(smoothed) + mask = smoothed > threshold * np.max(smoothed) # type: ignore maxima = maximum_filter(smoothed, footprint=np.ones((3, 3))) == smoothed - labeled, num_objects = label(maxima) + labeled, num_objects = label(maxima) # type: ignore peaks = [] for i in range(num_objects): indices = np.argwhere(labeled == i + 1) @@ -560,24 +538,6 @@ def peak_fit_new( return peaks -def test_peak_fit_new(): - import matplotlib.pyplot as plt - - # Generate test image - x, y = np.meshgrid(np.linspace(-5, 5, 100), np.linspace(-5, 5, 100)) - z = np.sin(np.sqrt(x**2 + y**2)) - - # Find peaks - peaks = peak_fit(z) - - # Plot image and detected peaks - fig, ax = plt.subplots() - ax.imshow(z, cmap="viridis") - for peak in peaks: - ax.scatter(peak.y, peak.x, marker="x", c="r", s=100) - plt.show() - - def target_recognition( img: np.ndarray, tpar: TargetPar, @@ -585,7 +545,7 @@ def target_recognition( cparam: ControlPar, subrange_x=None, subrange_y=None, -) -> TargetArray: +) -> List[Target]: """ Detect targets (contiguous bright blobs) in an image. diff --git a/openptv_python/track.py b/openptv_python/track.py index be9e4de..d1840bc 100644 --- a/openptv_python/track.py +++ b/openptv_python/track.py @@ -90,8 +90,18 @@ def copy_foundpix_array( def register_closest_neighbs( - targets, num_targets, cam, cent_x, cent_y, dl, dr, du, dd, reg, cpar -): + targets: List[Target], + num_targets: int, + cam: int, + cent_x: float, + cent_y: float, + dl: float, + dr: float, + du: float, + dd: float, + reg: List[Foundpix], + cpar: ControlPar, +) -> List[int]: """Register_closest_neighbs() finds candidates for continuing a particle's. path in the search volume, and registers their data in a foundpix array @@ -111,21 +121,23 @@ def register_closest_neighbs( reg -- an array of foundpix objects, one for each possible neighbour. Output array. cpar -- control parameter object """ - all_cands = [-999] * MAX_CANDS # Initialize all candidate indexes to -999 + # all_cands = [-999] * MAX_CANDS # Initialize all candidate indexes to -999 - candsearch_in_pix( - targets, num_targets, cent_x, cent_y, dl, dr, du, dd, all_cands, cpar + all_cands = candsearch_in_pix( + targets, num_targets, cent_x, cent_y, dl, dr, du, dd, cpar ) for cand_idx in range(MAX_CANDS): # Set default value for unused foundpix objects - if all_cands[cand_idx] == -999: + if all_cands[cand_idx] == TR_UNUSED: reg[cand_idx].ftnr = TR_UNUSED else: # Register candidate data in the foundpix object reg[cand_idx].whichcam[cam] = 1 reg[cand_idx].ftnr = targets[all_cands[cand_idx]].tnr + return all_cands + def search_volume_center_moving(prev_pos, curr_pos, output): """Find the position of the center of the search volume for a moving. @@ -238,13 +250,13 @@ def candsearch_in_pix( dr: float, du: float, dd: float, - p: List[int], cpar: ControlPar, -) -> int: +) -> List[int]: """Search for a nearest candidate in unmatched target list.""" counter = 0 dmin = 1e20 - p1 = p2 = p3 = p4 = -1 + p1 = p2 = p3 = p4 = TR_UNUSED + p = [-1] * MAX_CANDS d1 = d2 = d3 = d4 = dmin xmin, xmax, ymin, ymax = cent_x - dl, cent_x + dr, cent_y - du, cent_y + dd @@ -305,11 +317,12 @@ def candsearch_in_pix( # print("from inside p = ", p) + # TODO: check why we need counter for j in range(4): if p[j] != -1: counter += 1 - return counter + return p def candsearch_in_pix_rest( @@ -561,7 +574,7 @@ def sorted_candidates_in_volume(center, center_proj, frm, run): right[cam], up[cam], down[cam], - points[cam * MAX_CANDS], + points, run.cpar, ) diff --git a/openptv_python/tracking_frame_buf.py b/openptv_python/tracking_frame_buf.py index b6f950e..bc55f6c 100644 --- a/openptv_python/tracking_frame_buf.py +++ b/openptv_python/tracking_frame_buf.py @@ -168,7 +168,7 @@ def read_targets(file_base: str, frame_num: int) -> List[Target]: # file_base = file_base.split(".")[0] if frame_num > 0: - filename = f"{file_base}{frame_num}_targets" + filename = f"{file_base}{frame_num:04d}_targets" else: filename = f"{file_base}_targets" @@ -816,35 +816,41 @@ def write_path_frame( success = False try: - with open(corres_fname, "w", encoding="utf8") as corres_file: - corres_file.write(f"{num_parts}\n") - with open(linkage_fname, "w", encoding="utf8") as linkage_file: - linkage_file.write(f"{num_parts}\n") + corres_file = open(corres_fname, "w", encoding="utf8") + corres_file.write(f"{num_parts}\n") - if prio_file_base is not None: - with open(prio_fname, "w", encoding="utf8") as prio_file: # type: ignore - prio_file.write(f"{num_parts}\n") + linkage_file = open(linkage_fname, "w", encoding="utf8") + linkage_file.write(f"{num_parts}\n") + + if prio_file_base is not None: + prio_file = open(prio_fname, "w", encoding="utf8") # type: ignore + prio_file.write(f"{num_parts}\n") - for pix in range(num_parts): - linkage_file.write( + for pix in range(num_parts): + linkage_file.write( + f"{path_buf[pix].prev_frame} {path_buf[pix].next_frame} " + f"{path_buf[pix].x[0]:.3f} {path_buf[pix].x[1]:.3f} " + f"{path_buf[pix].x[2]:.3f}\n" + ) + + corres_file.write( + f"{pix + 1} {path_buf[pix].x[0]:.3f} " + f"{path_buf[pix].x[1]:.3f} {path_buf[pix].x[2]:.3f} " + f"{cor_buf[pix].p[0]} {cor_buf[pix].p[1]} " + f"{cor_buf[pix].p[2]} {cor_buf[pix].p[3]}\n" + ) + + if prio_file_base is not None: + prio_file.write( f"{path_buf[pix].prev_frame} {path_buf[pix].next_frame} " f"{path_buf[pix].x[0]:.3f} {path_buf[pix].x[1]:.3f} " - f"{path_buf[pix].x[2]:.3f}\n" - ) - - corres_file.write( - f"{pix + 1} {path_buf[pix].x[0]:.3f} " - f"{path_buf[pix].x[1]:.3f} {path_buf[pix].x[2]:.3f} " - f"{cor_buf[pix].p[0]} {cor_buf[pix].p[1]} " - f"{cor_buf[pix].p[2]} {cor_buf[pix].p[3]}\n" + f"{path_buf[pix].x[2]:.3f} {path_buf[pix].prio}\n" ) - if prio_file_base: - prio_file.write( - f"{path_buf[pix].prev_frame} {path_buf[pix].next_frame} " - f"{path_buf[pix].x[0]:.3f} {path_buf[pix].x[1]:.3f} " - f"{path_buf[pix].x[2]:.3f} {path_buf[pix].prio}\n" - ) + corres_file.close() + linkage_file.close() + if prio_file_base is not None: + prio_file.close() success = True diff --git a/openptv_python/tracking_run.py b/openptv_python/tracking_run.py index cc42ee3..8ae78f7 100644 --- a/openptv_python/tracking_run.py +++ b/openptv_python/tracking_run.py @@ -64,7 +64,7 @@ def __init__( corres_file_base, linkage_file_base, prio_file_base, - cpar.img_base_name, + seq_par.img_base_name, ) self.lmax = math.sqrt( diff --git a/tests/test_segmentation.py b/tests/test_segmentation.py index 27199f5..922d45c 100644 --- a/tests/test_segmentation.py +++ b/tests/test_segmentation.py @@ -122,6 +122,24 @@ def test_one_targets2(self): self.assertEqual(len(target_array), 1) self.assertEqual(target_array[0].count_pixels(), (4, 3, 2)) + # def test_peak_fit_new(): + # """ Test the peak_fit function.""" + # import matplotlib.pyplot as plt + + # # Generate test image + # x, y = np.meshgrid(np.linspace(-5, 5, 100), np.linspace(-5, 5, 100)) + # z = np.sin(np.sqrt(x**2 + y**2)) + + # # Find peaks + # peaks = peak_fit(z) + + # # Plot image and detected peaks + # _, ax = plt.subplots() + # ax.imshow(z, cmap="viridis") + # for peak in peaks: + # ax.scatter(peak.y, peak.x, marker="x", c="r", s=100) + # plt.show() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tracking.py b/tests/test_tracking.py index 1343300..7efe269 100644 --- a/tests/test_tracking.py +++ b/tests/test_tracking.py @@ -129,15 +129,18 @@ def test_pos3d_in_bounds(self): result = pos3d_in_bounds(inside, bounds) - self.assertEqual(result, 1, "Expected True but found %s" % result) + self.assertEqual(result, 1, f"Expected True but found {result}") result = pos3d_in_bounds(outside, bounds) - self.assertEqual(result, 0, "Expected False but found %s" % result) + self.assertEqual(result, 0, f"Expected False but found {result}") class TestAngleAcc(unittest.TestCase): + """Test the angle_acc function.""" + def test_angle_acc(self): + """Test the angle_acc function.""" start = np.array([0.0, 0.0, 0.0]) pred = np.array([1.0, 1.0, 1.0]) cand = np.array([1.1, 1.0, 1.0]) @@ -215,10 +218,12 @@ def test_candsearch_in_pix(self): test_cpar.mm.n3 = 1.33 test_cpar.mm.d[0] = 5 - counter = candsearch_in_pix( - test_targets, num_targets, cent_x, cent_y, dl, dr, du, dd, p, test_cpar + p = candsearch_in_pix( + test_targets, num_targets, cent_x, cent_y, dl, dr, du, dd, test_cpar ) - # print(f"p = {p}, counter = {counter}") + counter = len(p) - p.count(-1) + + print(f"p = {p}, counter = {counter}") self.assertEqual(counter, 2) @@ -227,10 +232,12 @@ def test_candsearch_in_pix(self): dl = dr = du = dd = 10.2 p = [-1] * num_cams # Initialize p with zeros - counter = candsearch_in_pix( - test_targets, num_targets, cent_x, cent_y, dl, dr, du, dd, p, test_cpar + p = candsearch_in_pix( + test_targets, num_targets, cent_x, cent_y, dl, dr, du, dd, test_cpar ) - # print(f"p = {p}, counter = {counter}") + counter = len(p) - p.count(-1) + print(f"p = {p}, counter = {counter}") + self.assertEqual(counter, 4) test_targets = [ diff --git a/tests/test_tracking_run.py b/tests/test_tracking_run.py index 2fa423a..207b27a 100644 --- a/tests/test_tracking_run.py +++ b/tests/test_tracking_run.py @@ -8,6 +8,7 @@ @author: yosef """ +import os import shutil import unittest from pathlib import Path @@ -82,17 +83,18 @@ class TestTrackCorrNoAdd(unittest.TestCase): def test_trackcorr_no_add(self): """Test tracking without adding particles.""" - import os + # import os - current_directory = os.getcwd() - directory = "tests/testing_fodder/track" + # current_directory = Path.cwd() + # print(f"working from {current_directory}") + # directory = Path("tests/testing_fodder/track") - os.chdir(directory) + # os.chdir(directory) print(os.path.abspath(os.curdir)) - copy_directory("res_orig", "res") - copy_directory("img_orig", "img") + copy_directory("res_orig/", "res/") + copy_directory("img_orig/", "img/") print("----------------------------") print("Test tracking multiple files 2 cameras, 1 particle") @@ -101,7 +103,7 @@ def test_trackcorr_no_add(self): calib = read_all_calibration(cpar.num_cams) run = tr_new( - "/parameters/sequence.par", + "parameters/sequence.par", "parameters/track.par", "parameters/criteria.par", "parameters/ptv.par", @@ -121,7 +123,10 @@ def test_trackcorr_no_add(self): trackcorr_c_loop(run, run.seq_par.first) for step in range(run.seq_par.first + 1, run.seq_par.last): + print(f"step = {step}") trackcorr_c_loop(run, step) + print(f"run.npart = {run.npart}") + trackcorr_c_finish(run, run.seq_par.last) remove_directory("res/") @@ -134,8 +139,15 @@ def test_trackcorr_no_add(self): self.assertAlmostEqual(npart, 0.8, delta=EPS) self.assertAlmostEqual(nlinks, 0.8, delta=EPS) - os.chdir(current_directory) + # os.chdir(current_directory) if __name__ == "__main__": + current_directory = Path.cwd() + print(f"working from {current_directory}") + directory = Path("tests/testing_fodder/track") + + os.chdir(directory) unittest.main() + + os.chdir(current_directory)