110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""
|
|
In-memory batch tracker for offline case review (no SQLite / no timers).
|
|
Logic aligned with jetson-counter/batch_store.py crossing rules.
|
|
"""
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any, Optional
|
|
|
|
|
|
@dataclass
|
|
class ClosedBatch:
|
|
batch_number: int
|
|
count: int
|
|
start_time: str
|
|
end_time: str
|
|
duration_sec: float
|
|
closed_by: str
|
|
persisted: bool
|
|
|
|
|
|
@dataclass
|
|
class CaseBatchTracker:
|
|
min_object_per_batch: int = 60
|
|
min_duration_per_batch: int = 60
|
|
ignore_batch_label: bool = False
|
|
current_batch_number: Optional[int] = None
|
|
current_batch_count: int = 0
|
|
current_start_time: Optional[str] = None
|
|
counted_ids: list[str] = field(default_factory=list)
|
|
closed_batches: list[ClosedBatch] = field(default_factory=list)
|
|
total_count: int = 0
|
|
_next_batch_number: int = 1
|
|
|
|
def display_total(self) -> int:
|
|
running = self.current_batch_count if self.current_batch_number else 0
|
|
return self.total_count + running
|
|
|
|
def _start_batch(self) -> bool:
|
|
self.current_batch_number = self._next_batch_number
|
|
self._next_batch_number += 1
|
|
self.current_batch_count = 0
|
|
self.current_start_time = datetime.now().isoformat()
|
|
self.counted_ids = []
|
|
return True
|
|
|
|
def _close_batch(self, closed_by: str) -> Optional[ClosedBatch]:
|
|
if self.current_batch_number is None:
|
|
return None
|
|
end_time = datetime.now().isoformat()
|
|
start_obj = datetime.fromisoformat(self.current_start_time or end_time)
|
|
end_obj = datetime.fromisoformat(end_time)
|
|
duration = (end_obj - start_obj).total_seconds()
|
|
count = self.current_batch_count
|
|
persisted = (
|
|
count >= self.min_object_per_batch
|
|
and duration >= self.min_duration_per_batch
|
|
)
|
|
closed = ClosedBatch(
|
|
batch_number=self.current_batch_number,
|
|
count=count,
|
|
start_time=self.current_start_time or end_time,
|
|
end_time=end_time,
|
|
duration_sec=duration,
|
|
closed_by=closed_by,
|
|
persisted=persisted,
|
|
)
|
|
self.closed_batches.append(closed)
|
|
if persisted:
|
|
self.total_count += count
|
|
self.current_batch_number = None
|
|
self.current_batch_count = 0
|
|
self.current_start_time = None
|
|
self.counted_ids = []
|
|
return closed
|
|
|
|
def record_ayam_crossing(self, track_id: int) -> tuple[int, bool]:
|
|
started_new = False
|
|
if self.current_batch_number is None:
|
|
self._start_batch()
|
|
started_new = True
|
|
key = str(track_id)
|
|
if key not in self.counted_ids:
|
|
self.counted_ids.append(key)
|
|
self.current_batch_count += 1
|
|
return self.current_batch_count, started_new
|
|
|
|
def record_talenan_crossing(self, track_id: int) -> Optional[ClosedBatch]:
|
|
if self.ignore_batch_label:
|
|
return None
|
|
self.ignore_batch_label = True
|
|
return self._close_batch(closed_by="talenan")
|
|
|
|
def finalize(self) -> None:
|
|
if self.current_batch_number is not None:
|
|
self._close_batch(closed_by="end_of_video")
|
|
|
|
def batches_summary(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"batch_number": b.batch_number,
|
|
"count": b.count,
|
|
"start_time": b.start_time,
|
|
"end_time": b.end_time,
|
|
"duration_sec": round(b.duration_sec, 1),
|
|
"closed_by": b.closed_by,
|
|
"persisted": b.persisted,
|
|
}
|
|
for b in self.closed_batches
|
|
]
|