{"cells":[{"metadata":{},"cell_type":"markdown","source":"This kernel provides some basics to benchmark video readers. In this competition, most models get better accuracy with more frames. As inference kernel is time limited (9 hours), it's important to be able to load/decode frames as fast as possible. Maximum average time limit per video is **8.1 second**. CPU and GPU video decoding options are available but some should be used carefully at inference time as final test set is not known. If error cannot be catched it could lead to error during final scoring (i.e. no private scoring!).  \n  \nThe way frames are selected is also a way to speed up frames retrieval. One can get either regular spaced interval frames or frames spaced with stride interval. For instance if you want 3 frames for a video with 300 frames:\n1. With regular interval you would get [0, 100, 200].\n2. With stride=3 you would get [0, 3, 6] which is faster to load as reader stops quickly after frame 6. \n\nMain problem with stride option is that your model will be evaluated only on first part or the video. But it might be enough. We have to find the correct balance.   \n\nSome conclusions of this kernel:\n  \n\n* **OpenCV (CPU)**\n    * Stable.\n    * Errors can be catched. Can be used for inference.\n\n* **[Decord](https://github.com/dmlc/decord) (CPU)**\n    * Slower than OpenCV.\n\n* **[DALI](https://docs.nvidia.com/deeplearning/sdk/dali-developer-guide/docs/index.html) (GPU)**\n    * Faster than OpenCV.\n    * Decoded image is not the same as the CPU readers (average differs a little).\n    * Video with variable frame rate not supported.\n    * Implementation below crashes for more than 100 frames loaded/decoded. It also crashes with around 90 frames on private test set.\n    * Errors cannot be catched (kernel crash). Avoid it for inference (need more investigation). \n\n* **[Decord](https://github.com/dmlc/decord) (GPU)**\n    * Faster than all the other ones.\n    * Frame shift? First frame is not the same as other reader.\n    * Decoded image is not the same as the CPU readers (average differs a little)\n    * Memory leak that leads to random crash after cumulative batches.\n    * Errors cannot be catched (kernel crash). Avoid it for inference until fixed. \n\n  \nThere is another option I've tested: FileVideoStream from ImUtils that use threads with OpenCV but speed results are so similar to OpenCV that why I've dropped it.\n  \nFor the GPU readers, this kernel moves the images to \"CPU\" but for training or inference you should try to keep images in GPU to avoid unnecessary transfers from GPU/CPU/GPU.\n    \n\n# Feel free to comment and share your improvements."},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import sys, os, glob, gc\nimport timeit\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\nfrom tqdm import tqdm\nimport cv2\nprint('Python         : ' + sys.version.split('\\n')[0])\nprint('Numpy          : ' + np.__version__)\nprint('OpenCV         : ' + cv2.__version__)\npd.set_option('display.max_colwidth', 100)\npd.set_option('display.max_columns', 500)\npd.set_option('display.max_rows', 4000)\nimport seaborn as sns\nimport matplotlib.pyplot as plt\n%matplotlib inline\nsns.set()\nseed = 0\n\n#for dirname, _, filenames in os.walk('/kaggle/input'):\n#    for filename in filenames:\n#        print(os.path.join(dirname, filename))\n\n# Any results you write to the current directory are saved as output.","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"%%capture\n!pip install /kaggle/input/nvidia-dali-019-cuda-10/nvidia_dali-0.19.0-1119077-cp36-cp36m-manylinux1_x86_64.whl\n!cp /kaggle/input/decord/install.sh . && chmod  +x install.sh && ./install.sh","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"sys.path.insert(0,'/kaggle/working/reader/python')\nimport decord\nfrom nvidia.dali.pipeline import Pipeline\nfrom nvidia.dali import ops","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class CV2VideoReader(object):\n    \n    def __init__(self, path, num_frames, stride=None, verbose=False):\n        self.path = path\n        self.num_frames = num_frames\n        self.capture = None\n        self.frame_count = None\n        self.frame_counter = -1\n        self.idx = 0\n        self.stride = stride\n        self.verbose = verbose\n\n    def initialize(self):\n        ret = False\n        try:\n            self.capture = cv2.VideoCapture(self.path)\n            if self.capture.isOpened():\n                self.frame_count = int(self.capture.get(cv2.CAP_PROP_FRAME_COUNT))\n                if self.frame_count < self.num_frames:\n                    self.num_frames = self.frame_count            \n                if self.stride is None:\n                    # Frames at regular interval\n                    self.frame_idxs = np.linspace(0, self.frame_count, self.num_frames, endpoint=False, dtype=np.int)\n                else:\n                    # Frames with stride interval\n                    self.frame_idxs = [i*self.stride for i in range(self.num_frames) if i*self.stride < self.frame_count]\n                ret = True\n        except Exception as ex:\n            print(\"Init Error:\", ex)              \n        return ret\n\n    def __len__(self):\n        return len(self.frame_idxs)\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        decoded_frame = None\n        decoded_frame_idx = None\n        ret = None\n        # Grab next frame until the selected one\n        while ((self.frame_count is not None) and (self.idx < len(self.frame_idxs)) and (self.frame_counter < self.frame_count) and (self.frame_counter < self.frame_idxs[self.idx])):\n            self.frame_counter = self.frame_counter + 1\n            if self.verbose: print(\"grab\", self.frame_counter)\n            ret = self.capture.grab()\n            if not ret:\n                print(\"Error grabbing frame %d from %s\" % (self.frame_counter, self.path))               \n        # Retrieve the frame if possible\n        if ret:\n            if self.verbose: print(\"retrieve\", self.frame_counter, self.frame_idxs[self.idx])\n            ret, frame_tmp = self.capture.retrieve()\n            if ret and frame_tmp is not None:\n                decoded_frame = cv2.cvtColor(frame_tmp, cv2.COLOR_BGR2RGB)\n                decoded_frame_idx = self.frame_counter\n            else:\n                print(\"Error retrieving frame %d from %s\" % (self.frame_counter, self.path)) \n        # End of stream?\n        if (self.frame_counter >= self.frame_count) | (self.idx >= len(self.frame_idxs)):\n            self.release()\n            raise StopIteration\n        self.idx = self.idx + 1\n        return (decoded_frame, decoded_frame_idx)\n\n    def release(self):\n        if self.capture is not None:\n            self.capture.release()\n            self.capture = None","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# https://docs.nvidia.com/deeplearning/sdk/dali-developer-guide/docs/supported_ops.html#nvidia.dali.ops.VideoReader.__call__\nclass VideoPipe(Pipeline):\n\n    def __init__(self, batch_size, num_threads, device_id, data, shuffle, num_frames, stride):\n        super(VideoPipe, self).__init__(batch_size, num_threads, device_id, seed=seed)\n        self.input = ops.VideoReader(device=\"gpu\", filenames=data, sequence_length=num_frames, stride=stride,\n                                     shard_id=0, num_shards=1,\n                                     random_shuffle=shuffle, initial_fill=16)\n    def define_graph(self):\n        output = self.input(name=\"Reader\")\n        return output","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class DALIVideoReader(object):\n    \n    def __init__(self, path, num_frames, cpus=1, shuffle=False, stride=None, verbose=False):\n        self.path = path\n        self.num_frames = num_frames\n        self.frame_count = None\n        self.frame_counter = 0\n        self.local_idx = 0\n        self.idx = 0\n        self.cpus = cpus\n        self.shuffle = shuffle\n        self.stride = stride\n        self.verbose = verbose\n\n    def initialize(self):\n        ret = False\n        try:\n            # How to get total frames with DALI? Is is required to avoid crash when asking frames beyond the end.                \n            tmp = cv2.VideoCapture(self.path)\n            if tmp.isOpened():\n                self.frame_count = int(tmp.get(cv2.CAP_PROP_FRAME_COUNT))\n                if self.verbose == True: print(\"Frames count:\", self.frame_count)\n                tmp.release()\n            if self.frame_count is not None:\n                # Limit frames to what is available\n                if self.frame_count < self.num_frames:\n                    self.num_frames = self.frame_count          \n                if self.stride is None:\n                    # Frames at regular interval\n                    self.frame_idxs = np.linspace(0, self.frame_count, self.num_frames, endpoint=False, dtype=np.int)\n                    self.skip = self.frame_idxs[1] - self.frame_idxs[0] - 1\n                else:\n                    # Frames with stride interval\n                    self.frame_idxs = [i*self.stride for i in range(self.num_frames) if i*self.stride < self.frame_count]\n                    self.skip = self.stride\n                if self.verbose == True: print(\"Stride:\", self.skip)\n                self.capture = VideoPipe(batch_size=1, num_threads=self.cpus, device_id=0, data=[self.path], \n                                        shuffle=False, num_frames=len(self.frame_idxs), stride=self.skip)\n                self.capture.build()\n                if self.capture is not None:\n                    pipe_out = self.capture.run()\n                    self.batch = pipe_out[0].as_cpu().as_array()[0]\n                    if self.verbose == True: print(\"Sequence:\", len(self.batch))\n                    ret = True\n        except Exception as ex:\n            print(\"Init Error:\", ex)                    \n        return ret\n                \n    def __len__(self):\n        return len(self.batch)\n\n    def __iter__(self):\n        return self\n\n    # return RGB image, frame index tuple\n    def __next__(self):\n        if self.local_idx >= len(self.batch):\n            self.release()\n            raise StopIteration \n        decoded_frame = self.batch[self.local_idx]\n        decoded_frame_idx = self.frame_idxs[self.frame_counter]\n        self.local_idx = self.local_idx + 1\n        self.frame_counter = self.frame_counter + 1\n        return (decoded_frame, decoded_frame_idx)\n\n    def release(self):\n        if self.capture is not None:\n            del self.capture\n            self.capture = None\n            gc.collect()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class DecordVideoReader(object):\n    \n    def __init__(self, path, num_frames, ctx=decord.cpu(), stride=None, verbose=False):\n        self.path = path\n        self.num_frames = num_frames\n        self.frame_count = None\n        self.frame_counter = 0\n        self.batch_idx = 0\n        self.stride = stride\n        self.local_idx = 0\n        self.idx = 0\n        self.ctx = ctx\n        self.verbose = verbose\n\n    def initialize(self):\n        ret = False\n        try:\n            self.capture = decord.VideoReader(self.path, ctx=self.ctx)\n            if self.capture is not None:\n                self.frame_count = len(self.capture)\n                self.shape = self.capture[0].shape\n                if self.frame_count < self.num_frames:\n                    self.num_frames = self.frame_count\n                if self.stride is None:\n                    # Frames at regular interval\n                    self.frame_idxs = np.linspace(0, self.frame_count, self.num_frames, endpoint=False, dtype=np.int)\n                else:\n                    # Frames with stride interval\n                    self.frame_idxs = [i*self.stride for i in range(self.num_frames) if i*self.stride < self.frame_count]\n                self.frames_batch = np.array_split(self.frame_idxs, int(np.ceil(len(self.frame_idxs)/8.0)))\n                if self.verbose: print(\"Frames:\", self.num_frames)\n                if self.verbose: print(\"Batches:\", len(self.frames_batch))\n                self.batch = self.load_next_batch()\n                ret = True\n        except Exception as ex:\n            print(\"Init Error:\", ex)\n        return ret\n\n    def load_next_batch(self):\n        if self.verbose: print(\"Load batch:\", self.batch_idx)\n        batch_content = self.capture.get_batch(self.frames_batch[self.batch_idx]).asnumpy()\n        self.batch_idx = self.batch_idx + 1\n        return batch_content.copy()\n                \n    def __len__(self):\n        return len(self.frame_idxs)\n\n    def __iter__(self):\n        return self\n\n    # return RGB image, frame index tuple\n    def __next__(self):\n        # Next batch?\n        if self.local_idx >= len(self.batch):\n            # Next batch available?\n            if self.batch_idx < len(self.frames_batch):\n                self.batch = self.load_next_batch()\n                self.local_idx = 0\n            else:\n                self.release()\n                raise StopIteration \n        decoded_frame = self.batch[self.local_idx]\n        decoded_frame_idx = self.frame_idxs[self.frame_counter]\n        self.local_idx = self.local_idx + 1\n        self.frame_counter = self.frame_counter + 1\n        return (decoded_frame, decoded_frame_idx)\n\n    def release(self):\n        if self.capture is not None:\n            del self.capture\n            self.capture = None\n            gc.collect()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_video_reader(name, video, frames, stride, cpus=1):\n    if name == \"CV2-CPU\":\n        return CV2VideoReader(video, frames, stride=stride)\n    elif name == \"DALI-GPU\":\n        return DALIVideoReader(video, frames, cpus=cpus, stride=stride)    \n    elif name == \"Decord-CPU\":\n        return DecordVideoReader(video, frames, ctx=decord.cpu(), stride=stride)\n    elif name == \"Decord-GPU\":\n        return DecordVideoReader(video, frames, ctx=decord.gpu(), stride=stride)  ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"TEST_HOME = \"/kaggle/input/deepfake-detection-challenge/test_videos/\"\nfilenames = glob.glob(TEST_HOME + \"*.mp4\")\nbasenames = [(os.path.basename(filename), filename) for filename in filenames]\nsubmission_pd = pd.DataFrame(basenames, columns=[\"filename\", \"path\"])\nsubmission_pd = submission_pd.sort_values('filename')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Plot some results\nreaders = [\"CV2-CPU\", \"Decord-CPU\",\"Decord-GPU\", \"DALI-GPU\"]\n\nvideo = submission_pd[2:3][\"path\"].values[0]\nprint(video)\nframes = 3\n\nfor stride in [None, 3]:\n    img_idx = 0\n    columns, rows=(3, 1)\n    for reader_name in readers:\n        print(\"\\nReader:\", reader_name)\n        reader_ = get_video_reader(reader_name, video, frames, stride=stride)\n        if reader_.initialize() == True:\n            try:\n                loaded_frames = []\n                for frame, idx in reader_:\n                    loaded_frames.append(idx)\n                    col = img_idx % columns\n                    if col == 0: fig = plt.figure(figsize=(26, rows*5))\n                    ax = fig.add_subplot(rows, columns, col + 1)\n                    ax.imshow(frame)\n                    ax.axis(\"off\")\n                    ax.set_title(\"%s [%s, %s] Frame#%s %s - Avg: %.1f\" % (reader_name, len(reader_), stride, idx, frame.shape, np.mean(frame)))\n                    if (col == columns -1): plt.show()\n                    img_idx = img_idx + 1\n                print(\"query frames:  \", list(reader_.frame_idxs))\n                print(\"loaded frames: \", loaded_frames)\n            except Exception as ex:\n                print(\"Cannot load:\", ex)\n        reader_.release()\n        plt.show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"videos = submission_pd[0:10][\"path\"].values\nvideos","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# readers = [\"CV2-CPU\", \"Decord-CPU\",\"Decord-GPU\", \"DALI-GPU\"]\nreaders = [\"DALI-GPU\", \"CV2-CPU\", \"Decord-CPU\"]\nSTRIDES = [None, 1, 3, 5]\n\n# Decord-GPU random crashes, DALI crashes after 100.\nFRAMES = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120, 130, 150, 170, 190, 210, 240, 270, 300]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"results = []\nfor reader_name in readers:\n    print(\"Reader:\", reader_name, \"videos:\", len(videos))\n    for frames in FRAMES if \"DALI\" not in reader_name else [i for i in FRAMES if i <= 100]: # DALI crashes after 100\n        for stride in STRIDES:\n            start_time = timeit.default_timer()\n            loaded_frames = []\n            for video in videos:\n                reader_ = get_video_reader(reader_name, video, frames, stride=stride)\n                if reader_.initialize() == True:\n                    for frame, idx in reader_:\n                        loaded_frames.append(idx)\n                reader_.release()\n            elapsed = timeit.default_timer() - start_time\n            duration_per_video = elapsed / len(videos)\n            fps = (len(loaded_frames) / elapsed)\n            results.append((reader_name, frames, stride, duration_per_video, fps))\n            # print(\"[%s, %s, %s, %s] Elapsed %.4f sec. Average per video: %.4f sec.\" % (reader_name, frames, stride, len(loaded_frames), elapsed, duration_per_video), \"FPS: %.4f\" % (fps))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"results_pd = pd.DataFrame(results, columns=[\"reader\", \"frames\", \"stride\", \"duration_per_video\", \"fps\"])\nresults_pd.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"for reader_name in readers:\n    f, ax = plt.subplots(1, 2, figsize=(20, 5))\n    for stride in STRIDES:\n        d = results_pd[(results_pd[\"reader\"] == reader_name) & (results_pd[\"stride\"].isin([stride]))].plot(kind=\"line\", x=\"frames\", y=\"duration_per_video\", grid=True, ax=ax[0], label=\"Stride=%s\" % stride, title=\"%s - duration_per_video\" % reader_name)\n        d = results_pd[(results_pd[\"reader\"] == reader_name) & (results_pd[\"stride\"].isin([stride]))].plot(kind=\"line\", x=\"frames\", y=\"fps\", grid=True, ax=ax[1], label=\"Stride=%s\" % stride, title=\"%s - fps\" % reader_name)\n    plt.plot()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"!rm -r reader && rm install.sh","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"!cp /kaggle/input/deepfake-detection-challenge/sample_submission.csv ./submission.csv","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}