Merge pull request #28 from AEmotionStudio/bolt-video-memory-optimization-3922534104789050158

⚡ Bolt: Zero-copy video frame writing

discord_video_node.py

@@ -640,7 +640,8 @@ class DiscordSendSaveVideo:
                     
                     # Convert tensor images to bytes
                     # Optimization: Use tensor_to_numpy_uint8 for faster conversion
-                    images_bytes = map(lambda x: tensor_to_numpy_uint8(x).tobytes(), image_sequence)
+                    # Ensure contiguity to avoid ValueError in subprocess.stdin.write
+                    image_chunks = map(lambda x: np.ascontiguousarray(tensor_to_numpy_uint8(x)), image_sequence)
                     
                     # Base ffmpeg arguments
                     args = [
@@ -660,9 +661,9 @@ class DiscordSendSaveVideo:
                         process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
                         
                         # Feed frames to ffmpeg
-                        for image_bytes in images_bytes:
+                        for chunk in image_chunks:
                             pbar.update(1)
-                            process.stdin.write(image_bytes)
+                            process.stdin.write(chunk)
                         
                         # Close stdin and get output
                         process.stdin.close()
@@ -697,7 +698,8 @@ class DiscordSendSaveVideo:
                     i_pix_fmt = 'rgb24'
 
                 # Optimization: Use tensor_to_numpy_uint8 for faster conversion
-                images_bytes = map(lambda x: tensor_to_numpy_uint8(x).tobytes(), image_sequence)
+                # Ensure contiguity to avoid ValueError in subprocess.stdin.write
+                image_chunks = map(lambda x: np.ascontiguousarray(tensor_to_numpy_uint8(x)), image_sequence)
                 
                 # Set up ffmpeg arguments based on format
                 loop_args = []
@@ -793,9 +795,9 @@ class DiscordSendSaveVideo:
                     process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
                     
                     # Feed frames to ffmpeg
-                    for image_bytes in images_bytes:
+                    for chunk in image_chunks:
                         pbar.update(1)
-                        process.stdin.write(image_bytes)
+                        process.stdin.write(chunk)
                     
                     # Close stdin and get output
                     process.stdin.close()
@@ -855,10 +857,13 @@ class DiscordSendSaveVideo:
                             output_file_with_audio_path
                         ]
                         
-                        audio_data = a_waveform.squeeze(0).transpose(0,1).numpy().tobytes()
+                        # Use ascontiguousarray to ensure the transposed array is C-contiguous.
+                        # This avoids BufferError in subprocess.run when using memoryview on non-contiguous arrays.
+                        audio_data = np.ascontiguousarray(a_waveform.squeeze(0).transpose(0,1).numpy())
                         
                         try:
-                            res = subprocess.run(mux_args, input=audio_data, env=env, capture_output=True, check=True)
+                            # Use memoryview to avoid copy while satisfying subprocess.run input check
+                            res = subprocess.run(mux_args, input=memoryview(audio_data), env=env, capture_output=True, check=True)
                             if res.stderr:
                                 print(res.stderr.decode(*ENCODE_ARGS), end="", file=sys.stderr)
                             

tests/test_numpy_subprocess.py

@@ -0,0 +1,101 @@
+import unittest
+import numpy as np
+import subprocess
+import sys
+import os
+
+class TestNumpyToSubprocess(unittest.TestCase):
+    """
+    Verify that subprocess.Popen.stdin.write accepts numpy arrays directly.
+    For subprocess.run(input=...), we need to be careful with numpy arrays due to ambiguity check in subprocess module.
+    """
+
+    def test_popen_stdin_write_numpy(self):
+        """Test writing numpy array to Popen.stdin"""
+        # Create a small numpy array
+        data = np.arange(256, dtype=np.uint8)
+
+        # Use python to echo input to output (cross-platform)
+        cmd = [sys.executable, '-c', 'import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())']
+        p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+        # Write numpy array directly
+        p.stdin.write(data)
+        out, _ = p.communicate()
+
+        # Verify output matches input data bytes
+        self.assertEqual(out, data.tobytes())
+        self.assertEqual(len(out), 256)
+
+    def test_run_input_memoryview(self):
+        """
+        Test passing numpy array as memoryview to subprocess.run input.
+        """
+        data = np.arange(256, dtype=np.uint8)
+
+        # Use python to echo input to output (cross-platform)
+        cmd = [sys.executable, '-c', 'import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())']
+        # memoryview works and avoids copy
+        res = subprocess.run(cmd, input=memoryview(data), capture_output=True)
+
+        self.assertEqual(res.stdout, data.tobytes())
+        self.assertEqual(len(res.stdout), 256)
+
+    def test_run_input_fixed_non_contiguous(self):
+        """
+        Test that using ascontiguousarray makes the non-contiguous array accepted by subprocess.run
+        """
+        # Create a 2D array and transpose it to make it non-contiguous
+        data = np.zeros((10, 10), dtype=np.uint8)
+        # Fill with some data
+        for i in range(10):
+            for j in range(10):
+                data[i, j] = i + j
+
+        # Transpose creates a non-contiguous view
+        transposed_data = data.T
+        self.assertFalse(transposed_data.flags['C_CONTIGUOUS'])
+
+        # Fix it using ascontiguousarray
+        contiguous_data = np.ascontiguousarray(transposed_data)
+        self.assertTrue(contiguous_data.flags['C_CONTIGUOUS'])
+
+        # Now pass to subprocess
+        mv = memoryview(contiguous_data)
+        cmd = [sys.executable, '-c', 'import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())']
+        res = subprocess.run(cmd, input=mv, capture_output=True)
+        self.assertEqual(res.stdout, contiguous_data.tobytes())
+
+    def test_popen_stdin_write_fixed_non_contiguous(self):
+        """
+        Test writing fixed (made contiguous) numpy array to Popen.stdin.
+        """
+        # Create a 2D array and transpose it to make it non-contiguous
+        data = np.zeros((10, 10), dtype=np.uint8)
+        # Fill with some data
+        for i in range(10):
+            for j in range(10):
+                data[i, j] = i + j
+
+        transposed_data = data.T
+        self.assertFalse(transposed_data.flags['C_CONTIGUOUS'])
+
+        # Fix it
+        contiguous_data = np.ascontiguousarray(transposed_data)
+        self.assertTrue(contiguous_data.flags['C_CONTIGUOUS'])
+
+        # Use python to echo input to output (cross-platform)
+        cmd = [sys.executable, '-c', 'import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())']
+        p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+        try:
+            # Should succeed now
+            p.stdin.write(contiguous_data)
+            out, _ = p.communicate()
+            self.assertEqual(out, contiguous_data.tobytes())
+
+        except Exception as e:
+            self.fail(f"Caught unexpected exception: {e}")
+
+if __name__ == "__main__":
+    unittest.main()