Add Packet.rescale_ts

Author: WyattBlue

CHANGELOG.rst

@@ -33,6 +33,7 @@ v18.1.0 (Unreleased)
 
 Features:
 
+- Add ``Packet.rescale_ts()`` to rescale packet PTS, DTS, and duration to a new ``AVRational`` time base by :gh-user:`WyattBlue`.
 - Support reusing the thread's current CUDA context via a ``current_ctx`` flag on ``CudaContext`` and ``VideoFrame.from_dlpack``, for interop with libraries like PyTorch that initialize CUDA first by :gh-user:`Yozer` (:pr:`2339`).
 - ``VideoFrame.from_dlpack`` no longer requires restating ``primary_ctx``/``current_ctx`` when passing an explicit ``cuda_context``; the flags are only validated when explicitly given by :gh-user:`WyattBlue`.
 

av/packet.py

@@ -11,6 +11,8 @@
 from cython.cimports.libc.stdint import uint8_t
 from cython.cimports.libc.string import memcpy
 
+from av.rational import AVRational
+
 # Check https://github.com/FFmpeg/FFmpeg/blob/master/libavcodec/packet.h#L41
 # for new additions in the future ffmpeg releases
 # Note: the order must follow that of the AVPacketSideDataType enum def
@@ -289,6 +291,25 @@ def _rebase_time(self, dst: lib.AVRational):
         lib.av_packet_rescale_ts(self.ptr, self.ptr.time_base, dst)
         self.ptr.time_base = dst
 
+    def rescale_ts(self, time_base):
+        """Rescale the packet timestamps to a new time base.
+
+        This rescales :attr:`pts`, :attr:`dts`, and :attr:`duration`, then updates
+        :attr:`time_base`. If the current time base is unset, the timestamp values
+        are unchanged and the new time base is assigned.
+
+        :meth:`~av.container.OutputContainer.mux` already performs this operation
+        automatically when necessary.
+
+        Wraps :ffmpeg:`av_packet_rescale_ts`.
+        """
+        if not isinstance(time_base, AVRational):
+            raise TypeError("time_base must be an AVRational")
+
+        dst: lib.AVRational
+        to_avrational(time_base, cython.address(dst))
+        self._rebase_time(dst)
+
     def decode(self):
         """
         Send the packet's data to the decoder and return a list of

av/packet.pyi

@@ -4,6 +4,7 @@ from typing import Generic, Literal, TypeVar, overload
 
 from av.audio.frame import AudioFrame
 from av.audio.stream import AudioStream
+from av.rational import AVRational
 from av.stream import Stream
 from av.subtitles.stream import SubtitleStream
 from av.subtitles.subtitle import AssSubtitle, BitmapSubtitle
@@ -92,6 +93,7 @@ class Packet(Buffer, Generic[StreamT]):
     is_disposable: bool
 
     def __init__(self: Packet[Stream], input: int | bytes | None = None) -> None: ...
+    def rescale_ts(self, time_base: AVRational) -> None: ...
 
     # Overloads that return the same type as the stream's decode method
     @overload

tests/test_packet.py

@@ -6,6 +6,7 @@
 from unittest import SkipTest
 
 import numpy
+import pytest
 
 import av
 
@@ -63,6 +64,45 @@ def test_data_packet_bytes(self):
 
 
 class TestProperties:
+    def test_rescale_ts(self) -> None:
+        packet = av.Packet()
+        packet.time_base = fractions.Fraction(1, 1000)
+        packet.pts = 1000
+        packet.dts = 900
+        packet.duration = 40
+
+        packet.rescale_ts(av.AVRational(1, 100))
+
+        assert packet.time_base == fractions.Fraction(1, 100)
+        assert packet.pts == 100
+        assert packet.dts == 90
+        assert packet.duration == 4
+
+    def test_rescale_ts_without_source_time_base(self) -> None:
+        packet = av.Packet()
+        packet.pts = 1000
+        packet.dts = 900
+        packet.duration = 40
+
+        packet.rescale_ts(av.AVRational(1, 100))
+
+        assert packet.time_base == fractions.Fraction(1, 100)
+        assert packet.pts == 1000
+        assert packet.dts == 900
+        assert packet.duration == 40
+
+    def test_rescale_ts_rejects_zero_time_base(self) -> None:
+        packet = av.Packet()
+
+        with pytest.raises(ValueError, match="Cannot rebase to zero time"):
+            packet.rescale_ts(av.AVRational(0, 1))
+
+    def test_rescale_ts_requires_avrational(self) -> None:
+        packet = av.Packet()
+
+        with pytest.raises(TypeError, match="time_base must be an AVRational"):
+            packet.rescale_ts(fractions.Fraction(1, 100))  # type: ignore[arg-type]
+
     def test_is_keyframe(self) -> None:
         with av.open(fate_suite("h264/interlaced_crop.mp4")) as container:
             stream = container.streams.video[0]