forked from zakaria/chicken-counting-sukawarna-det
124 lines
4.9 KiB
Python
Executable File
124 lines
4.9 KiB
Python
Executable File
"""Run CC1–CC4 sequentially, then compress videos and write the JSON report."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date as date_type
|
||
from pathlib import Path
|
||
|
||
from chicken_counter.batch_discovery import discover_camera_videos
|
||
from chicken_counter.compress import compress_video_to_target
|
||
from chicken_counter.config import BatchSettings, build_camera_config_from_batch
|
||
from chicken_counter.pipeline import run_pipeline
|
||
from chicken_counter.report import build_batch_report, persist_batch_reports
|
||
from chicken_counter.tracking import DetectionTracker
|
||
from chicken_counter.types import CameraBatchResult
|
||
|
||
|
||
def run_daily_batch(settings: BatchSettings, date: str | None = None, *, verbose: bool = False, no_video: bool = False, show_progress: bool = False) -> Path:
|
||
run_date = date or date_type.today().isoformat()
|
||
day_dir = Path(settings.batch.root_dir) / run_date
|
||
output_dir = day_dir / settings.batch.output_subdir
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
print(f"[batch] starting daily run for {run_date}")
|
||
print(f"[batch] input folder: {day_dir}")
|
||
print(f"[batch] output folder: {output_dir}")
|
||
if no_video:
|
||
print("[batch] --no-video: skipping video output, overlay, and compression")
|
||
|
||
discovery = discover_camera_videos(day_dir, settings)
|
||
camera_order = sorted(settings.cameras.items(), key=lambda item: item[1].camera_num)
|
||
|
||
first_camera_id = next(
|
||
camera_id for camera_id, _preset in camera_order if camera_id in discovery.found
|
||
)
|
||
first_source = discovery.found[first_camera_id]
|
||
init_output_path = output_dir / f"{first_camera_id}_vis.mp4" if not no_video else None
|
||
init_config = build_camera_config_from_batch(
|
||
settings,
|
||
first_camera_id,
|
||
source=first_source,
|
||
output_path=init_output_path,
|
||
checkpoint_dir=output_dir / "checkpoints" / first_camera_id,
|
||
)
|
||
shared_tracker = DetectionTracker(init_config)
|
||
|
||
camera_results: list[CameraBatchResult] = []
|
||
report_path = output_dir / f"counts_{run_date}.json"
|
||
|
||
for camera_id, _preset in camera_order:
|
||
if camera_id in discovery.skipped:
|
||
skip_reason = discovery.skipped[camera_id]
|
||
print(f"[batch] skipping {camera_id}: {skip_reason}")
|
||
camera_results.append(
|
||
CameraBatchResult(
|
||
camera_id=camera_id,
|
||
skipped=True,
|
||
skip_reason=skip_reason,
|
||
)
|
||
)
|
||
persist_batch_reports(run_date, camera_results, output_dir)
|
||
continue
|
||
|
||
source_path = discovery.found[camera_id]
|
||
vis_path = output_dir / f"{camera_id}_vis.mp4" if not no_video else None
|
||
checkpoint_dir = output_dir / "checkpoints" / camera_id
|
||
|
||
print(f"[batch] processing {camera_id} from {source_path.name}")
|
||
camera_config = build_camera_config_from_batch(
|
||
settings,
|
||
camera_id,
|
||
source=source_path,
|
||
output_path=vis_path,
|
||
checkpoint_dir=checkpoint_dir,
|
||
)
|
||
camera_config.performance.verbose = verbose
|
||
pipeline_result = run_pipeline(camera_config, tracker=shared_tracker, show_progress=show_progress)
|
||
camera_results.append(
|
||
CameraBatchResult(
|
||
camera_id=camera_id,
|
||
pipeline=pipeline_result,
|
||
)
|
||
)
|
||
print(
|
||
f"[batch] finished {camera_id}: total_entered={pipeline_result.total_entered_count} "
|
||
f"frames={pipeline_result.frames_processed} reason={pipeline_result.stopped_reason}"
|
||
)
|
||
persist_batch_reports(run_date, camera_results, output_dir)
|
||
|
||
if no_video:
|
||
report = build_batch_report(run_date, camera_results, output_dir=output_dir)
|
||
print(
|
||
f"[batch] complete for {run_date}: total_entered_sum={report.total_entered_sum} "
|
||
f"report={report_path}"
|
||
)
|
||
return report_path
|
||
|
||
print("[batch] all cameras complete; starting compression")
|
||
for item in camera_results:
|
||
if item.skipped or item.pipeline is None:
|
||
continue
|
||
vis_path = item.pipeline.vis_video_path
|
||
if not vis_path:
|
||
continue
|
||
compressed_path = output_dir / f"{item.camera_id}_compressed.mp4"
|
||
size_mb = compress_video_to_target(
|
||
vis_path,
|
||
compressed_path,
|
||
max_mb=settings.batch.compress_max_mb,
|
||
)
|
||
item.compressed_video_path = str(compressed_path)
|
||
item.compressed_size_mb = size_mb
|
||
|
||
if settings.batch.delete_intermediate:
|
||
Path(vis_path).unlink(missing_ok=True)
|
||
|
||
persist_batch_reports(run_date, camera_results, output_dir)
|
||
|
||
report = build_batch_report(run_date, camera_results, output_dir=output_dir)
|
||
print(
|
||
f"[batch] complete for {run_date}: total_entered_sum={report.total_entered_sum} "
|
||
f"report={report_path}"
|
||
)
|
||
return report_path
|