Allow opening OpusFile/OpusFileStream from memory.

pyogg/opus_file.py

@@ -1,25 +1,29 @@
 import ctypes
+from typing import Union
 
 from . import ogg
 from . import opus
 from .pyogg_error import PyOggError
 from .audio_file import AudioFile
 
 class OpusFile(AudioFile):
-    def __init__(self, path: str) -> None:
-        # Open the file
+    def __init__(self, path_or_data: Union[str, bytes]):
         error = ctypes.c_int()
-        of = opus.op_open_file(
-            ogg.to_char_p(path),
-            ctypes.pointer(error)
-        )
-
-        # Check for errors 
-        if error.value != 0:
-            raise PyOggError(
-                ("File '{}' couldn't be opened or doesn't exist. "+
-                 "Error code: {}").format(path, error.value)
-            )
+        if isinstance(path_or_data, str):
+            # Open the file
+            of = opus.op_open_file(ogg.to_char_p(path_or_data), ctypes.pointer(error))
+            # Check for errors 
+            if error.value != 0:
+                raise PyOggError(
+                    ("File '{}' couldn't be opened or doesn't exist. "+
+                    "Error code: {}").format(path_or_data, error.value)
+                )
+        else:
+            # Open from memory; avoid creating an unnecessary copy, since op_open_memory does not mutate data.
+            data = ctypes.cast(ctypes.c_char_p(path_or_data), ctypes.POINTER(ctypes.c_ubyte))
+            of = opus.op_open_memory(data, len(path_or_data), ctypes.pointer(error))
+            if error.value != 0:
+                raise PyOggError("Could not open from memory. Error code: {}".format(error.value))
 
         # Extract the number of channels in the newly opened file
         #: Number of channels in audio file.

pyogg/opus_file_stream.py

@@ -1,27 +1,38 @@
 import ctypes
+from typing import Union
 
 from . import ogg
 from . import opus
 from .pyogg_error import PyOggError
 
 class OpusFileStream:
-    def __init__(self, path):
+    def __init__(self, path_or_data: Union[str, bytes]):
         """Opens an OggOpus file as a stream.
 
-        path should be a string giving the filename of the file to
-        open.  Unicode file names may not work correctly.
+        path_or_data should be a string giving the filename of the file to
+        open, or a buffer containing the OggOpus data. Unicode file names may
+        not work correctly.
 
         An exception will be raised if the file cannot be opened
         correctly.
-
-        """ 
+        """
         error = ctypes.c_int()
-
-        self.of = opus.op_open_file(ogg.to_char_p(path), ctypes.pointer(error))
-
-        if error.value != 0:
-            self.of = None
-            raise PyOggError("file couldn't be opened or doesn't exist. Error code : {}".format(error.value))
+        if isinstance(path_or_data, str):
+            # Open the file
+            self.of = opus.op_open_file(ogg.to_char_p(path_or_data), ctypes.pointer(error))
+            # Check for errors 
+            if error.value != 0:
+                raise PyOggError(
+                    ("File '{}' couldn't be opened or doesn't exist. "+
+                    "Error code: {}").format(path_or_data, error.value)
+                )
+        else:
+            # Open from memory; avoid creating an unnecessary copy, since op_open_memory does not mutate data.
+            self._data = path_or_data  # Keep a reference around to prevent garbage collection.
+            data = ctypes.cast(ctypes.c_char_p(path_or_data), ctypes.POINTER(ctypes.c_ubyte))
+            self.of = opus.op_open_memory(data, len(self._data), ctypes.pointer(error))
+            if error.value != 0:
+                raise PyOggError("Could not open from memory. Error code: {}".format(error.value))
 
         #: Number of channels in audio file
         self.channels = opus.op_channel_count(self.of, -1)

tests/test_opus_file.py

@@ -1,6 +1,5 @@
 import pytest
 import pyogg
-import os
 
 from config import Config
 
@@ -86,3 +85,17 @@ def test_output_via_wav(pyogg_config: Config) -> None:
     wave_out.setsampwidth(opus_file.bytes_per_sample)
     wave_out.setframerate(opus_file.frequency)
     wave_out.writeframes(opus_file.buffer)
+
+
+def test_from_memory(pyogg_config: Config) -> None:
+    # Load the demonstration file that is exactly 5 seconds long
+    filename = str(
+        pyogg_config.rootdir
+        / "examples/left-right-demo-5s.opus"
+    )
+    # Load the file into memory, then into OpusFile.
+    with open(filename, "rb") as f:
+        from_memory = pyogg.OpusFile(f.read())
+    # For comparison, load the file directly with OpusFile.
+    from_file = pyogg.OpusFile(filename)
+    assert bytes(from_memory.buffer) == bytes(from_file.buffer)

tests/test_opus_file_stream.py

@@ -115,4 +115,31 @@ def test_same_data_as_opus_file_using_as_array(pyogg_config: Config):
 
     # Check that every byte is identical for both buffers
     assert numpy.all(buf_all == opus_file.as_array())
-
+
+
+def test_from_memory(pyogg_config: Config) -> None:
+    # Load the demonstration file that is exactly 5 seconds long
+    filename = str(
+        pyogg_config.rootdir
+        / "examples/left-right-demo-5s.opus"
+    )
+
+    # Load the file into memory, then into OpusFileStream.
+    with open(filename, "rb") as f:
+        from_memory = pyogg.OpusFileStream(f.read())
+    # For comparison, load directly with OpusFile.
+    from_file = pyogg.OpusFileStream(filename)
+
+    # Loop through the OpusFileStreams until we've read all the data
+    while True:
+        # Read the next part of the stream
+        from_memory_buf = from_memory.get_buffer()
+        from_file_buf = from_file.get_buffer()
+
+        # Check if we've reached the end of the stream
+        if from_memory_buf is None or from_file_buf is None:
+            break
+        # Check that every byte is identical for both buffers
+        assert bytes(from_memory_buf) == bytes(from_file_buf)
+    # Check that we've reached the end of both streams
+    assert from_memory_buf is None and from_file_buf is None