Commiting with problems in tracking.
This commit is contained in:
1 parent
152b3b9da5
commit
2c798c8ca5
5 files changed
+105
-49
No files matched your search
@@ -1,17 +1,17 @@
|
|||||||
import time
|
|
||||||
from contextlib import closing
|
from contextlib import closing
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from dodgeballtracker.helper import time_tracker
|
from dodgeballtracker.helper import time_tracker
|
||||||
from dodgeballtracker.logic import (DetectedObjectBox,
|
from dodgeballtracker.logic import (DodgeballTracker,
|
||||||
DodgeballTracker,
|
|
||||||
get_court_perspective_transform)
|
get_court_perspective_transform)
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import logging
|
import logging
|
||||||
import click
|
import click
|
||||||
from typing import ContextManager
|
from typing import ContextManager
|
||||||
|
|
||||||
from dodgeballtracker.video_writer import (MP4VVideoWriter, VideoWriter,
|
from dodgeballtracker.video_writer import (MP4VVideoWriter,
|
||||||
|
VideoWriter,
|
||||||
|
DetectedObjectBox,
|
||||||
MP4VVideoWriterWithFrame)
|
MP4VVideoWriterWithFrame)
|
||||||
|
|
||||||
here = Path(__file__).parent
|
here = Path(__file__).parent
|
||||||
@@ -34,10 +34,11 @@ def get_output(input_: Path, proposed_output: Path | None) -> Path:
|
|||||||
def get_video_writer(display_frame: bool,
|
def get_video_writer(display_frame: bool,
|
||||||
video_input: Path,
|
video_input: Path,
|
||||||
video_output: Path) -> ContextManager[VideoWriter]:
|
video_output: Path) -> ContextManager[VideoWriter]:
|
||||||
|
thresholds = list(np.arange(0.5, 1, 0.05))
|
||||||
if display_frame:
|
if display_frame:
|
||||||
return closing(
|
return closing(
|
||||||
MP4VVideoWriterWithFrame.from_frame(video_input, video_output))
|
MP4VVideoWriterWithFrame.from_frame(video_input, video_output, thresholds))
|
||||||
return closing(MP4VVideoWriter.from_frame(video_input, video_output))
|
return closing(MP4VVideoWriter.from_frame(video_input, video_output, thresholds))
|
||||||
|
|
||||||
|
|
||||||
@cli.command()
|
@cli.command()
|
||||||
|
|||||||
+18
-33
@@ -1,16 +1,14 @@
|
|||||||
import json
|
import json
|
||||||
from collections import Counter
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
import logging
|
import logging
|
||||||
import torch
|
import torch
|
||||||
from ultralytics import YOLO
|
from ultralytics import YOLO
|
||||||
import cv2
|
import cv2
|
||||||
from dodgeballtracker.helper import iter_frames
|
from dodgeballtracker.helper import iter_frames
|
||||||
from dodgeballtracker.video_writer import create_video_writer, VideoWriter
|
from dodgeballtracker.video_writer import VideoWriter, DetectedObjectBox
|
||||||
import matplotlib
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
from boxmot import DeepOcSort as DeepOCSORT
|
from boxmot import DeepOcSort as DeepOCSORT
|
||||||
from .typing import VideoFrame
|
from .typing import VideoFrame, DetectedObject
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -28,21 +26,6 @@ def get_best_available_torch_device() -> torch.device:
|
|||||||
return torch.device('cpu')
|
return torch.device('cpu')
|
||||||
|
|
||||||
|
|
||||||
class DetectedObjectBox:
|
|
||||||
def __init__(self, min_threshold: float, color: Color):
|
|
||||||
self.min_threshold = min_threshold
|
|
||||||
self.color = color
|
|
||||||
|
|
||||||
def __lt__(self, other):
|
|
||||||
return self.min_threshold < other.min_threshold
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_viridis(cls, min_threshold):
|
|
||||||
cm = matplotlib.colormaps['viridis']
|
|
||||||
color: Color = tuple((int(255 * cm(min_threshold)[i]) for i in range(3)))
|
|
||||||
return cls(min_threshold, color)
|
|
||||||
|
|
||||||
|
|
||||||
def get_court_perspective_transform():
|
def get_court_perspective_transform():
|
||||||
"""
|
"""
|
||||||
"""
|
"""
|
||||||
@@ -117,13 +100,17 @@ class DodgeballTracker:
|
|||||||
for detected_box_object in self.detected_box_objects:
|
for detected_box_object in self.detected_box_objects:
|
||||||
|
|
||||||
if detected_box_object.min_threshold < confidence:
|
if detected_box_object.min_threshold < confidence:
|
||||||
xmin, ymin, xmax, ymax = self.extract_box(data)
|
# xmin, ymin, xmax, ymax = self.extract_box(data)
|
||||||
cv2.rectangle(frame, (xmin, ymin), (xmax, ymax), detected_box_object.color, 2)
|
# cv2.rectangle(frame, (xmin, ymin), (xmax, ymax), detected_box_object.color, 2)
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def run_model_on_frame(self, model, frame: VideoFrame):
|
def run_model_on_frame(self, model, frame: VideoFrame):
|
||||||
return model(frame)[0]
|
model_outputs = model(frame)[0]
|
||||||
|
for data in model_outputs.boxes.data.tolist():
|
||||||
|
detected_obj = DetectedObject(*data)
|
||||||
|
if detected_obj.confidence > 0.5:
|
||||||
|
yield detected_obj
|
||||||
|
|
||||||
def persist_points(self):
|
def persist_points(self):
|
||||||
with open(self.output_json_path, "wt") as file_:
|
with open(self.output_json_path, "wt") as file_:
|
||||||
@@ -142,30 +129,28 @@ class DodgeballTracker:
|
|||||||
"""
|
"""
|
||||||
return YOLO("yolo11n.pt", verbose=True)
|
return YOLO("yolo11n.pt", verbose=True)
|
||||||
|
|
||||||
|
def update_tracker(self, tracker, detected_objs: list[DetectedObject], frame: VideoFrame) -> None:
|
||||||
|
important_detected_objs = [do for do in detected_objs if do.confidence > 0.5]
|
||||||
|
|
||||||
def main(self):
|
def main(self):
|
||||||
# define some constants
|
# define some constants
|
||||||
video_cap = cv2.VideoCapture(str(self.input_path))
|
video_cap = cv2.VideoCapture(str(self.input_path))
|
||||||
tracker = self._get_tracker()
|
tracker = self._get_tracker()
|
||||||
model = self._get_detection_model()
|
model = self._get_detection_model()
|
||||||
counter = Counter()
|
|
||||||
|
|
||||||
for i, (frame, ts) in enumerate(iter_frames(video_cap)):
|
for i, (frame, ts) in enumerate(iter_frames(video_cap)):
|
||||||
# run the YOLO model on the frame
|
# run the YOLO model on the frame
|
||||||
res = []
|
detected_objs = list(self.run_model_on_frame(model, frame))
|
||||||
detections = self.run_model_on_frame(model, frame)
|
self.video_writer.plot_boxes_to_frame(detected_objs, frame)
|
||||||
for data in detections.boxes.data.tolist():
|
# data_ = tracker.update(np.array([dobj.as_nparray() for dobj in detected_objs]), frame)
|
||||||
is_good_box = self.plot_box_to_frame(data, frame)
|
# if data_:
|
||||||
if is_good_box:
|
# print("dkld")
|
||||||
res.append(data)
|
# self.print_tracker_data(data_, i, ts)
|
||||||
|
|
||||||
data_ = tracker.update(np.array(res), frame)
|
|
||||||
self.print_tracker_data(data_, i, ts)
|
|
||||||
|
|
||||||
self.video_writer.write_frame(frame)
|
self.video_writer.write_frame(frame)
|
||||||
|
|
||||||
if self.video_writer.wait():
|
if self.video_writer.wait():
|
||||||
break
|
break
|
||||||
|
|
||||||
print(f"counter: {counter}")
|
|
||||||
video_cap.release()
|
video_cap.release()
|
||||||
self.persist_points()
|
self.persist_points()
|
||||||
@@ -1,7 +1,37 @@
|
|||||||
from typing import NewType
|
from typing import NewType
|
||||||
|
from dataclasses import dataclass
|
||||||
|
import numpy as np
|
||||||
|
|
||||||
type XCoord = float
|
XCoord = NewType('XCoord', float)
|
||||||
type YCoord = NewType('YCoord', float)
|
YCoord = NewType('YCoord', float)
|
||||||
type Point = tuple[XCoord, YCoord]
|
Confidence = NewType('Confidence', float)
|
||||||
type VideoFrame = NewType('Frame', float)
|
Category = NewType('Category', float)
|
||||||
type VideoTimestamp = float
|
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Point:
|
||||||
|
x: XCoord
|
||||||
|
y: YCoord
|
||||||
|
|
||||||
|
VideoFrame = NewType('Frame', float)
|
||||||
|
VideoTimestamp = float
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class DetectedObject:
|
||||||
|
xmin: XCoord
|
||||||
|
ymin: YCoord
|
||||||
|
xmax: XCoord
|
||||||
|
ymax: YCoord
|
||||||
|
confidence: Confidence
|
||||||
|
category: Category
|
||||||
|
|
||||||
|
def as_int_box(self) -> tuple[tuple[int, int], tuple[int, int]]:
|
||||||
|
return (int(self.xmin), int(self.ymin)), (int(self.xmax), int(self.ymax))
|
||||||
|
|
||||||
|
def as_nparray(self):
|
||||||
|
return np.array([self.xmin, self.ymax, self.xmax, self.ymax, self.confidence, self.category])
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
Color = tuple[int, int, int]
|
||||||
@@ -1,8 +1,25 @@
|
|||||||
import abc
|
import abc
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from dodgeballtracker.typing import VideoFrame
|
from dodgeballtracker.typing import VideoFrame, DetectedObject, Color
|
||||||
|
import matplotlib
|
||||||
import cv2
|
import cv2
|
||||||
|
|
||||||
|
|
||||||
|
class DetectedObjectBox:
|
||||||
|
def __init__(self, min_threshold: float, color: Color):
|
||||||
|
self.min_threshold = min_threshold
|
||||||
|
self.color = color
|
||||||
|
|
||||||
|
def __lt__(self, other):
|
||||||
|
return self.min_threshold < other.min_threshold
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_viridis(cls, min_threshold):
|
||||||
|
cm = matplotlib.colormaps['viridis']
|
||||||
|
color: Color = tuple((int(255 * cm(min_threshold)[i]) for i in range(3)))
|
||||||
|
return cls(min_threshold, color)
|
||||||
|
|
||||||
|
|
||||||
def create_video_writer(video_cap: cv2.VideoCapture, output_filename: Path) -> cv2.VideoWriter:
|
def create_video_writer(video_cap: cv2.VideoCapture, output_filename: Path) -> cv2.VideoWriter:
|
||||||
|
|
||||||
# grab the width, height, and fps of the frames in the video stream.
|
# grab the width, height, and fps of the frames in the video stream.
|
||||||
@@ -26,6 +43,10 @@ class VideoWriter(abc.ABC):
|
|||||||
def write_frame(self, frame: VideoFrame) -> None:
|
def write_frame(self, frame: VideoFrame) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@abc.abstractmethod
|
||||||
|
def plot_boxes_to_frame(self, data: list[DetectedObject], frame: VideoFrame) -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def close(self) -> None:
|
def close(self) -> None:
|
||||||
pass
|
pass
|
||||||
@@ -37,6 +58,9 @@ class VideoWriter(abc.ABC):
|
|||||||
|
|
||||||
class NoopVideoWriter(VideoWriter):
|
class NoopVideoWriter(VideoWriter):
|
||||||
|
|
||||||
|
def plot_boxes_to_frame(self, boxes: list[DetectedObject], frame: VideoFrame) -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
def write_frame(self, frame: VideoFrame) -> None:
|
def write_frame(self, frame: VideoFrame) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@@ -49,24 +73,39 @@ class NoopVideoWriter(VideoWriter):
|
|||||||
|
|
||||||
class MP4VVideoWriter(VideoWriter):
|
class MP4VVideoWriter(VideoWriter):
|
||||||
|
|
||||||
def __init__(self, video_writer: cv2.VideoWriter) -> None:
|
def __init__(self, video_writer: cv2.VideoWriter, thresholds: list[DetectedObjectBox]) -> None:
|
||||||
self._video_writer = video_writer
|
self._video_writer = video_writer
|
||||||
|
self._thresholds = thresholds
|
||||||
|
|
||||||
def write_frame(self, frame: VideoFrame) -> None:
|
def write_frame(self, frame: VideoFrame) -> None:
|
||||||
self._video_writer.write(frame)
|
self._video_writer.write(frame)
|
||||||
|
|
||||||
|
def plot_box_to_frame(self, data: DetectedObject, frame: VideoFrame) -> None:
|
||||||
|
confidence = float(data.confidence)
|
||||||
|
for threshold in self._thresholds:
|
||||||
|
if threshold.min_threshold < confidence:
|
||||||
|
# xmin, ymin, xmax, ymax = self._extract_box(data)
|
||||||
|
cv2.rectangle(frame, *data.as_int_box(), threshold.color, 2)
|
||||||
|
return
|
||||||
|
|
||||||
|
def plot_boxes_to_frame(self, boxes: list[DetectedObject], frame: VideoFrame) -> None:
|
||||||
|
for box in boxes:
|
||||||
|
self.plot_box_to_frame(box, frame)
|
||||||
|
|
||||||
|
|
||||||
def close(self) -> None:
|
def close(self) -> None:
|
||||||
self._video_writer.release()
|
self._video_writer.release()
|
||||||
cv2.destroyAllWindows()
|
cv2.destroyAllWindows()
|
||||||
|
|
||||||
def wait(self) -> bool:
|
def wait(self) -> bool:
|
||||||
return False #cv2.waitKey(1) == ord("q")
|
return False
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_frame(cls, input_: Path, output: Path) -> VideoWriter:
|
def from_frame(cls, input_: Path, output: Path, thresholds: list[float]) -> VideoWriter:
|
||||||
|
object_drawers = [DetectedObjectBox.from_viridis(t) for t in thresholds]
|
||||||
video_cap = cv2.VideoCapture(str(input_))
|
video_cap = cv2.VideoCapture(str(input_))
|
||||||
try:
|
try:
|
||||||
return cls(create_video_writer(video_cap, output))
|
return cls(create_video_writer(video_cap, output), object_drawers)
|
||||||
finally:
|
finally:
|
||||||
video_cap.release()
|
video_cap.release()
|
||||||
|
|
||||||
|
|||||||
@@ -27,6 +27,7 @@
|
|||||||
virtual-env = pkgs.python312.withPackages (ps: [
|
virtual-env = pkgs.python312.withPackages (ps: [
|
||||||
ps.matplotlib
|
ps.matplotlib
|
||||||
ps.torchWithCuda
|
ps.torchWithCuda
|
||||||
|
ps.jupyter
|
||||||
pythonPackages.opencv4
|
pythonPackages.opencv4
|
||||||
pythonPackages.boxmot
|
pythonPackages.boxmot
|
||||||
pythonPackages.ultralytics
|
pythonPackages.ultralytics
|
||||||
|
|||||||
Reference in new issue
Block a user