Upload 4 files

data_loader.py

@@ -0,0 +1,155 @@
+""" "This module contains a video loader."""
+
+import logging
+import os
+import sys
+from typing import List, Tuple, Union
+
+import numpy as np
+from torch import Tensor
+from torch.utils import data
+from torchvision.datasets.video_utils import VideoClips
+
+
+class VideoIter(data.Dataset):
+    """This class implements a loader for videos."""
+
+    def __init__(
+        self,
+        clip_length,
+        frame_stride,
+        dataset_path=None,
+        video_transform=None,
+        return_label=False,
+    ) -> None:
+        super().__init__()
+        # video clip properties
+        self.frames_stride = frame_stride
+        self.total_clip_length_in_frames = clip_length * frame_stride
+        self.video_transform = video_transform
+
+        # IO
+        self.dataset_path = dataset_path
+        self.video_list = self._get_video_list(dataset_path=self.dataset_path)
+        self.return_label = return_label
+
+        # data loading
+        self.video_clips = VideoClips(
+            video_paths=self.video_list,
+            clip_length_in_frames=self.total_clip_length_in_frames,
+            frames_between_clips=self.total_clip_length_in_frames,
+        )
+
+    @property
+    def video_count(self) -> int:
+        """Retrieve the number of the videos in the dataset."""
+        return len(self.video_list)
+
+    def getitem_from_raw_video(
+        self, idx: int
+    ) -> Union[Tuple[Tensor, int, str, str], Tuple[Tensor, int, int, str, str]]:
+        """Fetch a sample from the dataset.
+
+        Args:
+            idx (int): Index of the sample the retrieve.
+
+        Returns:
+            Tuple[Tensor, int, str, str]: Video clip, clip idx in the video, directory name, and file
+            Tuple[Tensor, int, int, str, str]: Video clip, label, clip idx in the video, directory name, and file
+        """
+        video, _, _, _ = self.video_clips.get_clip(idx)
+        video_idx, clip_idx = self.video_clips.get_clip_location(idx)
+        video_path = self.video_clips.video_paths[video_idx]
+        in_clip_frames = list(
+            range(0, self.total_clip_length_in_frames, self.frames_stride)
+        )
+        video = video[in_clip_frames]
+        if self.video_transform is not None:
+            video = self.video_transform(video)
+
+        dir, file = video_path.split(os.sep)[-2:]
+        file = file.split(".")[0]
+
+        if self.return_label:
+            label = 0 if "Normal" in video_path else 1
+            return video, label, clip_idx, dir, file
+
+        return video, clip_idx, dir, file
+
+    def __len__(self) -> int:
+        return len(self.video_clips)
+
+    def __getitem__(self, index: int):
+        succ = False
+        while not succ:
+            try:
+                batch = self.getitem_from_raw_video(index)
+                succ = True
+            except Exception as e:
+                index = np.random.choice(range(0, self.__len__()))
+                trace_back = sys.exc_info()[2]
+                if trace_back is not None:
+                    line = str(trace_back.tb_lineno)
+                else:
+                    line = "no-line"
+                # pylint: disable=line-too-long
+                logging.warning(
+                    f"VideoIter:: ERROR (line number {line}) !! (Force using another index:\n{index})\n{e}"
+                )
+
+        return batch
+
+    def _get_video_list(self, dataset_path: str) -> List[str]:
+        """Fetche all videos in a directory and sub-directories.
+
+        Args:
+            dataset_path (str): A string that represents the directory of the dataset.
+
+        Raises:
+            FileNotFoundError: The directory could not be found in the provided path.
+
+        Returns:
+            List[str]
+        """
+        if not os.path.exists(dataset_path):
+            raise FileNotFoundError(f"VideoIter:: failed to locate: `{dataset_path}'")
+
+        vid_list = []
+        for path, _, files in os.walk(dataset_path):
+            for name in files:
+                if "mp4" not in name:
+                    continue
+                vid_list.append(os.path.join(path, name))
+
+        logging.info(f"Found {len(vid_list)} video files in {dataset_path}")
+        return vid_list
+
+
+class SingleVideoIter(VideoIter):
+    """Loader for a single video."""
+
+    def __init__(
+        self,
+        clip_length,
+        frame_stride,
+        video_path,
+        video_transform=None,
+        return_label=False,
+    ) -> None:
+        super().__init__(
+            clip_length, frame_stride, video_path, video_transform, return_label
+        )
+
+    def _get_video_list(self, dataset_path: str) -> List[str]:
+        return [dataset_path]
+
+    def __getitem__(self, idx: int) -> Tensor:
+        video, _, _, _ = self.video_clips.get_clip(idx)
+        in_clip_frames = list(
+            range(0, self.total_clip_length_in_frames, self.frames_stride)
+        )
+        video = video[in_clip_frames]
+        if self.video_transform is not None:
+            video = self.video_transform(video)
+
+        return video

feature_extractor.py

@@ -0,0 +1,314 @@
+"""This module contains a training procedure for video feature extraction."""
+
+import argparse
+import logging
+import os
+from os import mkdir, path
+from typing import Dict, List, Optional, Tuple, Union
+
+import numpy as np
+import torch
+from torch import Tensor
+from torch.backends import cudnn
+from torch.utils.data import DataLoader
+
+from data_loader import VideoIter
+from network.TorchUtils import get_torch_device
+from utils.load_model import load_feature_extractor
+from utils.utils import build_transforms, register_logger
+
+
+def get_args() -> argparse.Namespace:
+    """Reads command line args and returns the parser object the represent the
+    specified arguments."""
+
+    parser = argparse.ArgumentParser(description="Video Feature Extraction Parser")
+
+    # io
+    parser.add_argument(
+        "--dataset_path",
+        default="../kinetics2/kinetics2/AnomalyDetection",
+        help="path to dataset",
+    )
+    parser.add_argument(
+        "--clip-length",
+        type=int,
+        default=16,
+        help="define the length of each input sample.",
+    )
+    parser.add_argument(
+        "--num_workers",
+        type=int,
+        default=8,
+        help="define the number of workers used for loading the videos",
+    )
+    parser.add_argument(
+        "--frame-interval",
+        type=int,
+        default=1,
+        help="define the sampling interval between frames.",
+    )
+    parser.add_argument(
+        "--log-every",
+        type=int,
+        default=50,
+        help="log the writing of clips every n steps.",
+    )
+    parser.add_argument("--log-file", type=str, help="set logging file.")
+    parser.add_argument(
+        "--save_dir",
+        type=str,
+        default="features",
+        help="set output directory for the features.",
+    )
+
+    # optimization
+    parser.add_argument("--batch-size", type=int, default=8, help="batch size")
+
+    # model
+    parser.add_argument(
+        "--model_type",
+        type=str,
+        required=True,
+        help="type of feature extractor",
+        choices=["c3d", "i3d", "mfnet", "3dResNet"],
+    )
+    parser.add_argument(
+        "--pretrained_3d", type=str, help="load default 3D pretrained model."
+    )
+
+    return parser.parse_args()
+
+
+def to_segments(
+    data: Union[Tensor, np.ndarray], n_segments: int = 32
+) -> List[np.ndarray]:
+    """These code is taken from:
+
+        # https://github.com/rajanjitenpatel/C3D_feature_extraction/blob/b5894fa06d43aa62b3b64e85b07feb0853e7011a/extract_C3D_feature.py#L805
+
+    Args:
+        data (Union[Tensor, np.ndarray]): List of features of a certain video
+        n_segments (int, optional): Number of segments
+
+    Returns:
+        List[np.ndarray]: List of `num` segments
+    """
+    data = np.array(data)
+    Segments_Features = []
+    thirty2_shots = np.round(np.linspace(0, len(data) - 1, num=n_segments + 1)).astype(
+        int
+    )
+    for ss, ee in zip(thirty2_shots[:-1], thirty2_shots[1:]):
+        if ss == ee:
+            temp_vect = data[min(ss, data.shape[0] - 1), :]
+        else:
+            temp_vect = data[ss:ee, :].mean(axis=0)
+
+        temp_vect = temp_vect / np.linalg.norm(temp_vect)
+
+        if np.linalg.norm(temp_vect) != 0:
+            Segments_Features.append(temp_vect.tolist())
+
+    return Segments_Features
+
+
+class FeaturesWriter:
+    """Accumulates and saves extracted features."""
+
+    def __init__(self, num_videos: int, chunk_size: int = 16) -> None:
+        self.path = ""
+        self.dir = ""
+        self.data = {}
+        self.chunk_size = chunk_size
+        self.num_videos = num_videos
+        self.dump_count = 0
+
+    def _init_video(self, video_name: str, dir: str) -> None:
+        """Initialize the state of the writer for a new video.
+
+        Args:
+            video_name (str): Name of the video to initialize for.
+            dir (str): Directory where the video is stored.
+        """
+        self.path = path.join(dir, f"{video_name}.txt")
+        self.dir = dir
+        self.data = {}
+
+    def has_video(self) -> bool:
+        """Checks whether the writer is initialized with a video.
+
+        Returns:
+            bool
+        """
+        return self.data is not None
+
+    def dump(self, dir: str) -> None:
+        """Saves the accumulated features to disk.
+
+        The features will be segmented and normalized.
+        """
+        logging.info(f"{self.dump_count} / {self.num_videos}:	Dumping {self.path}")
+        self.dump_count += 1
+        self.dir = dir
+        if not path.exists(self.dir):
+            os.makedirs(self.dir, exist_ok=True)
+        #####################################################
+        # Check if data is empty before attempting to process it
+        if len(self.data) == 0:
+            logging.warning("No data to dump, skipping.")
+            return  # If data is empty, skip this dump.
+        #####################################################
+        features = to_segments(np.array([self.data[key] for key in sorted(self.data)]))
+        with open(self.path, "w") as fp:
+            for d in features:
+                d_str = [str(x) for x in d]
+                fp.write(" ".join(d_str) + "\n")
+
+    def _is_new_video(self, video_name: str, dir: str) -> bool:
+        """Checks whether the given video is new or the writer is already
+        initialized with it.
+
+        Args:
+            video_name (str): Name of the possibly new video.
+            dir (str): Directory where the video is stored.
+
+        Returns:
+            bool
+        """
+        new_path = path.join(dir, f"{video_name}.txt")
+        if self.path != new_path and self.path is not None:
+            return True
+
+        return False
+
+    def store(self, feature: Union[Tensor, np.ndarray], idx: int) -> None:
+        """Accumulate features.
+
+        Args:
+            feature (Union[Tensor, np.ndarray]): Features to be accumulated.
+            idx (int): Indices of features in the video.
+        """
+        self.data[idx] = list(feature)
+
+    def write(
+        self, feature: Union[Tensor, np.ndarray], video_name: str, idx: int, dir: str
+    ) -> None:
+        if not self.has_video():
+            self._init_video(video_name, dir)
+        if self._is_new_video(video_name, dir):
+            self.dump(dir)
+            self._init_video(video_name, dir)
+
+        self.store(feature, idx)
+
+
+def read_features(file_path, cache: Optional[Dict[str, Tensor]] = None) -> Tensor:
+    """Reads features from file.
+
+    Args:
+        file_path (_type_): Path to a text file containing features. Each line should contain a feature
+            for a single video segment.
+        cache (Dict, optional): A cache that stores features that were already loaded.
+            If `None`, caching is disabled.Defaults to None.
+
+    Raises:
+        FileNotFoundError: The provided path does not exist.
+
+    Returns:
+        Tensor
+    """
+    if cache is not None and file_path in cache:
+        return cache[file_path]
+
+    if not path.exists(file_path):
+        raise FileNotFoundError(f"Feature doesn't exist: `{file_path}`")
+
+    features = None
+    with open(file_path) as fp:
+        data = fp.read().splitlines(keepends=False)
+        features = torch.tensor(
+            np.stack([line.split(" ") for line in data]).astype(np.float32)
+        )
+
+    if cache is not None:
+        cache[file_path] = features
+    return features
+
+
+def get_features_loader(
+    dataset_path: str,
+    clip_length: int,
+    frame_interval: int,
+    batch_size: int,
+    num_workers: int,
+    mode: str,
+) -> Tuple[VideoIter, DataLoader]:
+    data_loader = VideoIter(
+        dataset_path=dataset_path,
+        clip_length=clip_length,
+        frame_stride=frame_interval,
+        video_transform=build_transforms(mode),
+        return_label=False,
+    )
+
+    data_iter = torch.utils.data.DataLoader(
+        data_loader,
+        batch_size=batch_size,
+        shuffle=False,
+        num_workers=num_workers,
+        pin_memory=True,
+    )
+
+    return data_loader, data_iter
+
+
+if __name__ == "__main__":
+    device = get_torch_device()
+
+    args = get_args()
+    register_logger(log_file=args.log_file)
+
+    cudnn.benchmark = True
+
+    data_loader, data_iter = get_features_loader(
+        args.dataset_path,
+        args.clip_length,
+        args.frame_interval,
+        args.batch_size,
+        args.num_workers,
+        args.model_type,
+    )
+
+    network = load_feature_extractor(args.model_type, args.pretrained_3d, device).eval()
+
+    if not path.exists(args.save_dir):
+        mkdir(args.save_dir)
+
+    features_writer = FeaturesWriter(num_videos=data_loader.video_count)
+    loop_i = 0
+    global_dir: str = "none"
+    with torch.no_grad():
+        for data, clip_idxs, dirs, vid_names in data_iter:
+            outputs = network(data.to(device)).detach().cpu().numpy()
+
+            for i, (_dir, vid_name, clip_idx) in enumerate(
+                zip(dirs, vid_names, clip_idxs)
+            ):
+                if loop_i == 0:
+                    # pylint: disable=line-too-long
+                    logging.info(
+                        f"Video {features_writer.dump_count} / {features_writer.num_videos} : Writing clip {clip_idx} of video {vid_name}"
+                    )
+                loop_i += 1
+                loop_i %= args.log_every
+                _dir = path.join(args.save_dir, _dir)
+                global_dir = _dir
+                features_writer.write(
+                    feature=outputs[i],
+                    video_name=vid_name,
+                    idx=clip_idx,
+                    dir=_dir,
+                )
+
+    features_writer.dump(global_dir)

features_loader.py

@@ -0,0 +1,201 @@
+""" "This module contains a video feature loader."""
+
+import logging
+import os
+from typing import List, Tuple
+
+import numpy as np
+import torch
+from torch import Tensor
+from torch.utils import data
+
+from feature_extractor import read_features
+
+
+class FeaturesLoader:
+    """Loads video features that are stored as text files."""
+
+    def __init__(
+        self,
+        features_path: str,
+        annotation_path: str,
+        bucket_size: int = 30,
+        iterations: int = 20000,
+    ) -> None:
+        """
+        Args:
+            features_path: Path to the directory that contains the features in text files
+            annotation_path: Path to the annotation file
+            bucket_size: Size of each bucket
+            iterations: How many iterations the loader should perform
+        """
+
+        super().__init__()
+        self._features_path = features_path
+        self._bucket_size = bucket_size
+
+        # load video list
+        (
+            self.features_list_normal,
+            self.features_list_anomaly,
+        ) = FeaturesLoader._get_features_list(
+            features_path=self._features_path, annotation_path=annotation_path
+        )
+
+        self._iterations = iterations
+        self._features_cache = {}
+        self._i = 0
+
+    def __len__(self) -> int:
+        return self._iterations
+
+    def __getitem__(self, index: int) -> Tuple[Tensor, Tensor]:
+        if self._i == len(self):
+            self._i = 0
+            raise StopIteration
+
+        succ = False
+        while not succ:
+            try:
+                feature, label = self.get_features()
+                succ = True
+            except Exception as e:
+                index = np.random.choice(range(0, self.__len__()))
+                logging.warning(
+                    f"VideoIter:: ERROR!! (Force using another index:\n{index})\n{e}"
+                )
+
+        self._i += 1
+        return feature, label
+
+    def get_features(self) -> Tuple[Tensor, Tensor]:
+        """Fetches a bucket sample from the dataset."""
+        normal_paths = np.random.choice(
+            self.features_list_normal, size=self._bucket_size
+        )
+        abnormal_paths = np.random.choice(
+            self.features_list_anomaly, size=self._bucket_size
+        )
+        all_paths = np.concatenate([normal_paths, abnormal_paths])
+        features = torch.stack(
+            [
+                read_features(f"{feature_subpath}.txt", self._features_cache)
+                for feature_subpath in all_paths
+            ]
+        )
+        return (
+            features,
+            torch.cat([torch.zeros(self._bucket_size), torch.ones(self._bucket_size)]),
+        )
+
+    @staticmethod
+    def _get_features_list(
+        features_path: str, annotation_path: str
+    ) -> Tuple[List[str], List[str]]:
+        """Retrieves the paths of all feature files contained within the
+        annotation file.
+
+        Args:
+            features_path: Path to the directory that contains feature text files
+            annotation_path: Path to the annotation file
+
+        Returns:
+            Tuple[List[str], List[str]]: Two list that contain the corresponding paths of normal and abnormal
+                feature files.
+        """
+        assert os.path.exists(features_path)
+        features_list_normal = []
+        features_list_anomaly = []
+        with open(annotation_path) as f:
+            lines = f.read().splitlines(keepends=False)
+            for line in lines:
+                items = line.split()
+                file = items[0].split(".")[0]
+                file = file.replace("/", os.sep)
+                feature_path = os.path.join(features_path, file)
+                if "Normal" in feature_path:
+                    features_list_normal.append(feature_path)
+                else:
+                    features_list_anomaly.append(feature_path)
+
+        return features_list_normal, features_list_anomaly
+
+    @property
+    def get_feature_dim(self) -> int:
+        return self[0][0].shape[-1]
+
+
+class FeaturesLoaderVal(data.Dataset):
+    """Loader for video features for validation phase."""
+
+    def __init__(self, features_path, annotation_path):
+        super().__init__()
+        self.features_path = features_path
+        # load video list
+        self.state = "Normal"
+        self.features_list = FeaturesLoaderVal._get_features_list(
+            features_path=features_path, annotation_path=annotation_path
+        )
+
+    def __len__(self):
+        return len(self.features_list)
+
+    def __getitem__(self, index: int):
+        succ = False
+        while not succ:
+            try:
+                data = self.get_feature(index)
+                succ = True
+            except Exception as e:
+                logging.warning(
+                    f"VideoIter:: ERROR!! (Force using another index:\n{index})\n{e}"
+                )
+
+        return data
+
+    def get_feature(self, index: int):
+        """Fetch feature that matches given index in the dataset.
+
+        Args:
+            index (int): Index of the feature to fetch.
+
+        Returns:
+            _type_: _description_
+        """
+        feature_subpath, start_end_couples, length = self.features_list[index]
+        features = read_features(f"{feature_subpath}.txt")
+        return features, start_end_couples, length
+
+    @staticmethod
+    def _get_features_list(
+        features_path: str, annotation_path: str
+    ) -> List[Tuple[str, Tensor, int]]:
+        """Retrieves the paths of all feature files contained within the
+        annotation file.
+
+        Args:
+            features_path: Path to the directory that contains feature text files
+            annotation_path: Path to the annotation file
+
+        Returns:
+            List[Tuple[str, Tensor, int]]: A list of tuples that describe each video and the temporal annotations
+                of anomalies in the videos
+        """
+        assert os.path.exists(features_path)
+        features_list = []
+        with open(annotation_path) as f:
+            lines = f.read().splitlines(keepends=False)
+            for line in lines:
+                items = line.split()
+                anomalies_frames = [int(x) for x in items[3:]]
+                start_end_couples = torch.tensor(
+                    [anomalies_frames[:2], anomalies_frames[2:]]
+                )
+                file = items[0].split(".")[0]
+                file = file.replace("/", os.sep)
+                feature_path = os.path.join(features_path, file)
+                length = int(items[1])
+
+                features_list.append((feature_path, start_end_couples, length))
+
+        return features_list

generate_ROC.py

@@ -0,0 +1,113 @@
+""" "This module contains an evaluation procedure for video anomaly
+detection."""
+
+import argparse
+import os
+from os import path
+
+import matplotlib.pyplot as plt
+import numpy as np
+import torch
+from sklearn.metrics import auc, roc_curve
+from torch.backends import cudnn
+from tqdm import tqdm
+
+from features_loader import FeaturesLoaderVal
+from network.TorchUtils import TorchModel
+
+
+def get_args() -> argparse.Namespace:
+    """Reads command line args and returns the parser object the represent the
+    specified arguments."""
+    parser = argparse.ArgumentParser(
+        description="Video Anomaly Detection Evaluation Parser"
+    )
+    parser.add_argument(
+        "--features_path",
+        type=str,
+        default="../anomaly_features",
+        required=True,
+        help="path to features",
+    )
+    parser.add_argument(
+        "--annotation_path", default="Test_Annotation.txt", help="path to annotations"
+    )
+    parser.add_argument(
+        "--model_path", type=str, required=True, help="Path to the anomaly detector."
+    )
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    args = get_args()
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+    data_loader = FeaturesLoaderVal(
+        features_path=args.features_path,
+        annotation_path=args.annotation_path,
+    )
+
+    data_iter = torch.utils.data.DataLoader(
+        data_loader,
+        batch_size=1,
+        shuffle=False,
+        num_workers=0,  # 4, # change this part accordingly
+        pin_memory=True,
+    )
+
+    model = TorchModel.load_model(args.model_path).to(device).eval()
+
+    # enable cudnn tune
+    cudnn.benchmark = True
+
+    # pylint: disable=not-callable
+    y_trues = np.array([])
+    y_preds = np.array([])
+
+    with torch.no_grad():
+        for features, start_end_couples, lengths in tqdm(data_iter):
+            # features is a batch where each item is a tensor of 32 4096D features
+            features = features.to(device)
+            outputs = model(features).squeeze(-1)  # (batch_size, 32)
+            for vid_len, couples, output in zip(
+                lengths, start_end_couples, outputs.cpu().numpy()
+            ):
+                y_true = np.zeros(vid_len)
+                y_pred = np.zeros(vid_len)
+
+                segments_len = vid_len // 32
+                for couple in couples:
+                    if couple[0] != -1:
+                        y_true[couple[0] : couple[1]] = 1
+
+                for i in range(32):
+                    segment_start_frame = i * segments_len
+                    segment_end_frame = (i + 1) * segments_len
+                    y_pred[segment_start_frame:segment_end_frame] = output[i]
+
+                if y_trues is None:
+                    y_trues = y_true
+                    y_preds = y_pred
+                else:
+                    y_trues = np.concatenate([y_trues, y_true])
+                    y_preds = np.concatenate([y_preds, y_pred])
+
+    fpr, tpr, thresholds = roc_curve(y_true=y_trues, y_score=y_preds, pos_label=1)
+
+    plt.figure()
+    lw = 2
+    roc_auc = auc(fpr, tpr)
+    plt.plot(
+        fpr, tpr, color="darkorange", lw=lw, label=f"ROC curve (area = {roc_auc:0.2f})"
+    )
+    plt.plot([0, 1], [0, 1], color="navy", lw=lw, linestyle="--")
+    plt.xlim([0.0, 1.0])
+    plt.ylim([0.0, 1.05])
+    plt.xlabel("False Positive Rate")
+    plt.ylabel("True Positive Rate")
+    plt.legend(loc="lower right")
+
+    os.makedirs("graphs", exist_ok=True)
+    plt.savefig(path.join("graphs", "roc_auc.png"))
+    plt.close()
+    print(f"ROC curve (area = {roc_auc:0.2f})")