cleanup

pyav/demuxer.py

@@ -0,0 +1,69 @@
+# People's Video Editor: high quality, GPU accelerated mp4 editor
+# Copyright (C) 2025 Roz K <roz@rozk.net>
+#
+# This file is part of People's Video Editor.
+#
+# People's Video Editor is free software: you can redistribute it and/or modify it under the terms of the
+# GNU General Public License as published by the Free Software Foundation, either version 3 of the License,
+# or (at your option) any later version.
+#
+# People's Video Editor is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+# See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along with People's Video Editor.
+# If not, see <https://www.gnu.org/licenses/>.
+
+from . import libav
+from .codec import Codec
+from .stream import Stream
+from .packet import Packet
+
+class Demuxer:
+    __slots__ = '_context', 'video_stream', 'audio_stream'
+
+    def __init__(self, path):
+        self._context = libav.format_alloc_context()
+        if not self._context:
+            raise MemoryError
+        errcode = libav.format_open_input(self._context, "file:" + path)
+        if errcode < 0:
+            raise Exception(f"Failed to open: {path}")
+        errcode = libav.format_find_stream_info(self._context)
+        if errcode < 0:
+            libav.format_close_input(self._context)
+            raise Exception("Failed to find stream info")
+        self.video_stream = self._find_stream(libav.AVMEDIA_TYPE_VIDEO)
+        if self.video_stream is None:
+            libav.format_close_input(self._context)
+            raise Exception("Failed to find a video stream")
+        self.audio_stream = self._find_stream(libav.AVMEDIA_TYPE_AUDIO)
+        if self.audio_stream is None:
+            libav.format_close_input(self._context)
+            raise Exception("Failed to find an audio stream")
+
+    def __del__(self):
+        if self._context:
+            libav.format_close_input(self._context)
+
+    def _find_stream(self, type):
+        index, codec_ref = libav.format_find_best_stream(self._context, type)
+        if index < 0 or not codec_ref:
+            return None
+        parameters = self._context.contents.streams[index].contents.codecpar
+        return Stream(index, Codec(codec_ref), parameters)
+
+    @property
+    def nb_streams(self):
+        if not self._context:
+            return 0
+        return self._context.contents.nb_streams
+
+    def read_packet(self):
+        if not self._context:
+            return None
+        packet = Packet()
+        errcode = libav.read_frame(self._context, packet)
+        if errcode < 0:
+            return None
+        return packet