Issue #19641: Added the audioop.byteswap() function to convert big-endian

samples to little-endian and vice versa.

6 files changed, 34 insertions, 78 deletions

diff --git a/Lib/test/audiotests.py b/Lib/test/audiotests.py
index 7581fe2e..b7497ce 100644
--- a/Lib/test/audiotests.py
+++ b/Lib/test/audiotests.py
@@ -5,24 +5,6 @@ import io
 import pickle
 import sys
 
-def byteswap2(data):
-    a = array.array('h')
-    a.frombytes(data)
-    a.byteswap()
-    return a.tobytes()
-
-def byteswap3(data):
-    ba = bytearray(data)
-    ba[::3] = data[2::3]
-    ba[2::3] = data[::3]
-    return bytes(ba)
-
-def byteswap4(data):
-    a = array.array('i')
-    a.frombytes(data)
-    a.byteswap()
-    return a.tobytes()
-
 class UnseekableIO(io.FileIO):
     def tell(self):
         raise io.UnsupportedOperation
diff --git a/Lib/test/test_aifc.py b/Lib/test/test_aifc.py
index b4270bc..041b236 100644
--- a/Lib/test/test_aifc.py
+++ b/Lib/test/test_aifc.py
@@ -1,6 +1,7 @@
 from test.support import findfile, TESTFN, unlink
 import unittest
 from test import audiotests
+from audioop import byteswap
 import os
 import io
 import sys
@@ -122,7 +123,7 @@ class AifcULAWTest(AifcTest, unittest.TestCase):
       E5040CBC 617C0A3C 08BC0A3C 2C7C0B3C 517C0E3C 8A8410FC B6840EBC 457C0A3C \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap2(frames)
+        frames = byteswap(frames, 2)
 
 
 class AifcALAWTest(AifcTest, unittest.TestCase):
@@ -143,7 +144,7 @@ class AifcALAWTest(AifcTest, unittest.TestCase):
       E4800CC0 62000A40 08C00A40 2B000B40 52000E40 8A001180 B6000EC0 46000A40 \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap2(frames)
+        frames = byteswap(frames, 2)
 
 
 class AifcMiscTest(audiotests.AudioTests, unittest.TestCase):
diff --git a/Lib/test/test_audioop.py b/Lib/test/test_audioop.py
index fe96b75..d507545 100644
--- a/Lib/test/test_audioop.py
+++ b/Lib/test/test_audioop.py
@@ -448,6 +448,23 @@ class TestAudioop(unittest.TestCase):
             self.assertEqual(audioop.getsample(data, w, 3), maxvalues[w])
             self.assertEqual(audioop.getsample(data, w, 4), minvalues[w])
 
+    def test_byteswap(self):
+        swapped_datas = {
+            1: datas[1],
+            2: packs[2](0, 0x3412, 0x6745, -0x6646, -0x81, 0x80, -1),
+            3: packs[3](0, 0x563412, -0x7698bb, 0x7798ba, -0x81, 0x80, -1),
+            4: packs[4](0, 0x78563412, -0x547698bb, 0x557698ba,
+                        -0x81, 0x80, -1),
+        }
+        for w in 1, 2, 3, 4:
+            self.assertEqual(audioop.byteswap(b'', w), b'')
+            self.assertEqual(audioop.byteswap(datas[w], w), swapped_datas[w])
+            self.assertEqual(audioop.byteswap(swapped_datas[w], w), datas[w])
+            self.assertEqual(audioop.byteswap(bytearray(datas[w]), w),
+                             swapped_datas[w])
+            self.assertEqual(audioop.byteswap(memoryview(datas[w]), w),
+                             swapped_datas[w])
+
     def test_negativelen(self):
         # from issue 3306, previously it segfaulted
         self.assertRaises(audioop.error,
diff --git a/Lib/test/test_sunau.py b/Lib/test/test_sunau.py
index 81acd96..af9ffec 100644
--- a/Lib/test/test_sunau.py
+++ b/Lib/test/test_sunau.py
@@ -1,6 +1,7 @@
 from test.support import TESTFN
 import unittest
 from test import audiotests
+from audioop import byteswap
 import sys
 import sunau
 
@@ -124,7 +125,7 @@ class SunauULAWTest(audiotests.AudioWriteTests,
       E5040CBC 617C0A3C 08BC0A3C 2C7C0B3C 517C0E3C 8A8410FC B6840EBC 457C0A3C \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap2(frames)
+        frames = byteswap(frames, 2)
 
 
 if __name__ == "__main__":
diff --git a/Lib/test/test_wave.py b/Lib/test/test_wave.py
index 5be1251..549ca89 100644
--- a/Lib/test/test_wave.py
+++ b/Lib/test/test_wave.py
@@ -1,6 +1,7 @@
 from test.support import TESTFN
 import unittest
 from test import audiotests
+from audioop import byteswap
 import sys
 import wave
 
@@ -46,13 +47,7 @@ class WavePCM16Test(audiotests.AudioWriteTests,
       E4B50CEB 63440A5A 08CA0A1F 2BBA0B0B 51460E47 8BCB113C B6F50EEA 44150A59 \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap2(frames)
-
-    if sys.byteorder == 'big':
-        @unittest.expectedFailure
-        def test_unseekable_incompleted_write(self):
-            super().test_unseekable_incompleted_write()
-
+        frames = byteswap(frames, 2)
 
 
 class WavePCM24Test(audiotests.AudioWriteTests,
@@ -82,7 +77,7 @@ class WavePCM24Test(audiotests.AudioWriteTests,
       51486F0E44E1 8BCC64113B05 B6F4EC0EEB36 4413170A5B48 \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap3(frames)
+        frames = byteswap(frames, 3)
 
 
 class WavePCM32Test(audiotests.AudioWriteTests,
@@ -112,12 +107,7 @@ class WavePCM32Test(audiotests.AudioWriteTests,
       51486F800E44E190 8BCC6480113B0580 B6F4EC000EEB3630 441317800A5B48A0 \
       """)
     if sys.byteorder != 'big':
-        frames = audiotests.byteswap4(frames)
-
-    if sys.byteorder == 'big':
-        @unittest.expectedFailure
-        def test_unseekable_incompleted_write(self):
-            super().test_unseekable_incompleted_write()
+        frames = byteswap(frames, 4)
 
 
 if __name__ == '__main__':
diff --git a/Lib/wave.py b/Lib/wave.py
index 672d04b..b56395e 100644
--- a/Lib/wave.py
+++ b/Lib/wave.py
@@ -82,17 +82,12 @@ WAVE_FORMAT_PCM = 0x0001
 
 _array_fmts = None, 'b', 'h', None, 'i'
 
+import audioop
 import struct
 import sys
 from chunk import Chunk
 from collections import namedtuple
 
-def _byteswap3(data):
-    ba = bytearray(data)
-    ba[::3] = data[2::3]
-    ba[2::3] = data[::3]
-    return bytes(ba)
-
 _wave_params = namedtuple('_wave_params',
                      'nchannels sampwidth framerate nframes comptype compname')
 
@@ -243,29 +238,9 @@ class Wave_read:
             self._data_seek_needed = 0
         if nframes == 0:
             return b''
-        if self._sampwidth in (2, 4) and sys.byteorder == 'big':
-            # unfortunately the fromfile() method does not take
-            # something that only looks like a file object, so
-            # we have to reach into the innards of the chunk object
-            import array
-            chunk = self._data_chunk
-            data = array.array(_array_fmts[self._sampwidth])
-            assert data.itemsize == self._sampwidth
-            nitems = nframes * self._nchannels
-            if nitems * self._sampwidth > chunk.chunksize - chunk.size_read:
-                nitems = (chunk.chunksize - chunk.size_read) // self._sampwidth
-            data.fromfile(chunk.file.file, nitems)
-            # "tell" data chunk how much was read
-            chunk.size_read = chunk.size_read + nitems * self._sampwidth
-            # do the same for the outermost chunk
-            chunk = chunk.file
-            chunk.size_read = chunk.size_read + nitems * self._sampwidth
-            data.byteswap()
-            data = data.tobytes()
-        else:
-            data = self._data_chunk.read(nframes * self._framesize)
-            if self._sampwidth == 3 and sys.byteorder == 'big':
-                data = _byteswap3(data)
+        data = self._data_chunk.read(nframes * self._framesize)
+        if self._sampwidth != 1 and sys.byteorder == 'big':
+            data = audioop.byteswap(data, self._sampwidth)
         if self._convert and data:
             data = self._convert(data)
         self._soundpos = self._soundpos + len(data) // (self._nchannels * self._sampwidth)
@@ -441,20 +416,10 @@ class Wave_write:
         nframes = len(data) // (self._sampwidth * self._nchannels)
         if self._convert:
             data = self._convert(data)
-        if self._sampwidth in (2, 4) and sys.byteorder == 'big':
-            import array
-            a = array.array(_array_fmts[self._sampwidth])
-            a.frombytes(data)
-            data = a
-            assert data.itemsize == self._sampwidth
-            data.byteswap()
-            data.tofile(self._file)
-            self._datawritten = self._datawritten + len(data) * self._sampwidth
-        else:
-            if self._sampwidth == 3 and sys.byteorder == 'big':
-                data = _byteswap3(data)
-            self._file.write(data)
-            self._datawritten = self._datawritten + len(data)
+        if self._sampwidth != 1 and sys.byteorder == 'big':
+            data = audioop.byteswap(data, self._sampwidth)
+        self._file.write(data)
+        self._datawritten += len(data)
         self._nframeswritten = self._nframeswritten + nframes
 
     def writeframes(self, data):