Sync algorithm changes with bat_tracker v1.1.5

app.py

@@ -13,7 +13,7 @@ import yaml
 from bat_tracker.pipeline import run_pipeline
 
 
-APP_VERSION = "v1.1.4"
+APP_VERSION = "v1.1.5"
 APP_TITLE = f"Bat Tracker {APP_VERSION}"
 APP_DESCRIPTION = (
     "Sube un video IR monocromo para ejecutar el pipeline, revisar la region valida "

bat_tracker/pipeline.py

@@ -491,7 +491,7 @@ def _filter_track_points(
         if not accepted:
             reasons_set = set(reject_reasons)
             if (
-                reasons_set.issubset({"min_track_length", "valid_region_gate"})
+                reasons_set.issubset({"min_track_length"})
                 and len(track_points) >= min_track_length_from_sec
                 and score >= strong_short_score_min
             ):
@@ -538,6 +538,14 @@ def _vector_cosine(v0: tuple[float, float], v1: tuple[float, float]) -> float |
     return (v0[0] * v1[0] + v0[1] * v1[1]) / (n0 * n1)
 
 
+def _track_edge_vectors(points: List[TrackPoint]) -> tuple[tuple[float, float], tuple[float, float]]:
+    start_idx = min(2, len(points) - 1)
+    end_idx = max(0, len(points) - 3)
+    start_vec = (points[start_idx].x - points[0].x, points[start_idx].y - points[0].y)
+    end_vec = (points[-1].x - points[end_idx].x, points[-1].y - points[end_idx].y)
+    return start_vec, end_vec
+
+
 def _auto_merge_track_points(points: List[TrackPoint], tracking_cfg: Dict) -> tuple[List[TrackPoint], List[Dict]]:
     if not bool(tracking_cfg.get("auto_merge_suggested", False)):
         return points, []
@@ -556,6 +564,7 @@ def _auto_merge_track_points(points: List[TrackPoint], tracking_cfg: Dict) -> tu
     min_overlap_common = int(tracking_cfg.get("merge_overlap_min_common_frames", 3))
     max_overlap_mean_dist = float(tracking_cfg.get("merge_overlap_max_mean_distance", 60.0))
     min_overlap_cos = float(tracking_cfg.get("merge_overlap_min_direction_cosine", 0.8))
+    local_overlap_min_cos = max(0.65, min_overlap_cos - 0.15)
 
     parent: Dict[int, int] = {track_id: track_id for track_id in by_track}
 
@@ -581,16 +590,14 @@ def _auto_merge_track_points(points: List[TrackPoint], tracking_cfg: Dict) -> tu
         a_pts = by_track[track_a_id]
         a_start = a_pts[0]
         a_end = a_pts[-1]
-        a_start_vec = (a_pts[min(2, len(a_pts) - 1)].x - a_pts[0].x, a_pts[min(2, len(a_pts) - 1)].y - a_pts[0].y)
-        a_end_vec = (a_pts[-1].x - a_pts[max(0, len(a_pts) - 3)].x, a_pts[-1].y - a_pts[max(0, len(a_pts) - 3)].y)
+        a_start_vec, a_end_vec = _track_edge_vectors(a_pts)
         a_frames = {p.frame: p for p in a_pts}
 
         for track_b_id in track_ids[idx + 1 :]:
             b_pts = by_track[track_b_id]
             b_start = b_pts[0]
             b_end = b_pts[-1]
-            b_start_vec = (b_pts[min(2, len(b_pts) - 1)].x - b_pts[0].x, b_pts[min(2, len(b_pts) - 1)].y - b_pts[0].y)
-            b_end_vec = (b_pts[-1].x - b_pts[max(0, len(b_pts) - 3)].x, b_pts[-1].y - b_pts[max(0, len(b_pts) - 3)].y)
+            b_start_vec, b_end_vec = _track_edge_vectors(b_pts)
 
             reason = None
             reason_data: Dict[str, float | int] = {}
@@ -610,28 +617,42 @@ def _auto_merge_track_points(points: List[TrackPoint], tracking_cfg: Dict) -> tu
             else:
                 b_frames = {p.frame: p for p in b_pts}
                 common_frames = sorted(set(a_frames.keys()).intersection(b_frames.keys()))
-                if len(common_frames) >= min_overlap_common:
+                if len(common_frames) >= 2:
                     distances = []
-                    cosines = []
                     for frame in common_frames:
                         pa = a_frames[frame]
                         pb = b_frames[frame]
                         distances.append(hypot(pa.x - pb.x, pa.y - pb.y))
 
                     mean_distance = sum(distances) / len(distances)
-                    c0 = _vector_cosine(a_start_vec, b_start_vec)
-                    c1 = _vector_cosine(a_end_vec, b_end_vec)
-                    if c0 is not None:
-                        cosines.append(c0)
-                    if c1 is not None:
-                        cosines.append(c1)
-                    mean_cos = (sum(cosines) / len(cosines)) if cosines else None
-                    if mean_distance <= max_overlap_mean_dist and (mean_cos is None or mean_cos >= min_overlap_cos):
-                        reason = "overlap"
+                    start_cos = _vector_cosine(a_start_vec, b_start_vec)
+                    end_cos = _vector_cosine(a_end_vec, b_end_vec)
+                    connector_cos = _vector_cosine(a_end_vec, b_start_vec)
+                    global_cosines = [c for c in (start_cos, end_cos) if c is not None]
+                    mean_cos = (sum(global_cosines) / len(global_cosines)) if global_cosines else None
+
+                    overlap_reason = None
+                    if len(common_frames) >= min_overlap_common:
+                        if mean_distance <= max_overlap_mean_dist and (
+                            mean_cos is None or mean_cos >= min_overlap_cos or connector_cos is not None and connector_cos >= min_overlap_cos
+                        ):
+                            overlap_reason = "overlap"
+                    elif (
+                        mean_distance <= max_overlap_mean_dist
+                        and connector_cos is not None
+                        and connector_cos >= local_overlap_min_cos
+                    ):
+                        overlap_reason = "overlap_local"
+
+                    if overlap_reason is not None:
+                        direction_score = connector_cos
+                        if direction_score is None:
+                            direction_score = mean_cos if mean_cos is not None else 1.0
+                        reason = overlap_reason
                         reason_data = {
                             "common_frames": len(common_frames),
                             "mean_distance": mean_distance,
-                            "mean_direction_cosine": mean_cos if mean_cos is not None else 1.0,
+                            "mean_direction_cosine": direction_score,
                         }
 
             if reason is None:

tests/test_track_exports.py

@@ -9,8 +9,9 @@ import cv2
 import numpy as np
 import yaml
 
-from bat_tracker.pipeline import run_pipeline
+from bat_tracker.pipeline import _auto_merge_track_points, run_pipeline
 from bat_tracker.render import export_tracks_render_json, export_tracks_svg
+from bat_tracker.tracker import TrackPoint
 
 
 SVG_NS = {"svg": "http://www.w3.org/2000/svg"}
@@ -196,3 +197,124 @@ def test_svg_and_render_json_export_empty_tracks_as_valid_empty_documents(tmp_pa
     svg_root = ET.parse(svg_path).getroot()
     assert svg_root.attrib["viewBox"] == "0 0 64 48"
     assert svg_root.findall("svg:g[@class='track']", SVG_NS) == []
+
+
+def _make_track_point(track_id: int, frame: int, x: float, y: float) -> TrackPoint:
+    return TrackPoint(
+        video_id="video",
+        track_id=track_id,
+        frame=frame,
+        time_sec=frame / 30.0,
+        x=x,
+        y=y,
+        vx=0.0,
+        vy=0.0,
+        bbox_x1=int(round(x)) - 1,
+        bbox_y1=int(round(y)) - 1,
+        bbox_x2=int(round(x)) + 1,
+        bbox_y2=int(round(y)) + 1,
+        area=20.0,
+    )
+
+
+def test_auto_merge_uses_local_overlap_continuity_for_short_shared_window() -> None:
+    points = [
+        _make_track_point(202, 32671, 715.5, 621.0),
+        _make_track_point(202, 32672, 739.5, 631.5),
+        _make_track_point(202, 32673, 760.5, 635.5),
+        _make_track_point(202, 32674, 785.0, 644.0),
+        _make_track_point(202, 32675, 810.5, 641.5),
+        _make_track_point(202, 32676, 837.5, 625.5),
+        _make_track_point(202, 32677, 860.5, 609.0),
+        _make_track_point(202, 32678, 895.0, 570.0),
+        _make_track_point(202, 32679, 922.0, 531.0),
+        _make_track_point(202, 32680, 917.5, 465.5),
+        _make_track_point(203, 32679, 882.0, 494.5),
+        _make_track_point(203, 32680, 979.0, 449.0),
+        _make_track_point(203, 32681, 1027.0, 404.0),
+        _make_track_point(203, 32682, 1035.5, 285.0),
+        _make_track_point(203, 32683, 1078.0, 172.0),
+        _make_track_point(203, 32684, 1128.5, 54.0),
+    ]
+    cfg = {
+        "auto_merge_suggested": True,
+        "merge_max_gap_frames": 12,
+        "merge_max_endpoint_distance": 100.0,
+        "merge_overlap_min_common_frames": 3,
+        "merge_overlap_max_mean_distance": 60.0,
+        "merge_overlap_min_direction_cosine": 0.8,
+    }
+
+    merged_points, merges = _auto_merge_track_points(points, cfg)
+
+    assert any(
+        merge["track_a"] == 202 and merge["track_b"] == 203 and merge["reason"] == "overlap_local"
+        for merge in merges
+    )
+    assert {point.track_id for point in merged_points} == {202}
+
+
+def test_auto_merge_uses_connector_direction_for_overlapping_fragments() -> None:
+    points = [
+        _make_track_point(180, 13432, 708.0, 667.0),
+        _make_track_point(180, 13433, 738.0, 666.5),
+        _make_track_point(180, 13434, 769.0, 661.0),
+        _make_track_point(180, 13435, 797.5, 650.5),
+        _make_track_point(180, 13436, 830.0, 634.5),
+        _make_track_point(180, 13437, 859.0, 606.0),
+        _make_track_point(180, 13438, 881.5, 588.0),
+        _make_track_point(180, 13439, 916.0, 549.5),
+        _make_track_point(180, 13440, 956.0, 486.5),
+        _make_track_point(180, 13441, 1012.0, 461.0),
+        _make_track_point(180, 13442, 1029.0, 363.5),
+        _make_track_point(180, 13443, 1081.0, 303.0),
+        _make_track_point(181, 13439, 937.5, 550.0),
+        _make_track_point(181, 13440, 910.5, 504.5),
+        _make_track_point(181, 13441, 962.5, 477.5),
+        _make_track_point(181, 13442, 984.5, 393.0),
+        _make_track_point(181, 13443, 1002.5, 335.0),
+    ]
+    cfg = {
+        "auto_merge_suggested": True,
+        "merge_max_gap_frames": 12,
+        "merge_max_endpoint_distance": 100.0,
+        "merge_overlap_min_common_frames": 3,
+        "merge_overlap_max_mean_distance": 60.0,
+        "merge_overlap_min_direction_cosine": 0.8,
+    }
+
+    merged_points, merges = _auto_merge_track_points(points, cfg)
+
+    assert any(
+        merge["track_a"] == 180 and merge["track_b"] == 181 and merge["reason"] == "overlap"
+        for merge in merges
+    )
+    assert {point.track_id for point in merged_points} == {180}
+
+
+def test_auto_merge_keeps_nearby_tracks_separate_when_connector_direction_breaks() -> None:
+    points = [
+        _make_track_point(86, 1091, 1703.5, 36.0),
+        _make_track_point(86, 1092, 1703.5, 36.0),
+        _make_track_point(86, 1093, 1722.0, 47.5),
+        _make_track_point(86, 1094, 1693.5, 16.5),
+        _make_track_point(86, 1095, 1663.0, 24.5),
+        _make_track_point(86, 1096, 1671.5, 2.5),
+        _make_track_point(90, 1092, 1640.0, 71.0),
+        _make_track_point(90, 1093, 1651.0, 59.5),
+        _make_track_point(90, 1094, 1671.5, 46.5),
+        _make_track_point(90, 1095, 1707.5, 50.0),
+    ]
+    cfg = {
+        "auto_merge_suggested": True,
+        "merge_max_gap_frames": 12,
+        "merge_max_endpoint_distance": 100.0,
+        "merge_overlap_min_common_frames": 3,
+        "merge_overlap_max_mean_distance": 60.0,
+        "merge_overlap_min_direction_cosine": 0.8,
+    }
+
+    merged_points, merges = _auto_merge_track_points(points, cfg)
+
+    assert merges == []
+    assert {point.track_id for point in merged_points} == {86, 90}