dsutanto/zenai-kpc-python
1
0 Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update counting logic to use IN and OUT line to determine direction IN or OUT #3

Merged
dsutanto merged 1 commits from audrix/zenai-kpc-python:main into main 2026-07-14 17:11:08 +07:00
Conversation 0 Commits 1 Files Changed 3
+146 -43
3 changed files with 146 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

config.env.example
+5 -3
View File
@@ -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
+136 -37
View File
@@ -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
+5 -3
View File
@@ -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)