updated api

Author: alexlib

src/openptv_python/tracking_frame_buf.py

@@ -40,7 +40,6 @@ def quicksort_n_tupel(arr: np.recarray) -> np.recarray:
     -------
       A list of n_tupel instances, sorted by the corr attribute.
     """
-    # inline sorting
     arr.sort(order="corr")
     return arr
 
@@ -53,6 +52,16 @@ def quicksort_n_tupel(arr: np.recarray) -> np.recarray:
 )
 
 
+def _resolve_frame_base(file_base: str, frame_num: int, suffix: str = "") -> Path:
+    if frame_num > 0:
+        if "%" in file_base:
+            return Path(file_base % frame_num + suffix)
+        if file_base.endswith((".", "_", "-", "/")):
+            return Path(f"{file_base}{frame_num:04d}{suffix}")
+        return Path(f"{file_base}.{frame_num:04d}{suffix}")
+    return Path(f"{file_base}{suffix}")
+
+
 # def __eq__(self, other):
 #     return self.nr == other.nr and np.all(self.p == other.p)
 
@@ -165,17 +174,7 @@ def num_targs(self):
 def read_targets(file_base: str, frame_num: int) -> List[Target]:
     """Read targets from a file."""
     buffer = []
-
-    # # if file_base has an extension, remove it
-    # file_base = file_base.split(".")[0]
-
-    if frame_num > 0:
-        # filename = f"{file_base}{frame_num:04d}_targets"
-        fname = file_base % frame_num + "_targets"
-    else:
-        fname = f"{file_base}_targets"
-
-    filename = Path(fname)
+    filename = _resolve_frame_base(file_base, frame_num, "_targets")
     print(f" filename: {filename}")
 
     try:
@@ -214,17 +213,7 @@ def write_targets(
 ) -> bool:
     """Write targets to a file."""
     success = False
-
-    # fix old-type names, that are like cam1.# or just cam1.
-    if "#" in file_base:
-        file_base = file_base.replace("#", "%05d")
-    if "%" not in file_base:
-        file_base = file_base + "%05d"
-
-    if frame_num == 0:
-        frame_num = 123456789
-
-    file_name = file_base % frame_num + "_targets"
+    file_name = _resolve_frame_base(file_base, frame_num, "_targets")
 
     try:
         # Convert targets to a 2D numpy array
@@ -341,19 +330,16 @@ def read(
         frame_num: int,
     ) -> bool:
         """Read a frame from the disk."""
-        required_files = [Path(f"{corres_file_base}.{frame_num}")]
+        required_files = [_resolve_frame_base(corres_file_base, frame_num)]
 
         if linkage_file_base != "":
-            required_files.append(Path(f"{linkage_file_base}.{frame_num}"))
+            required_files.append(_resolve_frame_base(linkage_file_base, frame_num))
 
         if prio_file_base != "":
-            required_files.append(Path(f"{prio_file_base}.{frame_num}"))
+            required_files.append(_resolve_frame_base(prio_file_base, frame_num))
 
         for file_base in target_file_base:
-            if frame_num > 0:
-                required_files.append(Path(file_base % frame_num + "_targets"))
-            else:
-                required_files.append(Path(f"{file_base}_targets"))
+            required_files.append(_resolve_frame_base(file_base, frame_num, "_targets"))
 
         for path in required_files:
             if not path.exists():

src/pyptv/ptv.py

@@ -66,6 +66,10 @@
 SHORT_BASE = "cam"  # Use this as the short base for camera file naming
 
 
+def _as_text(value: str | bytes) -> str:
+    return value.decode() if isinstance(value, bytes) else value
+
+
 def _prepare_output_path(filename: str) -> Path:
     """Return a writable output path, creating parent directories when needed."""
     output_path = Path(filename)
@@ -503,7 +507,7 @@ def py_determination_proc_c(
         print_corresp = concatenated_corresp
 
     output_path = _prepare_output_path(
-        f"{default_naming['corres'].decode()}.{DEFAULT_FRAME_NUM}"
+        f"{_as_text(default_naming['corres'])}.{DEFAULT_FRAME_NUM}"
     )
 
     print(f"Prepared {output_path} to write positions")
@@ -683,14 +687,19 @@ def py_sequence_loop(exp) -> None:
         flat = np.array(
             [corr.get_by_pnrs(corresp) for corr, corresp in zip(corrected, sorted_corresp)]
         )
-        pos, _ = point_positions(flat.transpose(1, 0, 2), exp.cpar, exp.cals, exp.vpar)
+        pos, _ = point_positions(
+            flat.transpose(1, 0, 2),
+            exp.cpar.mm,
+            exp.cals,
+            exp.vpar,
+        )
         if len(exp.cals) < 4:
             print_corresp = -1 * np.ones((4, sorted_corresp.shape[1]))
             print_corresp[: len(exp.cals), :] = sorted_corresp
         else:
             print_corresp = sorted_corresp
 
-        output_path = _prepare_output_path(f"{default_naming['corres'].decode()}.{frame}")
+        output_path = _prepare_output_path(f"{_as_text(default_naming['corres'])}.{frame}")
         try:
             with open(output_path, "w", encoding="utf8") as rt_is:
                 rt_is.write(f"{pos.shape[0]}\n")
@@ -795,9 +804,20 @@ def write_targets(targets: TargetArray, short_file_base: str, frame: int) -> boo
         return True  # No targets to write, but file created successfully
 
     try:
+        def _value(field):
+            return field() if callable(field) else field
+
         target_arr = np.array(
             [
-                ([t.pnr(), *t.pos(), *t.count_pixels(), t.sum_grey_value(), t.tnr()])
+                (
+                    [
+                        _value(t.pnr),
+                        *t.pos(),
+                        *t.count_pixels(),
+                        t.sum_grey_value(),
+                        _value(t.tnr),
+                    ]
+                )
                 for t in targets
             ]
         )

tests/test_fb.py

@@ -1,11 +1,14 @@
 import os
+import tempfile
 import unittest
+from pathlib import Path
 
 import numpy as np
 
 from openptv_python.constants import POSI
 from openptv_python.tracking_frame_buf import (
     Corres_dtype,
+    Frame,
     Pathinfo,
     Target,
     compare_corres,
@@ -103,6 +106,37 @@ def test_write_targets(self):
         os.remove(file_base % frame_num + "_targets")
 
 
+class TestPlainBaseFrameIO(unittest.TestCase):
+    def test_frame_read_accepts_plain_short_bases(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmp_path = Path(tmpdir)
+            corres_base = tmp_path / "res" / "rt_is"
+            target_base = tmp_path / "img" / "cam1"
+
+            corres_base.parent.mkdir(parents=True, exist_ok=True)
+            target_base.parent.mkdir(parents=True, exist_ok=True)
+
+            frame_num = 42
+            (corres_base.parent / f"{corres_base.name}.0042").write_text("0\n", encoding="utf-8")
+            (target_base.parent / f"{target_base.name}.0042_targets").write_text(
+                "0\n",
+                encoding="utf-8",
+            )
+
+            frame = Frame(num_cams=1, max_targets=1)
+
+            result = frame.read(
+                str(corres_base),
+                "",
+                "",
+                [str(target_base)],
+                frame_num,
+            )
+
+            self.assertTrue(result)
+            self.assertEqual(frame.num_targets[0], 0)
+
+
 class TestReadPathFrame(unittest.TestCase):
     def test_read_path_frame(self):
         """Test reading path frame with and without links.