Skip to content

Instantly share code, notes, and snippets.

@mikelgg93
Created March 20, 2023 10:16
Show Gist options
  • Save mikelgg93/85923c4e2d9cd4917c42047faa146160 to your computer and use it in GitHub Desktop.
Save mikelgg93/85923c4e2d9cd4917c42047faa146160 to your computer and use it in GitHub Desktop.
Read number of frames

To install it:

pip install -r requirements.txt

To execute it:

python read_video.py path/to/video.mp4

Note: Depending on your python configuration you may need to call it pip3 and python3. Note2: The function will return fps, frame, pts, and "timestamps".

import av
import numpy as np
import logging
from rich.progress import Progress
logging.basicConfig(
format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO
)
def read_video_ts(video_path, audio=False, auto_thread_type=True):
"""
A function to read a video and return the fps,
the number of frames, the pts and timestamps.
:param video_path: the path to the video
"""
nframes = []
# Read the video
with av.open(video_path) as video_container:
if audio:
stream = video_container.streams.audio[0]
else:
stream = video_container.streams.video[0]
if auto_thread_type:
stream.thread_type = "AUTO"
fps = stream.average_rate # alt base_rate or guessed_rate
nframes = stream.frames
logging.info("Extracting pts...")
pts, dts, ts = (list() for i in range(3))
with Progress() as progress:
decode_task = progress.add_task("👓 Decoding...", total=nframes)
for packet in video_container.demux(stream):
for frame in packet.decode():
if frame is not None and frame.pts is not None:
pts.append(frame.pts)
dts.append(
frame.dts
) if frame.dts is not None else logging.info(
f"Decoding timestamp is missing at frame {len(pts)}"
)
ts.append(
(
frame.pts * frame.time_base
- stream.start_time * frame.time_base
)
* 1e9
)
progress.advance(decode_task)
progress.stop_task(decode_task)
pts, dts, ts = (
np.array(pts, dtype=np.uint64),
np.array(dts, dtype=np.uint64),
np.array(ts, dtype=np.uint64),
)
if not isMonotonicInc(pts):
logging.warning("Pts are not monotonic increasing!.")
if np.array_equal(pts, dts):
logging.info("Pts and dts are equal, using pts")
idc = pts.argsort()
pts = pts[idc]
ts = ts[idc]
if nframes != len(pts):
nframes = len(pts)
else:
logging.info(f"This video has {nframes} frames")
return fps, nframes, pts, ts
def isMonotonicInc(a2check):
return all(a2check[i] <= a2check[i + 1] for i in range(len(a2check) - 1))
if __name__ == "__main__":
import sys
read_video_ts(sys.argv[1])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment