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

add 2 second window for snapshot and counter

Browse Source
This commit is contained in:
Alberto-Audrix committed 2026-07-24 13:51:17 +07:00
1 parent a7ce970f78
commit 7721c84406
3 files changed
+160 -95
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

counter_live_rknn.py
+148 -95
View File
@@ -99,12 +99,17 @@ INHERIT_PX = float(os.getenv("INHERIT_PX", "60"))
ZONE_COOLDOWN_LEFT_SEC = float(os.getenv("ZONE_COOLDOWN_LEFT_SEC", "0"))
ZONE_COOLDOWN_RIGHT_SEC = float(os.getenv("ZONE_COOLDOWN_RIGHT_SEC", "0"))
# Per-feeder dwell: sack must remain inside the zone for N seconds before it is
# counted (and a cross snapshot is saved). 0 = count on first frame in zone.
ZONE_DWELL_LEFT_SEC = float(os.getenv("ZONE_DWELL_LEFT_SEC", "2"))
ZONE_DWELL_RIGHT_SEC = float(os.getenv("ZONE_DWELL_RIGHT_SEC", "2"))
DAILY_CUTOFF_TIME = os.getenv("DAILY_CUTOFF_TIME", "20:00")
EXPORT_CSV = os.getenv("EXPORT_CSV", "true").lower() == "true"
CROSS_CSV = os.getenv("CROSS_CSV", f"{OUTPUT_DIR}/crossings.csv")
# Save an annotated frame snapshot each time an object enters a feeder zone and
# Save an annotated frame snapshot each time a sack finishes its zone dwell and
# the counter increases.
SAVE_CROSS_SNAPSHOT = os.getenv("SAVE_CROSS_SNAPSHOT", "false").lower() == "true"
# Also save one snapshot the first time each object is detected (before it enters
@@ -738,6 +743,29 @@ def feeder_cooldown_sec(feeder):
return ZONE_COOLDOWN_RIGHT_SEC
def feeder_dwell_sec(feeder):
if feeder == "left":
return ZONE_DWELL_LEFT_SEC
return ZONE_DWELL_RIGHT_SEC
def resolve_feeder_at(cx, cy, zone_left, zone_right):
"""Return 'left' | 'right' | None for the zone containing (cx, cy)."""
now_left = point_in_zone(cx, cy, zone_left)
now_right = point_in_zone(cx, cy, zone_right)
if now_left and now_right:
lx1, ly1, lx2, ly2 = zone_left
rx1, ry1, rx2, ry2 = zone_right
dl = abs(cx - (lx1 + lx2) / 2) + abs(cy - (ly1 + ly2) / 2)
dr = abs(cx - (rx1 + rx2) / 2) + abs(cy - (ry1 + ry2) / 2)
return "left" if dl <= dr else "right"
if now_left:
return "left"
if now_right:
return "right"
return None
def feeder_display_name(feeder):
if feeder == "left":
return FEEDER_LEFT_NAME
@@ -781,14 +809,23 @@ def _default_zone_state():
counted: already contributed one LEFT or RIGHT count (at most one per object).
side: 'left' | 'right' | None after a count.
dwell_feeder / dwell_since: which zone the track is currently dwelling in,
and when continuous presence in that zone began.
"""
return {"counted": False, "side": None}
return {
"counted": False,
"side": None,
"dwell_feeder": None,
"dwell_since": None,
}
def _copy_zone_state(src):
return {
"counted": bool(src.get("counted", False)),
"side": src.get("side"),
"dwell_feeder": src.get("dwell_feeder"),
"dwell_since": src.get("dwell_since"),
}
@@ -1310,7 +1347,8 @@ def run():
)
print(
f"Cooldownoldown: left={ZONE_COOLDOWN_LEFT_SEC}s right={ZONE_COOLDOWN_RIGHT_SEC}s "
f"(0=off)"
f"(0=off) | Dwell: left={ZONE_DWELL_LEFT_SEC}s right={ZONE_DWELL_RIGHT_SEC}s "
f"(0=immediate)"
)
print(f"Model: {MODEL_PATH} | imgsz={IMGSZ} | core_mask={CORE_MASK}")
print(
@@ -1443,9 +1481,10 @@ def run():
flush=True,
)
# --- zone entry detection only on DETECTED objects this frame ---
# (tracker still updates every frame to preserve IDs, but we do NOT
# count on Kalman-predicted/coasting tracks to avoid double counts)
# --- zone dwell counting only on DETECTED objects this frame ---
# A sack must remain inside a feeder zone for ZONE_DWELL_*_SEC before
# it is counted (and a cross snapshot is saved). Leaving the zone
# resets the dwell timer. Cooldown still applies after a successful count.
for di in range(len(object_boxes_xyxy)):
tid = object_det_to_track.get(di)
if tid is None:
@@ -1474,7 +1513,8 @@ def run():
f"[DEBUG F{frame_idx}] INHERIT prev for new tid={tid} "
f"from tid={src_tid} ({inherited[0]:.1f},{inherited[1]:.1f})"
f" counted={src_st.get('counted', False)}"
f" side={src_st.get('side')}",
f" side={src_st.get('side')}"
f" dwell={src_st.get('dwell_feeder')}",
flush=True,
)
@@ -1482,105 +1522,118 @@ def run():
object_zone_state[tid] = _default_zone_state()
st = object_zone_state[tid]
now_left = point_in_zone(cx, cy, zone_left)
now_right = point_in_zone(cx, cy, zone_right)
feeder = resolve_feeder_at(cx, cy, zone_left, zone_right)
entered_left = False
entered_right = False
if tid in object_tracked:
prev_cx, prev_cy = object_tracked[tid][0], object_tracked[tid][1]
prev_left = point_in_zone(prev_cx, prev_cy, zone_left)
prev_right = point_in_zone(prev_cx, prev_cy, zone_right)
entered_left = now_left and not prev_left
entered_right = now_right and not prev_right
else:
# First observation: count if the centroid already sits in a zone.
entered_left = now_left
entered_right = now_right
if st["counted"]:
object_tracked[tid] = (cx, cy, mono)
continue
if (entered_left or entered_right) and not st["counted"]:
# Prefer the zone whose center is closer if both fire in one jump.
feeder = None
if entered_left and entered_right:
lx1, ly1, lx2, ly2 = zone_left
rx1, ry1, rx2, ry2 = zone_right
dl = abs(cx - (lx1 + lx2) / 2) + abs(cy - (ly1 + ly2) / 2)
dr = abs(cx - (rx1 + rx2) / 2) + abs(cy - (ry1 + ry2) / 2)
feeder = "left" if dl <= dr else "right"
elif entered_left:
feeder = "left"
else:
feeder = "right"
if feeder is None:
# Left all zones — reset dwell progress.
if st["dwell_feeder"] is not None and DEBUG_TRACKING:
print(
f"[DEBUG F{frame_idx}] DWELL RESET: tid={tid} "
f"left feeder={st['dwell_feeder']}",
flush=True,
)
st["dwell_feeder"] = None
st["dwell_since"] = None
object_tracked[tid] = (cx, cy, mono)
continue
recent = (
recent_cross_left if feeder == "left" else recent_cross_right
)
if is_duplicate_cross(recent, cx, cy, frame_idx):
st["counted"] = True
st["side"] = feeder
if DEBUG_TRACKING:
print(
f"[DEBUG F{frame_idx}] DUP ZONE IGNORED: tid={tid} "
f"cx={cx:.1f} cy={cy:.1f} feeder={feeder}",
flush=True,
)
object_tracked[tid] = (cx, cy, mono)
continue
if st["dwell_feeder"] != feeder:
st["dwell_feeder"] = feeder
st["dwell_since"] = mono
if DEBUG_TRACKING:
print(
f"[DEBUG F{frame_idx}] DWELL START: tid={tid} "
f"feeder={feeder} need={feeder_dwell_sec(feeder):.1f}s",
flush=True,
)
object_tracked[tid] = (cx, cy, mono)
continue
if feeder_in_cooldown(feeder, mono, last_count_mono):
st["counted"] = True
st["side"] = feeder
if DEBUG_TRACKING:
cool = feeder_cooldown_sec(feeder)
elapsed = mono - (last_count_mono.get(feeder) or mono)
print(
f"[DEBUG F{frame_idx}] COOLDOWN IGNORED: tid={tid} "
f"feeder={feeder} elapsed={elapsed:.2f}s / {cool}s",
flush=True,
)
object_tracked[tid] = (cx, cy, mono)
continue
dwell_need = feeder_dwell_sec(feeder)
dwell_elapsed = mono - (st["dwell_since"] or mono)
if dwell_need > 0 and dwell_elapsed < dwell_need:
object_tracked[tid] = (cx, cy, mono)
continue
recent.append((frame_idx, cx, cy))
# Dwell complete — apply dedup / cooldown, then count + snapshot.
recent = recent_cross_left if feeder == "left" else recent_cross_right
if is_duplicate_cross(recent, cx, cy, frame_idx):
st["counted"] = True
st["side"] = feeder
last_count_mono[feeder] = mono
if feeder == "left":
counter_left += 1
count_left_pulse = COUNT_PULSE_FRAMES
feeder_total = counter_left
else:
counter_right += 1
count_right_pulse = COUNT_PULSE_FRAMES
feeder_total = counter_right
# Frigate-style count log, e.g.
# "Total karung di kandang_bawah_feeder_kanan: 5"
if DEBUG_TRACKING:
print(
f"[DEBUG F{frame_idx}] DUP ZONE IGNORED: tid={tid} "
f"cx={cx:.1f} cy={cy:.1f} feeder={feeder}",
flush=True,
)
object_tracked[tid] = (cx, cy, mono)
continue
if feeder_in_cooldown(feeder, mono, last_count_mono):
# Keep dwelling; count once cooldown expires if still in zone.
if DEBUG_TRACKING:
cool = feeder_cooldown_sec(feeder)
elapsed = mono - (last_count_mono.get(feeder) or mono)
print(
f"[DEBUG F{frame_idx}] DWELL WAIT COOLDOWN: tid={tid} "
f"feeder={feeder} elapsed={elapsed:.2f}s / {cool}s",
flush=True,
)
object_tracked[tid] = (cx, cy, mono)
continue
if DEBUG_TRACKING:
print(
f"Total {OBJECT_LABEL} di {feeder_display_name(feeder)}: {feeder_total}",
f"[DEBUG F{frame_idx}] DWELL COUNT: tid={tid} "
f"cx={cx:.1f} cy={cy:.1f} feeder={feeder} "
f"dwell={dwell_elapsed:.2f}s",
flush=True,
)
store.record_zone_entry(tid, feeder)
if cross_logger:
cross_logger.write_row(
[
store.get_counting_date(),
frame_idx,
feeder,
tid,
]
)
object_crossed_frame = True
cross_events_frame.append((tid, feeder))
crossing_times.append(mono)
object_cross_flash[tid] = CROSS_FLASH_FRAMES
popups.append(
{
"x": int(cx) - 12,
"y": int(cy),
"born": frame_idx,
"text": f"+1 {feeder.upper()}",
}
recent.append((frame_idx, cx, cy))
st["counted"] = True
st["side"] = feeder
last_count_mono[feeder] = mono
if feeder == "left":
counter_left += 1
count_left_pulse = COUNT_PULSE_FRAMES
feeder_total = counter_left
else:
counter_right += 1
count_right_pulse = COUNT_PULSE_FRAMES
feeder_total = counter_right
# Frigate-style count log, e.g.
# "Total karung di kandang_bawah_feeder_kanan: 5"
print(
f"Total {OBJECT_LABEL} di {feeder_display_name(feeder)}: {feeder_total}",
flush=True,
)
store.record_zone_entry(tid, feeder)
if cross_logger:
cross_logger.write_row(
[
store.get_counting_date(),
frame_idx,
feeder,
tid,
]
)
object_crossed_frame = True
cross_events_frame.append((tid, feeder))
crossing_times.append(mono)
object_cross_flash[tid] = CROSS_FLASH_FRAMES
popups.append(
{
"x": int(cx) - 12,
"y": int(cy),
"born": frame_idx,
"text": f"+1 {feeder.upper()}",
}
)
object_tracked[tid] = (cx, cy, mono)
for tid, (cx, cy) in object_lost_map.items():