config.env.example

@@ -92,9 +92,11 @@ CLASS_OBJECT=object
 OBJECT_CLASS_ID=0
 
 # --- Line crossing ---
-# Two horizontal counting lines:
-#   Line 1 (default ~33%): counts top-to-down (IN)
-#   Line 2 (default ~66%): counts bottom-to-up (OUT)
+# Two horizontal counting lines with sequence logic:
+#   Line 1 (IN):  top→down counts IN  (OUT→IN sequence, or IN-only)
+#   Line 2 (OUT): bottom→up counts OUT (IN→OUT sequence, or OUT-only)
+# Passing the opposite line first only arms; the destination line still counts
+# even if the opposite line was never crossed.
 # Fixed y-coordinate for line 1/IN (overrides LINE_Y1_FRAC if set)
 LINE_Y1=
 # Fraction of frame height for line 1 (default 0.33)

counter_live_rknn.py

@@ -694,8 +694,9 @@ def is_duplicate_cross(recent, cx, frame_idx):
 
 
 def _inherit_prev(tracked, new_tid, cx, cy, mono, max_age, max_px):
-    """Find a recently-seen nearby track and return its (cx, cy, mono) so a new
-    track ID can inherit the trajectory across an ID switch. Returns None if none.
+    """Find a recently-seen nearby track for ID-switch continuation.
+
+    Returns (cx, cy, mono, source_tid) or None.
 
     The predecessor must be close in BOTH x and y (same physical object at the same
     spot). Matching on x only would let a new track inherit a far-away y, seeding a
@@ -712,10 +713,27 @@ def _inherit_prev(tracked, new_tid, cx, cy, mono, max_age, max_px):
         dist = abs(tcx - cx)
         if dist <= best_dist:
             best_dist = dist
-            best = (tcx, tcy, ts)
+            best = (tcx, tcy, ts, tid)
     return best
 
 
+def _default_cross_state():
+    """Per-track line-crossing state for sequence counting.
+
+    seen_out / seen_in: ever registered a pass of that line (arm or count).
+    counted: already contributed one IN or OUT (at most one per physical object).
+    """
+    return {"seen_out": False, "seen_in": False, "counted": False}
+
+
+def _copy_cross_state(src):
+    return {
+        "seen_out": bool(src.get("seen_out", False)),
+        "seen_in": bool(src.get("seen_in", False)),
+        "counted": bool(src.get("counted", False)),
+    }
+
+
 def now_str():
     return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
@@ -1176,8 +1194,9 @@ def run():
         min_hits=TRACK_MIN_HITS,
     )
 
-    object_line1_crossed = set()
-    object_line2_crossed = set()
+    # Per-track sequence state: OUT→IN / IN-only → IN; IN→OUT / OUT-only → OUT.
+    # Keys survive ID switches via inheritance (see _inherit_prev).
+    object_cross_state = {}
 
     recent_cross_in = deque()
     recent_cross_out = deque()
@@ -1226,7 +1245,9 @@ def run():
     line_y1 = resolve_line_y1(h)
     line_y2 = resolve_line_y2(h)
     print(
-        f"RKNN+ByteTrack counter | {w}x{h} @ {fps}fps | line1 y={line_y1} (in v)  line2 y={line_y2} (out ^)"
+        f"RKNN+ByteTrack counter | {w}x{h} @ {fps}fps | "
+        f"line1 y={line_y1} (IN ↓)  line2 y={line_y2} (OUT ↑) | "
+        f"seq: OUT→IN/IN-only→in, IN→OUT/OUT-only→out"
     )
     print(f"Model: {MODEL_PATH} | imgsz={IMGSZ} | core_mask={CORE_MASK}")
     print(
@@ -1336,16 +1357,24 @@ def run():
                     if object_det_to_track
                     else ""
                 )
-                crossing_str1 = (
-                    f"  line1_crossed: {sorted(object_line1_crossed)}"
-                    if object_line1_crossed
-                    else ""
+                crossing_str1 = ""
+                crossing_str2 = ""
+                if object_cross_state:
+                    seen_in_ids = sorted(
+                        tid for tid, st in object_cross_state.items() if st.get("seen_in")
                     )
-                crossing_str2 = (
-                    f"  line2_crossed: {sorted(object_line2_crossed)}"
-                    if object_line2_crossed
-                    else ""
+                    seen_out_ids = sorted(
+                        tid for tid, st in object_cross_state.items() if st.get("seen_out")
                     )
+                    counted_ids = sorted(
+                        tid for tid, st in object_cross_state.items() if st.get("counted")
+                    )
+                    if seen_in_ids:
+                        crossing_str1 = f"  seen_in: {seen_in_ids}"
+                    if seen_out_ids:
+                        crossing_str2 = f"  seen_out: {seen_out_ids}"
+                    if counted_ids:
+                        crossing_str2 += f"  counted: {counted_ids}"
                 print(
                     f"[DEBUG F{frame_idx}] dets={len(object_boxes_xyxy)} "
                     f"tracks={len(object_track_map)} "
@@ -1367,36 +1396,92 @@ def run():
                 detect_snapshot_ids.add(tid)
                 detect_events_frame.append(tid)
 
+            inherited_this_frame = False
             if tid not in object_tracked:
                 inherited = _inherit_prev(
                     object_tracked, tid, cx, cy, mono, INHERIT_SEC, INHERIT_PX
                 )
                 if inherited is not None:
-                    object_tracked[tid] = inherited
+                    object_tracked[tid] = inherited[:3]
+                    src_tid = inherited[3]
+                    if src_tid in object_cross_state:
+                        object_cross_state[tid] = _copy_cross_state(
+                            object_cross_state[src_tid]
+                        )
+                    inherited_this_frame = True
                     if os.getenv("DEBUG_TRACKING", "").lower() == "true":
+                        src_st = object_cross_state.get(tid, {})
                         print(
                             f"[DEBUG F{frame_idx}] INHERIT prev for new tid={tid} "
-                            f"from ({inherited[0]:.1f},{inherited[1]:.1f})"
+                            f"from tid={src_tid} ({inherited[0]:.1f},{inherited[1]:.1f})"
+                            f" seen_in={src_st.get('seen_in', False)}"
+                            f" seen_out={src_st.get('seen_out', False)}"
+                            f" counted={src_st.get('counted', False)}"
                         )
 
+            if tid not in object_cross_state:
+                object_cross_state[tid] = _default_cross_state()
+            st = object_cross_state[tid]
+
             if tid in object_tracked:
                 prev_cy = object_tracked[tid][1]
-                crossed1 = (
-                    crossed_top_down(prev_cy, cy, line_y1)
-                    and tid not in object_line1_crossed
+                # Line 1 = IN, line 2 = OUT.
+                # Arm (no count): OUT top→down, IN bottom→up.
+                # Count:       IN top→down → IN  (OUT→IN or IN-only)
+                #              OUT bottom→up → OUT (IN→OUT or OUT-only)
+                out_down = (
+                    crossed_top_down(prev_cy, cy, line_y2) and not st["seen_out"]
                 )
-                crossed2 = (
-                    crossed_bottom_up(prev_cy, cy, line_y2)
-                    and tid not in object_line2_crossed
+                in_up = (
+                    crossed_bottom_up(prev_cy, cy, line_y1) and not st["seen_in"]
                 )
-                if crossed1 or crossed2:
-                    direction = "in" if crossed1 else "out"
-                    recent = recent_cross_in if crossed1 else recent_cross_out
+                in_down = (
+                    crossed_top_down(prev_cy, cy, line_y1) and not st["seen_in"]
+                )
+                out_up = (
+                    crossed_bottom_up(prev_cy, cy, line_y2) and not st["seen_out"]
+                )
+
+                if out_down or in_up or in_down or out_up:
+                    if out_down:
+                        st["seen_out"] = True
+                    if in_up:
+                        st["seen_in"] = True
+                    if in_down:
+                        st["seen_in"] = True
+                    if out_up:
+                        st["seen_out"] = True
+
+                    # Prefer decisive motion if both count triggers fire in one jump.
+                    count_in = in_down and not st["counted"]
+                    count_out = out_up and not st["counted"]
+                    if count_in and count_out:
+                        if cy >= prev_cy:
+                            count_out = False
+                        else:
+                            count_in = False
+                    direction = "in" if count_in else ("out" if count_out else None)
+
+                    if direction is None:
+                        # Arm-only pass (OUT↓ or IN↑). No count yet.
+                        if os.getenv("DEBUG_TRACKING", "").lower() == "true":
+                            arm = []
+                            if out_down:
+                                arm.append("out_down")
+                            if in_up:
+                                arm.append("in_up")
+                            print(
+                                f"[DEBUG F{frame_idx}] CROSS ARM: tid={tid} "
+                                f"prev_cy={prev_cy:.1f} -> cy={cy:.1f} "
+                                f"arm={'+'.join(arm)} "
+                                f"seen_in={st['seen_in']} seen_out={st['seen_out']}"
+                            )
+                        object_tracked[tid] = (cx, cy, mono)
+                        continue
+
+                    recent = recent_cross_in if direction == "in" else recent_cross_out
                     if is_duplicate_cross(recent, cx, frame_idx):
-                        if crossed1:
-                            object_line1_crossed.add(tid)
-                        if crossed2:
-                            object_line2_crossed.add(tid)
+                        st["counted"] = True
                         if os.getenv("DEBUG_TRACKING", "").lower() == "true":
                             print(
                                 f"[DEBUG F{frame_idx}] DUP CROSS IGNORED: tid={tid} "
@@ -1404,20 +1489,34 @@ def run():
                             )
                         object_tracked[tid] = (cx, cy, mono)
                         continue
+
+                    seq = (
+                        "out→in"
+                        if direction == "in" and st["seen_out"]
+                        else (
+                            "in→out"
+                            if direction == "out" and st["seen_in"]
+                            else ("in-only" if direction == "in" else "out-only")
+                        )
+                    )
                     if os.getenv("DEBUG_TRACKING", "").lower() == "true":
-                        line_label = "1" if crossed1 else "2"
-                        line_y_val = line_y1 if crossed1 else line_y2
                         print(
                             f"[DEBUG F{frame_idx}] CROSS DETECTED: tid={tid} "
-                            f"prev_cy={prev_cy:.1f} -> cy={cy:.1f} line={line_label} y={line_y_val} dir={direction}"
+                            f"prev_cy={prev_cy:.1f} -> cy={cy:.1f} dir={direction} "
+                            f"seq={seq}"
+                            + (
+                                " (inherit)"
+                                if inherited_this_frame
+                                else ""
                             )
+                        )
+
                     recent.append((frame_idx, cx))
-                    if crossed1:
-                        object_line1_crossed.add(tid)
+                    st["counted"] = True
+                    if direction == "in":
                         counter_in += 1
                         count_in_pulse = COUNT_PULSE_FRAMES
-                    if crossed2:
-                        object_line2_crossed.add(tid)
+                    else:
                         counter_out += 1
                         count_out_pulse = COUNT_PULSE_FRAMES
                     store.record_object_crossing(tid, direction)
@@ -1440,7 +1539,7 @@ def run():
                             "x": int(cx) - 12,
                             "y": int(cy),
                             "born": frame_idx,
-                            "text": f"+1" if not (crossed1 and crossed2) else "x2",
+                            "text": "+1",
                         }
                     )
             object_tracked[tid] = (cx, cy, mono)

env.example

@@ -68,9 +68,11 @@ CLASS_OBJECT=karung
 OBJECT_CLASS_ID=0
 
 # --- Line crossing ---
-# Two horizontal counting lines:
-#   Line 1 (default ~33%): counts top-to-down (IN)
-#   Line 2 (default ~66%): counts bottom-to-up (OUT)
+# Two horizontal counting lines with sequence logic:
+#   Line 1 (IN):  top→down counts IN  (OUT→IN sequence, or IN-only)
+#   Line 2 (OUT): bottom→up counts OUT (IN→OUT sequence, or OUT-only)
+# Passing the opposite line first only arms; the destination line still counts
+# even if the opposite line was never crossed.
 # Fixed y-coordinate for line 1/IN (overrides LINE_Y1_FRAC if set)
 LINE_Y1=
 # Fraction of frame height for line 1 (default 0.33)