diff --git a/config.env.example b/config.env.example index 6b4bb92..ba09c98 100644 --- a/config.env.example +++ b/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) diff --git a/counter_live_rknn.py b/counter_live_rknn.py index 4f8f0e0..497cf31 100644 --- a/counter_live_rknn.py +++ b/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_str2 = ( - f" line2_crossed: {sorted(object_line2_crossed)}" - if object_line2_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") + ) + 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) diff --git a/env.example b/env.example index bf99158..bf5ddad 100644 --- a/env.example +++ b/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)