Update pipeline_full.py

pipeline_full.py

@@ -390,6 +390,8 @@ def train_team_classifier_on_video(video_path: str, stride: int = 30) -> None:
 
 
 def resolve_goalkeepers_team_id(players: sv.Detections, goalkeepers: sv.Detections) -> np.ndarray:
+    if len(goalkeepers) == 0 or len(players) == 0:
+        return np.array([])
     goalkeepers_xy = goalkeepers.get_anchors_coordinates(sv.Position.BOTTOM_CENTER)
     players_xy = players.get_anchors_coordinates(sv.Position.BOTTOM_CENTER)
     team_0_centroid = players_xy[players.class_id == 0].mean(axis=0)
@@ -739,7 +741,7 @@ def step_ball_path(video_path: str, out_dir: str) -> Dict[str, Any]:
     }
 
 
-# -------------------- 8. NEW: full-match analysis + event-annotated video --------------------
+# -------------------- 8. full-match analysis + event-annotated video --------------------
 
 
 def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str, Any]:
@@ -748,15 +750,23 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
       * tracks players & ball
       * computes distance & speed per player (pitch coordinates)
       * estimates ball possession per team & per player
-      * detects simple events (pass, tackle/interception, clearance, shot)
-      * renders an annotated MP4 with overlays
+      * estimates time spent in defensive/middle/attacking thirds
+      * detects simple events:
+          - passes (successful between teammates)
+          - tackles / interceptions (winning ball from opponent)
+          - clearances
+          - shots (high-speed ball towards goal)
+      * renders an annotated MP4 with overlays:
+          - per-player labels: id, team, speed, distance
+          - possession HUD per team
+          - event banners
     """
     ensure_models_loaded()
     os.makedirs(out_dir, exist_ok=True)
 
     video_info = sv.VideoInfo.from_video_path(video_path)
     fps = video_info.fps
-    dt = 1.0 / fps
+    dt = 1.0 / max(fps, 1.0)
 
     tracker = sv.ByteTrack()
     tracker.reset()
@@ -768,11 +778,30 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
     distance_covered_m = defaultdict(float)        # tid -> meters
     possession_time_player = defaultdict(float)    # tid -> seconds
     possession_time_team = defaultdict(float)      # team_id -> seconds
-    team_of_player = {}                            # tid -> team_id
+    team_of_player: Dict[int, int] = {}            # tid -> team_id
+
+    # per-player richer stats for coaches
+    player_stats: Dict[int, Dict[str, Any]] = defaultdict(
+        lambda: {
+            "distance_m": 0.0,
+            "max_speed_kmh": 0.0,
+            "time_def_third_s": 0.0,
+            "time_mid_third_s": 0.0,
+            "time_att_third_s": 0.0,
+            "touches": 0,
+            "successful_passes": 0,
+            "received_passes": 0,
+            "shots": 0,
+            "tackles": 0,
+            "interceptions": 0,
+            "clearances": 0,
+        }
+    )
+
     events: List[Dict[str, Any]] = []
 
-    # last positions for speed / distance
-    last_pitch_pos: Dict[int, np.ndarray] = {}
+    # last positions for speed / distance (per frame)
+    prev_positions: Dict[int, np.ndarray] = {}
     prev_owner_tid: Optional[int] = None
     prev_ball_pos_pitch: Optional[np.ndarray] = None
 
@@ -808,8 +837,9 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
     frame_generator = sv.get_video_frames_generator(video_path)
 
     with sink:
-        for frame_idx, frame in enumerate(tqdm(frame_generator, total=video_info.total_frames,
-                                               desc="analyze + annotate")):
+        for frame_idx, frame in enumerate(
+            tqdm(frame_generator, total=video_info.total_frames, desc="analyze + annotate")
+        ):
             t = frame_idx * dt
 
             # --- detections + tracking ---
@@ -834,6 +864,14 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
             frame_ref = key_points.xy[0][filt]
             pitch_ref = np.array(PITCH_CONFIG.vertices)[filt]
 
+            if len(frame_ref) < 4:
+                # Not enough field points: just draw detections and skip advanced stats
+                annotated = frame.copy()
+                annotated = ellipse_annotator.annotate(scene=annotated, detections=players_dets)
+                annotated = triangle_annotator.annotate(scene=annotated, detections=ball_dets)
+                sink.write_frame(annotated)
+                continue
+
             transformer = ViewTransformer(source=frame_ref, target=pitch_ref)
             Ms.append(transformer.m)
             transformer.m = np.mean(np.array(Ms), axis=0)
@@ -841,31 +879,55 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
             # --- team classification & pitch positions ---
             frame_players_xy_pitch = None
             frame_ball_pos_pitch = None
+            current_positions: Dict[int, np.ndarray] = {}
+            current_speed_kmh: Dict[int, float] = {}
 
             if len(players_dets) > 0:
                 crops = [sv.crop_image(frame, xyxy) for xyxy in players_dets.xyxy]
                 team_preds = TEAM_CLASSIFIER.predict(crops)
                 players_dets.class_id = team_preds  # now class_id = team_id (0/1)
 
-                frame_players_xy_img = players_dets.get_anchors_coordinates(sv.Position.BOTTOM_CENTER)
-                frame_players_xy_pitch = transformer.transform_points(points=frame_players_xy_img)
+                frame_players_xy_img = players_dets.get_anchors_coordinates(
+                    sv.Position.BOTTOM_CENTER
+                )
+                frame_players_xy_pitch = transformer.transform_points(
+                    points=frame_players_xy_img
+                )
+
+                pitch_length = PITCH_CONFIG.length
 
                 for tid, team_id, pos_pitch in zip(
                     players_dets.tracker_id, players_dets.class_id, frame_players_xy_pitch
                 ):
                     tid_int = int(tid)
                     team_of_player[tid_int] = int(team_id)
+                    current_positions[tid_int] = pos_pitch
 
-                    prev_pos = last_pitch_pos.get(tid_int)
+                    prev_pos = prev_positions.get(tid_int)
                     speed_kmh = 0.0
                     if prev_pos is not None:
                         dist_m = float(np.linalg.norm(pos_pitch - prev_pos))
                         distance_covered_m[tid_int] += dist_m
-                        speed_kmh = (dist_m / dt) * 3.6  # m/s -> km/h
-                    last_pitch_pos[tid_int] = pos_pitch
+                        player_stats[tid_int]["distance_m"] += dist_m
+                        speed_kmh = (dist_m / dt) * 3.6
+                        player_stats[tid_int]["max_speed_kmh"] = max(
+                            player_stats[tid_int]["max_speed_kmh"], speed_kmh
+                        )
+                    current_speed_kmh[tid_int] = speed_kmh
+
+                    # zone times: defensive / middle / attacking thirds
+                    x_pos = pos_pitch[0]
+                    if x_pos < pitch_length / 3.0:
+                        player_stats[tid_int]["time_def_third_s"] += dt
+                    elif x_pos < 2.0 * pitch_length / 3.0:
+                        player_stats[tid_int]["time_mid_third_s"] += dt
+                    else:
+                        player_stats[tid_int]["time_att_third_s"] += dt
 
             if len(ball_dets) > 0:
-                frame_ball_xy_img = ball_dets.get_anchors_coordinates(sv.Position.BOTTOM_CENTER)
+                frame_ball_xy_img = ball_dets.get_anchors_coordinates(
+                    sv.Position.BOTTOM_CENTER
+                )
                 frame_ball_xy_pitch = transformer.transform_points(points=frame_ball_xy_img)
                 frame_ball_pos_pitch = frame_ball_xy_pitch[0]
 
@@ -886,22 +948,28 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                 if owner_team is not None:
                     possession_time_team[owner_team] += dt
 
-            # --- event detection (simple heuristics) ---
+            # --- helper to register events & banner text ---
             def register_event(ev: Dict[str, Any], text: str):
                 nonlocal current_event_text, event_text_frames_left
                 events.append(ev)
-                current_event_text = text
-                event_text_frames_left = EVENT_TEXT_DURATION_FRAMES
+                if text:
+                    current_event_text = text
+                    event_text_frames_left = EVENT_TEXT_DURATION_FRAMES
 
-            # possession change events, passes, tackles, interceptions
+            # --- possession change events, passes, tackles, interceptions ---
             if owner_tid != prev_owner_tid:
+                if owner_tid is not None:
+                    player_stats[owner_tid]["touches"] += 1
+
                 if owner_tid is not None and prev_owner_tid is not None:
                     prev_team = team_of_player.get(prev_owner_tid)
                     cur_team = team_of_player.get(owner_tid)
 
                     travel_m = 0.0
                     if prev_ball_pos_pitch is not None and frame_ball_pos_pitch is not None:
-                        travel_m = float(np.linalg.norm(frame_ball_pos_pitch - prev_ball_pos_pitch))
+                        travel_m = float(
+                            np.linalg.norm(frame_ball_pos_pitch - prev_ball_pos_pitch)
+                        )
 
                     MIN_PASS_TRAVEL_M = 3.0
 
@@ -919,15 +987,15 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                                 },
                                 f"Pass: #{prev_owner_tid} → #{owner_tid} (Team {cur_team})",
                             )
+                            player_stats[prev_owner_tid]["successful_passes"] += 1
+                            player_stats[owner_tid]["received_passes"] += 1
                         elif prev_team != cur_team:
                             # tackle vs interception
                             d_pp = 999.0
-                            if frame_players_xy_pitch is not None:
-                                # get current positions
-                                pos_prev = last_pitch_pos.get(int(prev_owner_tid))
-                                pos_cur = last_pitch_pos.get(int(owner_tid))
-                                if pos_prev is not None and pos_cur is not None:
-                                    d_pp = float(np.linalg.norm(pos_prev - pos_cur))
+                            pos_prev = prev_positions.get(int(prev_owner_tid))
+                            pos_cur = current_positions.get(int(owner_tid))
+                            if pos_prev is not None and pos_cur is not None:
+                                d_pp = float(np.linalg.norm(pos_prev - pos_cur))
                             ev_type = "tackle" if d_pp < 3.0 else "interception"
                             label = "Tackle" if ev_type == "tackle" else "Interception"
                             register_event(
@@ -937,10 +1005,17 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                                     "from_tid": int(prev_owner_tid),
                                     "to_tid": int(owner_tid),
                                     "team_id": int(cur_team),
-                                    "extra": {"player_distance_m": d_pp, "ball_travel_m": travel_m},
+                                    "extra": {
+                                        "player_distance_m": d_pp,
+                                        "ball_travel_m": travel_m,
+                                    },
                                 },
                                 f"{label}: #{owner_tid} wins ball from #{prev_owner_tid}",
                             )
+                            if ev_type == "tackle":
+                                player_stats[owner_tid]["tackles"] += 1
+                            else:
+                                player_stats[owner_tid]["interceptions"] += 1
 
                 # generic possession-change event
                 register_event(
@@ -949,13 +1024,15 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                         "t": float(t),
                         "from_tid": int(prev_owner_tid) if prev_owner_tid is not None else None,
                         "to_tid": int(owner_tid) if owner_tid is not None else None,
-                        "team_id": int(team_of_player.get(owner_tid)) if owner_tid is not None else None,
+                        "team_id": int(team_of_player.get(owner_tid))
+                        if owner_tid is not None
+                        else None,
                         "extra": {},
                     },
                     "" if owner_tid is None else f"Team {team_of_player.get(owner_tid)} in possession",
                 )
 
-            # shot / clearance based on ball speed & direction
+            # --- shot / clearance based on ball speed & direction ---
             if (
                 prev_ball_pos_pitch is not None
                 and frame_ball_pos_pitch is not None
@@ -964,7 +1041,7 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                 v = (frame_ball_pos_pitch - prev_ball_pos_pitch) / dt  # m/s
                 speed_mps = float(np.linalg.norm(v))
                 speed_kmh = speed_mps * 3.6
-                HIGH_SPEED_KMH = 18.0
+                HIGH_SPEED_KMH = 18.0  # threshold for "hard" actions
 
                 if speed_kmh > HIGH_SPEED_KMH:
                     shooter_team = team_of_player.get(owner_tid)
@@ -988,6 +1065,7 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                                 },
                                 f"Shot by #{owner_tid} (Team {shooter_team}) – {speed_kmh:.1f} km/h",
                             )
+                            player_stats[owner_tid]["shots"] += 1
                         else:
                             register_event(
                                 {
@@ -1000,25 +1078,23 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
                                 },
                                 f"Clearance by #{owner_tid} (Team {shooter_team})",
                             )
+                            player_stats[owner_tid]["clearances"] += 1
 
             prev_owner_tid = owner_tid
             prev_ball_pos_pitch = frame_ball_pos_pitch
+            prev_positions = current_positions
 
             # --- frame drawing ---
             annotated = frame.copy()
 
-            # build labels for players: id + speed + distance
+            # build labels for players: id + team + current speed + total distance
             player_labels: List[str] = []
             if frame_players_xy_pitch is not None and len(players_dets) > 0:
                 for tid, pos_pitch in zip(players_dets.tracker_id, frame_players_xy_pitch):
                     tid_int = int(tid)
-                    prev_pos = last_pitch_pos.get(tid_int)
-                    speed_kmh = 0.0
-                    if prev_pos is not None:
-                        dist_m = float(np.linalg.norm(pos_pitch - prev_pos))
-                        speed_kmh = (dist_m / dt) * 3.6
-                    d_total = distance_covered_m[tid_int]
                     team_id = team_of_player.get(tid_int, -1)
+                    speed_kmh = current_speed_kmh.get(tid_int, 0.0)
+                    d_total = distance_covered_m[tid_int]
                     player_labels.append(
                         f"#{tid_int} T{team_id} {speed_kmh:4.1f} km/h {d_total:.1f} m"
                     )
@@ -1035,10 +1111,21 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
 
             # --- HUD: possession percentages ---
             total_poss_time = sum(possession_time_team.values()) + 1e-6
-            team0_pct = 100.0 * possession_time_team.get(0, 0.0) / total_poss_time
-            team1_pct = 100.0 * possession_time_team.get(1, 0.0) / total_poss_time
+            team0_pct = (
+                100.0 * possession_time_team.get(0, 0.0) / total_poss_time
+                if total_poss_time > 0
+                else 0.0
+            )
+            team1_pct = (
+                100.0 * possession_time_team.get(1, 0.0) / total_poss_time
+                if total_poss_time > 0
+                else 0.0
+            )
 
-            hud_text = f"Team 0 Ball Control: {team0_pct:5.2f}%   Team 1 Ball Control: {team1_pct:5.2f}%"
+            hud_text = (
+                f"Team 0 Ball Control: {team0_pct:5.2f}%   "
+                f"Team 1 Ball Control: {team1_pct:5.2f}%"
+            )
             cv2.rectangle(
                 annotated,
                 (20, annotated.shape[0] - 60),
@@ -1059,7 +1146,9 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
 
             # --- event banner ---
             if event_text_frames_left > 0 and current_event_text:
-                cv2.rectangle(annotated, (20, 20), (annotated.shape[1] - 20, 90), (255, 255, 255), -1)
+                cv2.rectangle(
+                    annotated, (20, 20), (annotated.shape[1] - 20, 90), (255, 255, 255), -1
+                )
                 cv2.putText(
                     annotated,
                     current_event_text,
@@ -1082,10 +1171,21 @@ def step_analyze_and_annotate_video(video_path: str, out_dir: str) -> Dict[str,
 
     stats = {
         "distance_covered_m": {str(tid): float(d) for tid, d in distance_covered_m.items()},
-        "possession_time_player_s": {str(tid): float(t_sec) for tid, t_sec in possession_time_player.items()},
-        "possession_time_team_s": {int(team): float(t_sec) for team, t_sec in possession_time_team.items()},
+        "possession_time_player_s": {
+            str(tid): float(t_sec) for tid, t_sec in possession_time_player.items()
+        },
+        "possession_time_team_s": {
+            int(team): float(t_sec) for team, t_sec in possession_time_team.items()
+        },
         "possession_percent_team": possession_percent_team,
         "team_of_player": {str(tid): int(team) for tid, team in team_of_player.items()},
+        "player_stats": {
+            str(tid): {
+                k: float(v) if isinstance(v, (int, float)) else v
+                for k, v in stats_dict.items()
+            }
+            for tid, stats_dict in player_stats.items()
+        },
     }
 
     return {
@@ -1124,7 +1224,11 @@ def run_full_pipeline(video_path: str, job_dir: str) -> Dict[str, Any]:
     update_progress("ball_path", 0.60, "Computing ball path and heatmap...")
     ball_paths = step_ball_path(video_path, os.path.join(job_dir, "ball_path"))
 
-    update_progress("events_video", 0.80, "Analyzing match and rendering event-annotated video...")
+    update_progress(
+        "events_video",
+        0.80,
+        "Analyzing match, computing speed/distance, and rendering event-annotated video...",
+    )
     analysis_out = step_analyze_and_annotate_video(
         video_path, os.path.join(job_dir, "analysis")
     )