Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import cv2 | |
| import numpy as np | |
| import tempfile | |
| import os | |
| import time | |
| import logging | |
| import sys | |
| import io | |
| from datetime import datetime | |
| from ultralytics import YOLO | |
| LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s" | |
| DATE_FORMAT = "%H:%M:%S" | |
| logging.basicConfig( | |
| level=logging.DEBUG, | |
| format=LOG_FORMAT, | |
| datefmt=DATE_FORMAT, | |
| handlers=[ | |
| logging.StreamHandler(sys.stdout), | |
| logging.FileHandler("people_counter.log", encoding="utf-8"), | |
| ] | |
| ) | |
| log_main = logging.getLogger("PeopleCounter.Main") | |
| log_model = logging.getLogger("PeopleCounter.Model") | |
| log_tracker = logging.getLogger("PeopleCounter.Tracker") | |
| log_video = logging.getLogger("PeopleCounter.Video") | |
| log_ui = logging.getLogger("PeopleCounter.UI") | |
| # Ultralytics verbose chiqishini bostiramiz | |
| logging.getLogger("ultralytics").setLevel(logging.WARNING) | |
| log_main.info("=" * 60) | |
| log_main.info(" People Counter — Ishga tushmoqda") | |
| log_main.info(f" Sana/Vaqt: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") | |
| log_main.info("=" * 60) | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| # MODEL YUKLASH — YOLOv11n | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| MODEL_NAME = "yolo11n.pt" # YOLOv11 nano | |
| PERSON_CLASS = 0 # COCO dataset: 0 = person | |
| log_model.info(f"Model yuklanmoqda: {MODEL_NAME}") | |
| log_model.info("YOLOv11n — A2-Attention mexanizmi, real-time surveillance grade.") | |
| log_model.info("COCO 80-klass, person = class index 0") | |
| _t0 = time.time() | |
| try: | |
| model = YOLO(MODEL_NAME) | |
| load_time = time.time() - _t0 | |
| log_model.info(f"Model muvaffaqiyatli yuklandi ({load_time:.2f}s)") | |
| log_model.debug(f" Model fayl : {MODEL_NAME}") | |
| log_model.debug(f" Task : {model.task}") | |
| except Exception as exc: | |
| log_model.error(f"Model yuklanmadi: {exc}") | |
| log_model.warning("Fallback: yolov8n.pt ga o'tilmoqda...") | |
| model = YOLO("yolov8n.pt") | |
| MODEL_NAME = "yolov8n.pt" | |
| log_model.info("Fallback model yuklandi: yolov8n.pt") | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| # IoU TRACKER | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| class IoUTracker: | |
| """ | |
| Sodda IoU-asosidagi odam tracker. | |
| Har bir frame'da: | |
| 1. Detection <-> Track IoU solishtirish. | |
| 2. Mos kelgan detection — eski track_id saqlanadi. | |
| 3. Mos kelmagan — yangi track_id beriladi. | |
| 4. max_lost frame ko'rinmasa — track o'chiriladi. | |
| """ | |
| def __init__(self, iou_threshold: float = 0.3, max_lost: int = 30): | |
| self.tracks = {} # {tid: {'bbox': [...], 'lost': int}} | |
| self.next_id = 1 | |
| self.unique_ids = set() | |
| self.iou_thr = iou_threshold | |
| self.max_lost = max_lost | |
| self.stat_match = 0 | |
| self.stat_new = 0 | |
| self.stat_del = 0 | |
| log_tracker.info( | |
| f"IoUTracker yaratildi | iou_thr={iou_threshold} | max_lost={max_lost}" | |
| ) | |
| def _iou(a, b) -> float: | |
| ax1, ay1, ax2, ay2 = a | |
| bx1, by1, bx2, by2 = b | |
| ix1 = max(ax1, bx1); iy1 = max(ay1, by1) | |
| ix2 = min(ax2, bx2); iy2 = min(ay2, by2) | |
| inter = max(0, ix2 - ix1) * max(0, iy2 - iy1) | |
| if inter == 0: | |
| return 0.0 | |
| union = (ax2-ax1)*(ay2-ay1) + (bx2-bx1)*(by2-by1) - inter | |
| return inter / union if union > 0 else 0.0 | |
| def update(self, detections: list, frame_idx: int = 0) -> dict: | |
| matched_tids = set() | |
| matched_didxs = set() | |
| # Step 1 — Match existing tracks | |
| for det_idx, det_bbox in enumerate(detections): | |
| best_iou, best_tid = 0.0, None | |
| for tid, tdata in self.tracks.items(): | |
| if tid in matched_tids: | |
| continue | |
| iou = self._iou(det_bbox, tdata['bbox']) | |
| if iou > best_iou: | |
| best_iou, best_tid = iou, tid | |
| if best_iou >= self.iou_thr and best_tid is not None: | |
| self.tracks[best_tid]['bbox'] = det_bbox | |
| self.tracks[best_tid]['lost'] = 0 | |
| matched_tids.add(best_tid) | |
| matched_didxs.add(det_idx) | |
| self.stat_match += 1 | |
| log_tracker.debug( | |
| f"F{frame_idx:04d} | Track #{best_tid:02d} MATCH iou={best_iou:.3f}" | |
| ) | |
| # Step 2 — New detections -> new tracks | |
| for det_idx, det_bbox in enumerate(detections): | |
| if det_idx not in matched_didxs: | |
| tid = self.next_id | |
| self.next_id += 1 | |
| self.tracks[tid] = {'bbox': det_bbox, 'lost': 0} | |
| self.unique_ids.add(tid) | |
| self.stat_new += 1 | |
| log_tracker.info( | |
| f"F{frame_idx:04d} | Track #{tid:02d} NEW " | |
| f"bbox={det_bbox} | Unique jami: {len(self.unique_ids)}" | |
| ) | |
| # Step 3 — Increment lost counter | |
| for tid in list(self.tracks.keys()): | |
| if tid not in matched_tids: | |
| self.tracks[tid]['lost'] += 1 | |
| # Step 4 — Prune dead tracks | |
| before = len(self.tracks) | |
| self.tracks = {t: v for t, v in self.tracks.items() | |
| if v['lost'] < self.max_lost} | |
| deleted = before - len(self.tracks) | |
| if deleted: | |
| self.stat_del += deleted | |
| log_tracker.debug( | |
| f"F{frame_idx:04d} | {deleted} track silindi (max_lost={self.max_lost})" | |
| ) | |
| active = {t: v['bbox'] for t, v in self.tracks.items() if v['lost'] == 0} | |
| if frame_idx % 25 == 0: | |
| log_tracker.info( | |
| f"F{frame_idx:04d} | Aktiv={len(active):2d} " | |
| f"Unique={len(self.unique_ids):2d} " | |
| f"Tracks={len(self.tracks):2d}" | |
| ) | |
| return active | |
| def summary(self): | |
| return { | |
| "unique_people" : len(self.unique_ids), | |
| "stat_match" : self.stat_match, | |
| "stat_new" : self.stat_new, | |
| "stat_del" : self.stat_del, | |
| } | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| # ASOSIY HISOBLASH FUNKSIYASI | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| COLORS = [ | |
| (255, 80, 80), (80, 200, 120), (80, 160, 255), | |
| (255,200, 50), (200, 80, 255), (50, 220, 220), | |
| (255,140, 0), ( 0, 200, 200), (180,255, 80), | |
| (255, 80, 180), | |
| ] | |
| def count_people( | |
| video_path: str, | |
| conf_threshold: float = 0.4, | |
| progress=gr.Progress(), | |
| stream_handler=None, | |
| ) -> tuple: | |
| session = datetime.now().strftime("%H%M%S") | |
| log_main.info("─" * 55) | |
| log_main.info(f"[{session}] YANGI SESSION boshlandi") | |
| log_main.info(f"[{session}] Conf threshold : {conf_threshold}") | |
| log_main.info(f"[{session}] Model : {MODEL_NAME}") | |
| if video_path is None: | |
| log_main.warning("Video yuklanmadi.") | |
| return None, "Video yuklanmadi." | |
| # ── Video ochish | |
| log_video.info(f"Video ochilmoqda ...") | |
| cap = cv2.VideoCapture(video_path) | |
| if not cap.isOpened(): | |
| log_video.error("Video fayl ochilmadi!") | |
| return None, "Video ochilmadi." | |
| total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) | |
| fps = cap.get(cv2.CAP_PROP_FPS) or 25.0 | |
| width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) | |
| height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) | |
| dur = total_frames / fps | |
| log_video.info(f" Kadrlar : {total_frames}") | |
| log_video.info(f" FPS : {fps:.1f}") | |
| log_video.info(f" Hajm : {width}x{height} px") | |
| log_video.info(f" Davomiylik: {dur:.1f}s") | |
| # ── Output video | |
| out_path = tempfile.mktemp(suffix="_result.mp4") | |
| fourcc = cv2.VideoWriter_fourcc(*"mp4v") | |
| writer = cv2.VideoWriter(out_path, fourcc, fps, (width, height)) | |
| log_video.info(f"Output yozuvchi ochildi -> {os.path.basename(out_path)}") | |
| # ── Tracker | |
| max_lost = int(fps * 1.5) | |
| tracker = IoUTracker(iou_threshold=0.3, max_lost=max_lost) | |
| frame_idx = 0 | |
| total_dets = 0 | |
| t_start = time.time() | |
| log_main.info("Frame loop boshlandi ...") | |
| log_main.info(f" Har {25} ta kadrda tracker holati ko'rsatiladi.") | |
| while True: | |
| ret, frame = cap.read() | |
| if not ret: | |
| break | |
| # ── YOLO inference | |
| t_inf = time.time() | |
| results = model(frame, classes=[PERSON_CLASS], | |
| conf=conf_threshold, verbose=False)[0] | |
| inf_ms = (time.time() - t_inf) * 1000 | |
| detections = [] | |
| for box in results.boxes: | |
| x1, y1, x2, y2 = map(int, box.xyxy[0]) | |
| conf_val = float(box.conf[0]) | |
| detections.append([x1, y1, x2, y2]) | |
| log_model.debug( | |
| f"F{frame_idx:04d} | bbox=[{x1},{y1},{x2},{y2}] " | |
| f"conf={conf_val:.3f}" | |
| ) | |
| total_dets += len(detections) | |
| if frame_idx % 25 == 0: | |
| log_model.info( | |
| f"F{frame_idx:04d} | det={len(detections):2d} | " | |
| f"inf={inf_ms:5.1f}ms" | |
| ) | |
| # ── Tracking | |
| active = tracker.update(detections, frame_idx) | |
| # ── Frame annotatsiyasi | |
| for tid, (x1, y1, x2, y2) in active.items(): | |
| color = COLORS[tid % len(COLORS)] | |
| cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2) | |
| lbl = f"#{tid}" | |
| (tw, th), _ = cv2.getTextSize(lbl, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2) | |
| cv2.rectangle(frame, (x1, y1-th-8), (x1+tw+6, y1), color, -1) | |
| cv2.putText(frame, lbl, (x1+3, y1-4), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255,255,255), 2) | |
| # ── Overlay panel (yuqori chap) | |
| total_unique = len(tracker.unique_ids) | |
| currently = len(active) | |
| cv2.rectangle(frame, (8, 8), (330, 80), (12,12,18), -1) | |
| cv2.rectangle(frame, (8, 8), (330, 80), (50,230,120), 1) | |
| cv2.putText(frame, f"Jami: {total_unique} ta odam", | |
| (14, 37), cv2.FONT_HERSHEY_SIMPLEX, 0.82, (50,230,120), 2) | |
| cv2.putText(frame, f"Hozir: {currently} Frame: {frame_idx}", | |
| (14, 68), cv2.FONT_HERSHEY_SIMPLEX, 0.52, (160,160,160), 1) | |
| # ── Model tegi (quyi o'ng) | |
| cv2.putText(frame, f"{MODEL_NAME} + IoUTracker", | |
| (width-230, height-10), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.42, (80,80,80), 1) | |
| writer.write(frame) | |
| frame_idx += 1 | |
| if total_frames > 0: | |
| progress( | |
| frame_idx / total_frames, | |
| desc=f"Frame {frame_idx}/{total_frames} | Unique: {total_unique} odam" | |
| ) | |
| cap.release() | |
| writer.release() | |
| elapsed = time.time() - t_start | |
| avg_fps_p = frame_idx / elapsed if elapsed > 0 else 0 | |
| stats = tracker.summary() | |
| log_main.info("─" * 55) | |
| log_main.info(f"[{session}] YAKUNIY STATISTIKA") | |
| log_main.info(f" Jami kadrlar : {frame_idx}") | |
| log_main.info(f" Jami detectionlar: {total_dets}") | |
| log_main.info(f" Unique odamlar : {stats['unique_people']}") | |
| log_main.info(f" Track match : {stats['stat_match']}") | |
| log_main.info(f" Yangi track : {stats['stat_new']}") | |
| log_main.info(f" O'chirilgan track: {stats['stat_del']}") | |
| log_main.info(f" Ishlash vaqti : {elapsed:.2f}s") | |
| log_main.info(f" O'rtacha tezlik : {avg_fps_p:.1f} fps") | |
| log_main.info("─" * 55) | |
| n = stats['unique_people'] | |
| if n == 0: | |
| result = "Odam yo'q" | |
| elif n == 1: | |
| result = "1 ta odam" | |
| else: | |
| result = f"{n} ta odam" | |
| log_main.info(f"[{session}] NATIJA: {result}") | |
| return out_path, result | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| # GRADIO UI | |
| # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| log_ui.info("Gradio interfeysi qurilmoqda ...") | |
| css = """ | |
| @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;700&family=Space+Grotesk:wght@400;600;700&display=swap'); | |
| body { font-family: 'Space Grotesk', sans-serif; } | |
| #ttl h1 { font-family:'JetBrains Mono',monospace; color:#50e37c; text-align:center; } | |
| #ttl p { text-align:center; color:#888; font-size:.9rem; } | |
| #rbox { | |
| font-family:'JetBrains Mono',monospace; font-size:2rem; font-weight:700; | |
| text-align:center; padding:1.4rem; background:#0d1a12; color:#50e37c; | |
| border-radius:10px; border:1px solid #50e37c55; margin-top:8px; | |
| } | |
| #logbox textarea { | |
| font-family:'JetBrains Mono',monospace !important; | |
| font-size:.7rem !important; | |
| background:#080d08 !important; | |
| color:#7ddb8a !important; | |
| border:1px solid #1a2e1a !important; | |
| } | |
| """ | |
| with gr.Blocks(css=css, title="People Counter — YOLOv11n") as demo: | |
| gr.Markdown( | |
| "# 👁️ Videodagi Odamlar Sonini Hisoblash\n" | |
| "**YOLOv11n** (A²-Attention Modules) + **IoU Tracker** — surveillance grade", | |
| elem_id="ttl" | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| video_in = gr.Video(label="📹 Video yuklang", sources=["upload"]) | |
| conf_sl = gr.Slider(0.2, 0.85, value=0.4, step=0.05, | |
| label="Conf threshold") | |
| run_btn = gr.Button("▶ Hisoblashni boshlash", | |
| variant="primary", size="lg") | |
| with gr.Column(scale=1): | |
| video_out = gr.Video(label="📊 Annotated video") | |
| result_htm = gr.HTML("<div id='rbox'>— natija —</div>") | |
| def run(video, conf): | |
| log_ui.info(f"UI: Run bosildi | conf={conf}") | |
| # In-memory stream handler to capture logs for UI | |
| buf = io.StringIO() | |
| handler = logging.StreamHandler(buf) | |
| handler.setLevel(logging.DEBUG) | |
| handler.setFormatter( | |
| logging.Formatter( | |
| "%(asctime)s | %(levelname)-8s | %(name)-18s | %(message)s", | |
| datefmt="%H:%M:%S" | |
| ) | |
| ) | |
| root_pc = logging.getLogger("PeopleCounter") | |
| root_pc.addHandler(handler) | |
| try: | |
| out_vid, text = count_people(video, conf) | |
| finally: | |
| root_pc.removeHandler(handler) | |
| html = f"<div id='rbox'>{'✅ ' if 'ta odam' in text or '1 ta' in text else '🚶 '}{text}</div>" | |
| return out_vid, html, buf.getvalue() | |
| run_btn.click( | |
| fn=run, | |
| inputs=[video_in, conf_sl], | |
| outputs=[video_out, result_htm] | |
| ) | |
| log_ui.info("Gradio tayyor.") | |
| if __name__ == "__main__": | |
| log_main.info("demo.launch() ...") | |
| demo.launch() |