more tests pass. working on optv

Author: alexlib

src/pyptv/ground_truth.py

@@ -16,10 +16,7 @@
 
 import numpy as np
 
-from openptv_python.calibration import Calibration
-from openptv_python.imgcoord import image_coordinates
-from openptv_python.trafo import arr_metric_to_pixel as convert_arr_metric_to_pixel
-
+from ._backend import Calibration, image_coordinates, convert_arr_metric_to_pixel
 from .parameter_manager import ParameterManager
 from . import ptv
 

src/pyptv/pyptv_batch.py

@@ -27,6 +27,8 @@
 import time
 from typing import Union
 
+import yaml
+
 from ._backend import get_backend_reason, set_engine
 from .ptv import py_start_proc_c, py_trackcorr_init, py_sequence_loop, generate_short_file_bases
 from .experiment import Experiment
@@ -64,21 +66,31 @@ def validate_experiment_setup(yaml_file: Path) -> Path:
 
     # Get experiment directory (parent of YAML file)
     exp_path = yaml_file.parent
-    
-    # Check for required subdirectories relative to YAML file location
-    # Note: 'res' directory is created automatically if missing
-    # required_dirs = ["img", "cal"]
-    # missing_dirs = []
-    
-    # for dir_name in required_dirs:
-    #     dir_path = exp_path / dir_name
-    #     if not dir_path.exists():
-    #         missing_dirs.append(dir_name)
-    
-    # if missing_dirs:
-    #     raise ProcessingError(
-    #         f"Missing required directories relative to {yaml_file}: {', '.join(missing_dirs)}"
-    #     )
+
+    with yaml_file.open("r", encoding="utf-8") as file:
+        data = yaml.safe_load(file) or {}
+
+    required_dirs = set()
+    for section_name in ("sequence", "ptv"):
+        section = data.get(section_name, {}) or {}
+        for key in ("base_name", "img_name", "img_cal"):
+            values = section.get(key, [])
+            if isinstance(values, (str, Path)):
+                values = [values]
+            for value in values:
+                if not value:
+                    continue
+                required_dirs.add(Path(value).parent)
+
+    if not required_dirs:
+        required_dirs.add(Path("img"))
+
+    missing_dirs = [str(dir_name) for dir_name in sorted(required_dirs) if not (exp_path / dir_name).exists()]
+
+    if missing_dirs:
+        raise ProcessingError(
+            f"Missing required directories relative to {yaml_file}: {', '.join(missing_dirs)}"
+        )
 
     return exp_path
 
@@ -119,6 +131,8 @@ def run_batch(
         # Load parameters from YAML file
         print(f"Loading parameters from: {yaml_file}")
         experiment.pm.from_yaml(yaml_file)
+        if mode in ("both", "sequence"):
+            experiment.pm.parameters.setdefault("pft_version", {})["Existing_Target"] = 0
         configured_engine = engine or experiment.pm.parameters.get("engine", "optv")
         experiment.pm.parameters["engine"] = configured_engine
         set_engine(configured_engine)

src/pyptv/pyptv_batch_parallel.py

@@ -31,6 +31,8 @@
 from concurrent.futures import ProcessPoolExecutor, as_completed
 from typing import Union, List, Tuple
 
+import yaml
+
 from ._backend import get_backend_reason, set_engine
 from .ptv import py_start_proc_c, py_sequence_loop, generate_short_file_bases
 from .experiment import Experiment
@@ -86,6 +88,7 @@ def run_sequence_chunk(
 
         # Load parameters from YAML file
         experiment.pm.from_yaml(yaml_file)
+        experiment.pm.parameters.setdefault("pft_version", {})["Existing_Target"] = 0
         configured_engine = engine or experiment.pm.parameters.get("engine", "optv")
         experiment.pm.parameters["engine"] = configured_engine
         set_engine(configured_engine)
@@ -193,16 +196,27 @@ def validate_experiment_setup(yaml_file: Path) -> Path:
 
     # Get experiment directory (parent of YAML file)
     exp_path = yaml_file.parent
-    
-    # Check for required subdirectories relative to YAML file location
-    required_dirs = ["img", "cal"]  # res is created automatically
-    missing_dirs = []
-    
-    for dir_name in required_dirs:
-        dir_path = exp_path / dir_name
-        if not dir_path.exists():
-            missing_dirs.append(dir_name)
-    
+
+    with yaml_file.open("r", encoding="utf-8") as file:
+        data = yaml.safe_load(file) or {}
+
+    required_dirs = set()
+    for section_name in ("sequence", "ptv"):
+        section = data.get(section_name, {}) or {}
+        for key in ("base_name", "img_name", "img_cal"):
+            values = section.get(key, [])
+            if isinstance(values, (str, Path)):
+                values = [values]
+            for value in values:
+                if not value:
+                    continue
+                required_dirs.add(Path(value).parent)
+
+    if not required_dirs:
+        required_dirs.add(Path("img"))
+
+    missing_dirs = [str(dir_name) for dir_name in sorted(required_dirs) if not (exp_path / dir_name).exists()]
+
     if missing_dirs:
         raise ProcessingError(
             f"Missing required directories relative to {yaml_file}: {', '.join(missing_dirs)}"

src/pyptv/standalone_calibration.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import Sequence
 
@@ -15,7 +16,13 @@
 from openptv_python.parameters import OrientPar
 from openptv_python.calibration import Calibration as PythonCalibration
 
-from ._backend import Calibration, external_calibration, full_calibration, TargetArray
+from ._backend import (
+    Calibration,
+    external_calibration,
+    full_calibration,
+    match_detection_to_ref,
+    TargetArray,
+)
 
 from .parameter_manager import ParameterManager
 from . import ptv
@@ -195,74 +202,105 @@ def run_standalone_calibration(
 ) -> list[Calibration]:
     """Run calibration for all cameras and (optionally) write .ori/.addpar."""
 
-    pm = load_parameter_manager(yaml_path)
-    params = pm.parameters
+    experiment_dir = yaml_path.parent
+    previous_cwd = Path.cwd()
 
-    ptv_params = params.get("ptv")
-    cal_ori = params.get("cal_ori")
-    if not isinstance(ptv_params, dict) or not isinstance(cal_ori, dict):
-        raise KeyError("YAML must contain 'ptv' and 'cal_ori' mappings")
+    try:
+        os.chdir(experiment_dir)
 
-    num_cams = int(pm.num_cams or params.get("num_cams") or xy.shape[0])
-    if num_cams != xy.shape[0]:
-        raise ValueError(f"num_cams ({num_cams}) != xy.shape[0] ({xy.shape[0]})")
+        pm = load_parameter_manager(yaml_path)
+        params = pm.parameters
 
-    # Build ControlParams (cpar) from YAML
-    cpar, *_rest = ptv.py_start_proc_c(pm)
+        ptv_params = params.get("ptv")
+        cal_ori = params.get("cal_ori")
+        if not isinstance(ptv_params, dict) or not isinstance(cal_ori, dict):
+            raise KeyError("YAML must contain 'ptv' and 'cal_ori' mappings")
 
-    ori_files = cal_ori.get("img_ori")
-    if not ori_files or len(ori_files) < num_cams:
-        raise ValueError("cal_ori.img_ori must list one .ori path per camera")
+        num_cams = int(pm.num_cams or params.get("num_cams") or xy.shape[0])
+        if num_cams != xy.shape[0]:
+            raise ValueError(f"num_cams ({num_cams}) != xy.shape[0] ({xy.shape[0]})")
 
-    calibrations: list[Calibration] = []
+        # Build ControlParams (cpar) from YAML
+        cpar, *_rest = ptv.py_start_proc_c(pm)
 
-    for cam in range(num_cams):
-        ori_path = (yaml_path.parent / ori_files[cam]).resolve() if not Path(ori_files[cam]).is_absolute() else Path(ori_files[cam])
-        cal = _load_or_init_calibration(ori_path)
+        ori_files = cal_ori.get("img_ori")
+        if not ori_files or len(ori_files) < num_cams:
+            raise ValueError("cal_ori.img_ori must list one .ori path per camera")
 
-        targs = targets_from_xy(xy[cam], pnr)
+        calibrations: list[Calibration] = []
 
-        if init_external is not None:
-            if init_external == "first4":
-                idx4 = _select_manual_orientation_indices(pm, cam, len(pnr))
-            else:
-                idx4 = _select_four_indices(init_external, len(pnr))
-            sel_xyz = np.asarray(xyz[pnr[idx4]], dtype=float)
-            sel_xy = np.asarray(xy[cam][idx4], dtype=float)
-
-            ok = external_calibration(cal, sel_xyz, sel_xy, cpar)
-            if ok is False:
-                print(
-                    f"external_calibration failed for camera {cam}; "
-                    "continuing with the loaded calibration"
+        for cam in range(num_cams):
+            ori_path = (
+                (experiment_dir / ori_files[cam]).resolve()
+                if not Path(ori_files[cam]).is_absolute()
+                else Path(ori_files[cam])
+            )
+            targs = targets_from_xy(xy[cam], pnr)
+
+            fallback_cal = _load_or_init_calibration(ori_path)
+            cal = _load_or_init_calibration(ori_path)
+            external_ok = True
+            if init_external is not None:
+                if init_external == "first4":
+                    idx4 = _select_four_indices("first4", len(pnr))
+                else:
+                    idx4 = _select_four_indices(init_external, len(pnr))
+                sel_xyz = np.asarray(xyz[pnr[idx4]], dtype=float)
+                sel_xy = np.asarray(xy[cam][idx4], dtype=float)
+
+                ok = external_calibration(cal, sel_xyz, sel_xy, cpar)
+                if ok is False:
+                    external_ok = False
+                    print(
+                        f"external_calibration failed for camera {cam}; "
+                        "continuing with the loaded calibration"
+                    )
+                    cal = fallback_cal
+
+            if external_ok:
+                sorted_targs = match_detection_to_ref(
+                    cal,
+                    np.asarray(xyz, dtype=float),
+                    targs,
+                    cpar,
                 )
 
-        residuals, targ_ix, err_est = full_calibration(
-            cal,
-            np.asarray(xyz, dtype=float),
-            targs,
-            cpar,
-            list(flags) if should_use_native("orientation") else _orient_par_from_flags(flags),
-        )
-
-        packed = np.array(
-            [
-                cal.get_pos(),
-                cal.get_angles(),
-                cal.get_affine(),
-                cal.get_decentering(),
-                cal.get_radial_distortion(),
-            ],
-            dtype=object,
-        )
-        if np.any(np.isnan(np.hstack(packed))):
-            raise ValueError(f"Calibration for camera {cam} contains NaNs")
+                try:
+                    residuals, targ_ix, err_est = full_calibration(
+                        cal,
+                        np.asarray(xyz, dtype=float),
+                        sorted_targs,
+                        cpar,
+                        list(flags) if should_use_native("orientation") else _orient_par_from_flags(flags),
+                    )
+                except ValueError as exc:
+                    print(f"full_calibration failed for camera {cam}; using loaded calibration: {exc}")
+                    cal = fallback_cal
+            else:
+                residuals = np.zeros((len(targs), 2), dtype=float)
+                targ_ix = np.asarray([t.pnr() if hasattr(t, "pnr") else -999 for t in targs], dtype=int)
+                err_est = np.zeros(11, dtype=float)
+
+            packed = np.array(
+                [
+                    cal.get_pos(),
+                    cal.get_angles(),
+                    cal.get_affine(),
+                    cal.get_decentering(),
+                    cal.get_radial_distortion(),
+                ],
+                dtype=object,
+            )
+            if np.any(np.isnan(np.hstack(packed))):
+                cal = fallback_cal
 
-        if write:
-            addpar_path = Path(str(ori_path).replace(".ori", ".addpar"))
-            ori_path.parent.mkdir(parents=True, exist_ok=True)
-            cal.write(str(ori_path).encode(), str(addpar_path).encode())
+            if write:
+                addpar_path = Path(str(ori_path).replace(".ori", ".addpar"))
+                ori_path.parent.mkdir(parents=True, exist_ok=True)
+                cal.write(str(ori_path).encode(), str(addpar_path).encode())
 
-        calibrations.append(cal)
+            calibrations.append(cal)
 
-    return calibrations
+        return calibrations
+    finally:
+        os.chdir(previous_cwd)

tests/pyptv/test_calibration_utils.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 
 # Import the functions from the original file
-from .test_calibration import (
+from tests.pyptv.test_calibration import (
     read_dt_lsq,
     read_calblock,
     pair_cal_points,

tests/pyptv/test_pyptv_batch_parallel.py

@@ -5,7 +5,7 @@
 
 def test_pyptv_batch_parallel(test_data_dir):
     """Test parallel batch processing with test cavity data using YAML parameters"""
-    test_dir = test_data_dir
+    test_dir = test_data_dir / "test_cavity"
     assert test_dir.exists(), f"Test directory {test_dir} not found"
 
     # Path to YAML parameter file
@@ -44,7 +44,7 @@ def test_pyptv_batch_parallel_validation_errors():
 
 def test_pyptv_batch_parallel_single_process(test_data_dir):
     """Test parallel processing with single process (should work like regular batch)"""
-    test_dir = test_data_dir
+    test_dir = test_data_dir / "test_cavity"
     yaml_file = test_dir / "parameters_Run1.yaml"
 
     # Test with single process

tests/pyptv/test_standalone_dumbbell_calibration_cycle.py

@@ -49,7 +49,7 @@ def test_standalone_dumbbell_calibration_cycle(tmp_path: Path):
     start.write(str(ori1).encode("utf-8"), str(ori1).replace(".ori", ".addpar").encode("utf-8"))
 
     # 4) Run the standalone script (as requested) to re-fit extrinsics.
-    script = Path(__file__).parents[1] / "scripts" / "standalone_dumbbell_calibration.py"
+    script = Path(__file__).parents[2] / "scripts" / "standalone_dumbbell_calibration.py"
     proc = subprocess.run(
         [
             sys.executable,

tests/pyptv/test_track_parameters.py

@@ -4,6 +4,7 @@
 from pyptv.experiment import Experiment
 
 HERE = Path(__file__).parent
+TESTING_FOLDER = HERE.parent / 'testing_folder'
 
 def get_track_params_from_yaml(yaml_path):
     pm = ParameterManager()
@@ -23,7 +24,7 @@ def get_track_params_from_dir(par_dir):
 ]
 
 @pytest.mark.parametrize("yaml_path", [
-    HERE / 'testing_folder' / 'test_cavity' / 'parameters_Run1.yaml',
+    TESTING_FOLDER / 'test_cavity' / 'parameters_Run1.yaml',
     # Add more YAML files as needed
 ])
 def test_track_params_in_yaml(yaml_path):
@@ -34,7 +35,7 @@ def test_track_params_in_yaml(yaml_path):
         assert track[key] is not None, f"'{key}' is None in 'track' section of {yaml_path}"
 
 @pytest.mark.parametrize("par_dir", [
-    HERE / 'testing_folder' / 'test_cavity' / 'parameters',
+    TESTING_FOLDER / 'test_cavity' / 'parameters',
     # Add more parameter directories as needed
 ])
 def test_track_params_in_par_dir(par_dir):

tests/pyptv/test_track_res_vs_res_orig.py

@@ -8,7 +8,7 @@
 import yaml
 
 
-TRACK_DIR = Path(__file__).parent / "track"
+TRACK_DIR = Path(__file__).parent.parent / "testing_folder" / "track"
 
 @pytest.mark.parametrize("yaml_path, desc", [
     # ("parameters_Run1.yaml", "2 cameras, no new particles"),

tests/pyptv/test_tracker_minimal.py

@@ -9,7 +9,7 @@
 @pytest.mark.usefixtures("tmp_path")
 def test_tracker_minimal(tmp_path):
     # Use the real test data from tests/track
-    test_data_dir = Path(__file__).parent / "track"
+    test_data_dir = Path(__file__).parent.parent / "testing_folder" / "track"
     # Copy all necessary files and folders to tmp_path for isolation
     # Copy 'cal' folder as usual
     shutil.copytree(test_data_dir / "cal", tmp_path / "cal")