Repository: activeloopai/deeplake
Pull request: [AL-1796] Timestamps for video
## 🚀 🚀 Pull Request

### Checklist:

- [ ]  [My code follows the style guidelines of this project](https://www.notion.so/activeloop/Engineering-Guidelines-d6e502306d0e4133a8ca507516d1baab) and the [Contributing document](https://github.com/activeloopai/Hub/blob/release/2.0/CONTRIBUTING.md)
- [ ]  I have commented my code, particularly in hard-to-understand areas
- [ ]  I have kept the `coverage-rate` up
- [ ]  I have performed a self-review of my own code and resolved any problems
- [ ]  I have checked to ensure there aren't any other open [Pull Requests](https://github.com/activeloopai/Hub/pulls) for the same change
- [ ]  I have described and made corresponding changes to the relevant documentation
- [ ]  New and existing unit tests pass locally with my changes


### Changes
Added `tensor.timestamp` to get timestamps of video frames. Closes #1695
<!-- Describe your changes! Describe the scope of this PR's responsibilities. -->

Problem statement:
[FEATURE] Please add API support for timestamp of each frame retrieved from video
## 🚨🚨 Feature Request

- [ ] A new implementation (Improvement, Extension) 

### Is your feature request related to a problem?

Using Hub, we can get frames from a video. But currently there is no way to get timestamp of those frames. 

### If your feature will improve `HUB`

This feature will improve HUB because many video analytics applications depends on the timestamp of the frame - such as video search, key moment etc.


### Description of the possible solution

Anytime I get a frame from a video using `HUB`, I also want to get the timestamp of that frame

### An alternative solution to the problem can look like

An alternate solution can be to use `ffmpeg` and/or `ffprobe`

Hints from issue discussion:
Hi @nahidalam, can you tell us how you expect this output to look like?
Say a video is 10 minutes. The timestamps should be in seconds referring to anywhere in the video from 0 seconds to 10 minutes depending on the frame.

Commit message: fix

Patch to review:
diff --git a/hub/api/tests/test_video.py b/hub/api/tests/test_video.py
index 1e0068f1c7..eafe5cbf7e 100644
--- a/hub/api/tests/test_video.py
+++ b/hub/api/tests/test_video.py
@@ -1,9 +1,11 @@
 import os
+from re import L
 import sys
 import pytest
 
 import hub
 from hub.core.dataset import Dataset
+from hub.util.exceptions import DynamicTensorNumpyError
 
 import numpy as np
 
@@ -62,15 +64,16 @@ def test_video_slicing(local_ds: Dataset, video_paths):
     os.name == "nt" and sys.version_info < (3, 7), reason="requires python 3.7 or above"
 )
 @pytest.mark.parametrize(
-    "vstream_path",
-    ["gcs_vstream_path", "s3_vstream_path", "hub_cloud_vstream_path"],
+    ("vstream_path", "hub_token"),
+    [
+        ("gcs_vstream_path", "hub_cloud_dev_token"),
+        ("s3_vstream_path", "hub_cloud_dev_token"),
+        ("hub_cloud_vstream_path", "hub_cloud_dev_token"),
+    ],
     indirect=True,
 )
-def test_video_streaming(vstream_path, hub_cloud_dev_token):
-    if vstream_path.startswith("hub://"):
-        ds = hub.load(vstream_path, read_only=True, token=hub_cloud_dev_token)
-    else:
-        ds = hub.load(vstream_path, read_only=True)
+def test_video_streaming(vstream_path, hub_token):
+    ds = hub.load(vstream_path, read_only=True, token=hub_token)
 
     # no streaming, downloads chunk
     assert ds.mp4_videos[0].shape == (400, 360, 640, 3)
@@ -82,3 +85,70 @@ def test_video_streaming(vstream_path, hub_cloud_dev_token):
     assert ds.large_video[0, 13500].numpy().shape == (546, 1280, 3)
     # will use cached url
     assert ds.large_video[0, 18000].numpy().shape == (546, 1280, 3)
+
+
+@pytest.mark.skipif(
+    os.name == "nt" and sys.version_info < (3, 7), reason="requires python 3.7 or above"
+)
+@pytest.mark.parametrize(
+    ("vstream_path", "hub_token"),
+    [
+        ("gcs_vstream_path", "hub_cloud_dev_token"),
+        ("s3_vstream_path", "hub_cloud_dev_token"),
+        ("hub_cloud_vstream_path", "hub_cloud_dev_token"),
+    ],
+    indirect=True,
+)
+def test_video_timestamps(vstream_path, hub_token):
+    ds = hub.load(vstream_path, read_only=True, token=hub_token)
+
+    with pytest.raises(ValueError):
+        stamps = ds.mp4_videos[:2].timestamp
+
+    stamps = ds.large_video[0, 12000:1199:-100].timestamp
+
+    assert len(stamps) == 109
+
+    # timestamp is 50, 24 fps video, 50 * 24 = 1200th frame
+    assert stamps[-1] == 50
+
+    # cover stepping without seeking
+    stamps = ds.large_video[0, 1200:1300:2].timestamp
+
+    assert len(stamps) == 50
+    assert stamps[0] == 50
+
+
+def test_video_exception(local_ds):
+    with local_ds as ds:
+        ds.create_tensor("abc")
+        with pytest.raises(Exception):
+            stamps = ds.abc.timestamp
+
+
+@pytest.mark.skipif(
+    os.name == "nt" and sys.version_info < (3, 7), reason="requires python 3.7 or above"
+)
+def test_video_data(local_ds, video_paths):
+    with local_ds as ds:
+        ds.create_tensor("video", htype="video", sample_compression="mp4")
+        for _ in range(3):
+            ds.video.append(hub.read(video_paths["mp4"][0]))
+        ds.video.append(hub.read(video_paths["mp4"][1]))
+
+        assert ds.video[:2, 4, :5, :5].data().shape == (2, 5, 5, 3)
+
+        data = ds.video[:2, 10:20].data()
+        assert data["frames"].shape == (2, 10, 360, 640, 3)
+        assert data["timestamps"].shape == (2, 10)
+
+        with pytest.raises(DynamicTensorNumpyError):
+            ds.video[2:].data()
+
+        data = ds.video[2:].data(aslist=True)
+        assert len(data["frames"]) == 2
+        assert data["frames"][0].shape == ds.video[2].shape
+        assert data["frames"][1].shape == ds.video[3].shape
+        assert len(data["timestamps"]) == 2
+        assert data["timestamps"][0].shape == (ds.video[2].shape[0],)
+        assert data["timestamps"][1].shape == (ds.video[3].shape[0],)
diff --git a/hub/core/chunk/sample_compressed_chunk.py b/hub/core/chunk/sample_compressed_chunk.py
index 317d3ffde6..867a274cfb 100644
--- a/hub/core/chunk/sample_compressed_chunk.py
+++ b/hub/core/chunk/sample_compressed_chunk.py
@@ -64,8 +64,6 @@ def read_sample(  # type: ignore
         stream: bool = False,
         decompress: bool = True,
     ):
-        if not decompress and stream:
-            raise Exception("`decompress=False` is not valid when `stream=True`")
         if self.is_empty_tensor:
             raise EmptyTensorError
 
@@ -86,6 +84,8 @@ def read_sample(  # type: ignore
                     f"subfile,,start,{header_size + sb},end,{header_size + eb},,:"
                     + bytes(buffer[:-4]).decode("utf-8")
                 )
+                if not decompress:
+                    return buffer
             else:
                 buffer = buffer[sb:eb]
         if not decompress:
diff --git a/hub/core/chunk_engine.py b/hub/core/chunk_engine.py
index 1bff847cc6..2c94894d1c 100644
--- a/hub/core/chunk_engine.py
+++ b/hub/core/chunk_engine.py
@@ -1281,7 +1281,7 @@ def numpy(
             index, aslist, use_data_cache, fetch_chunks
         )
 
-    def get_video_sample(self, global_sample_index, index):
+    def get_video_sample(self, global_sample_index, index, decompress=True):
         enc = self.chunk_id_encoder
         chunk_ids = enc[global_sample_index]
         local_sample_index = enc.translate_index_relative_to_chunks(global_sample_index)
@@ -1291,7 +1291,10 @@ def get_video_sample(self, global_sample_index, index):
             local_sample_index,
             sub_index=sub_index,
             stream=stream,
-        )[tuple(entry.value for entry in index.values[2:])]
+            decompress=decompress,
+        )
+        if decompress:
+            return sample[tuple(entry.value for entry in index.values[2:])]
         return sample
 
     def get_basic_sample(self, global_sample_index, index, fetch_chunks=False):
diff --git a/hub/core/compression.py b/hub/core/compression.py
index 4d8c115f55..24477ad9d6 100644
--- a/hub/core/compression.py
+++ b/hub/core/compression.py
@@ -835,9 +835,9 @@ def _read_video_shape(
 
 def _decompress_video(
     file: Union[str, bytes],
-    start: Optional[int],
-    stop: Optional[int],
-    step: Optional[int],
+    start: int,
+    stop: int,
+    step: int,
     reverse: bool,
 ):
     container, vstream = _open_video(file)
@@ -896,6 +896,68 @@ def _decompress_video(
     return video
 
 
+def _read_timestamps(
+    file: Union[str, bytes],
+    start: int,
+    stop: int,
+    step: int,
+    reverse: bool,
+) -> np.ndarray:
+    container, vstream = _open_video(file)
+
+    nframes = math.ceil((stop - start) / step)
+
+    seek_target = _frame_to_stamp(start, vstream)
+    step_time = _frame_to_stamp(step, vstream)
+
+    stamps = []
+    if vstream.duration is None:
+        time_base = 1 / av.time_base
+    else:
+        time_base = vstream.time_base.numerator / vstream.time_base.denominator
+
+    gop_size = (
+        vstream.codec_context.gop_size
+    )  # gop size is distance (in frames) between 2 I-frames
+    if step > gop_size:
+        step_seeking = True
+    else:
+        step_seeking = False
+
+    seekable = True
+    try:
+        container.seek(seek_target, stream=vstream)
+    except av.error.PermissionError:
+        seekable = False
+        container, vstream = _open_video(file)  # try again but this time don't seek
+        warning(
+            "Cannot seek. Possibly a corrupted video file. Retrying with seeking disabled..."
+        )
+
+    i = 0
+    for packet in container.demux(video=0):
+        pts = packet.pts
+        if pts and pts >= seek_target:
+            stamps.append(pts * time_base)
+            i += 1
+            seek_target += step_time
+            if step_seeking and seekable:
+                container.seek(seek_target, stream=vstream)
+
+        if i == nframes:
+            break
+
+    # need to sort because when demuxing, frames are in order of dts (decoder timestamp)
+    # we need it in order of pts (presentation timestamp)
+    stamps.sort()
+    stamps_arr = np.zeros((nframes,), dtype=np.float32)
+    stamps_arr[: len(stamps)] = stamps
+
+    if reverse:
+        return stamps_arr[::-1]
+    return stamps_arr
+
+
 def _open_audio(file: Union[str, bytes, memoryview]):
     if not _PYAV_INSTALLED:
         raise ModuleNotFoundError(
diff --git a/hub/core/tensor.py b/hub/core/tensor.py
index f292ecc4df..b7c1011af5 100644
--- a/hub/core/tensor.py
+++ b/hub/core/tensor.py
@@ -12,6 +12,7 @@
 from hub.core.meta.tensor_meta import TensorMeta
 from hub.core.storage import StorageProvider
 from hub.core.chunk_engine import ChunkEngine
+from hub.core.compression import _read_timestamps
 from hub.core.tensor_link import get_link_transform
 from hub.api.info import load_info
 from hub.util.keys import (
@@ -51,6 +52,7 @@
 
 
 from hub.util.version_control import auto_checkout
+from hub.util.video import normalize_index
 
 from hub.compression import get_compression_type, VIDEO_COMPRESSION
 from hub.util.notebook import is_jupyter, video_html, is_colab
@@ -750,6 +752,27 @@ def data(self, aslist: bool = False) -> Any:
                 return list(self.numpy())
             else:
                 return list(map(list, self.numpy(aslist=True)))
+        elif htype == "video":
+            if len(self.index.values) > 2:
+                return self.numpy(aslist=aslist)
+            else:
+                data = {}
+                data["frames"] = self.numpy(aslist=aslist)
+                if self.index.values[0].subscriptable():
+                    start = self.index.values[0].value
+                    if isinstance(start, slice):
+                        start = start.start or 0
+                    data["timestamps"] = np.array(
+                        [
+                            self[i - start].timestamp
+                            for i in self.index.values[0].indices(self.num_samples)
+                        ]
+                    )
+                else:
+                    data["timestamps"] = self.timestamp
+                if aslist:
+                    data["timestamps"] = data["timestamps"].tolist()  # type: ignore
+                return data
         else:
             return self.numpy(aslist=aslist)
 
@@ -909,3 +932,22 @@ def play(self):
             )
         else:
             webbrowser.open(self._get_video_stream_url())
+
+    @property
+    def timestamp(self) -> np.ndarray:
+        if get_compression_type(self.meta.sample_compression) != VIDEO_COMPRESSION:
+            raise Exception("Only supported for video tensors.")
+        if self.index.values[0].subscriptable():
+            raise ValueError("Only supported for exactly 1 sample.")
+        index = self.index
+        sub_index = index.values[1].value if len(index.values) > 1 else None
+        global_sample_index = next(index.values[0].indices(self.num_samples))
+        sample = self.chunk_engine.get_video_sample(
+            global_sample_index, index, decompress=False
+        )
+
+        nframes = self.shape[0]
+        start, stop, step, reverse = normalize_index(sub_index, nframes)
+
+        stamps = _read_timestamps(sample, start, stop, step, reverse)
+        return stamps


Code review comments:
diff_hunk: @@ -909,3 +932,22 @@ def play(self):
             )
         else:
             webbrowser.open(self._get_video_stream_url())
+
+    @property
+    def timestamp(self) -> np.ndarray:
+        if get_compression_type(self.meta.sample_compression) != VIDEO_COMPRESSION:
+            raise Exception("Only supported for video tensors.")
+        if self.index.values[0].subscriptable():
path: hub/core/tensor.py
text: @user1:
Sequence case not covered.

diff_hunk: @@ -750,6 +752,27 @@ def data(self, aslist: bool = False) -> Any:
                 return list(self.numpy())
             else:
                 return list(map(list, self.numpy(aslist=True)))
+        elif htype == "video":
+            if len(self.index.values) > 2:
+                return self.numpy(aslist=aslist)
path: hub/core/tensor.py
text: @user1:
this seems like weird behaviour, .data returns numpy array in some cases and dict in other cases

@author:
If the tensor is indexed further than frames, it can't be called 'frames' and it doesnt make sense to return timestamps for it

@user1:
but a slice of a frame can also have a time stamp right? It's essentially just a part of the frame itself and timestamp is relevant for that too I guess

diff_hunk: @@ -750,6 +752,27 @@ def data(self, aslist: bool = False) -> Any:
                 return list(self.numpy())
             else:
                 return list(map(list, self.numpy(aslist=True)))
+        elif htype == "video":
+            if len(self.index.values) > 2:
path: hub/core/tensor.py
text: @user1:
Wherever self.index is accessed, make sure sequence htype is also taken into account. Add test for sequence htype + timestamp as well.