Potential for memory leak if using the examples with this implementation.

[kaine-bruce-dmt]

TLDR: The example code is incorrect for the current implementation and does not align with the original NDI spec. This means it is easy to create a memory leak if you're using the example code.

In the NDI SDK documentation, they specifically call NDIlib_recv_capture_v3 (or NDIlib_recv_capture_v2) with a receiver and then the pointers for each frame. You can also supply NULL if you don't want a frame.

However in the code base for ndi-python we specify ALL 3 pointers, which means we need to free all 3 of these pointers.

ndi-python/src/main.cpp

Lines 456 to 488 in 8aad19d

	m.def(
	"recv_capture_v2",
	[](py::capsule instance, uint32_t timeout_in_ms) {
	auto p_instance =
	static_cast<NDIlib_recv_instance_type *>(instance.get_pointer());
	NDIlib_video_frame_v2_t video_frame;
	NDIlib_audio_frame_v2_t audio_frame;
	NDIlib_metadata_frame_t metadata_frame;
	auto type =
	NDIlib_recv_capture_v2(p_instance, &video_frame, &audio_frame,
	&metadata_frame, timeout_in_ms);
	return std::tuple<NDIlib_frame_type_e, NDIlib_video_frame_v2_t,
	NDIlib_audio_frame_v2_t, NDIlib_metadata_frame_t>(
	type, video_frame, audio_frame, metadata_frame);
	},
	py::arg("instance"), py::arg("timeout_in_ms"));
	m.def(
	"recv_capture_v3",
	[](py::capsule instance, uint32_t timeout_in_ms) {
	auto p_instance =
	static_cast<NDIlib_recv_instance_type *>(instance.get_pointer());
	NDIlib_video_frame_v2_t video_frame;
	NDIlib_audio_frame_v3_t audio_frame;
	NDIlib_metadata_frame_t metadata_frame;
	auto type =
	NDIlib_recv_capture_v3(p_instance, &video_frame, &audio_frame,
	&metadata_frame, timeout_in_ms);
	return std::tuple<NDIlib_frame_type_e, NDIlib_video_frame_v2_t,
	NDIlib_audio_frame_v3_t, NDIlib_metadata_frame_t>(
	type, video_frame, audio_frame, metadata_frame);
	},
	py::arg("instance"), py::arg("timeout_in_ms"));

In the examples folder, we do not explicitly free them, so there is potential for memory leaks if this is running for a long time (I got to 10GB after 17hours while testing)

ndi-python/example/recv_cv.py

Lines 37 to 44 in 8aad19d

	while True:
	t, v, _, _ = ndi.recv_capture_v2(ndi_recv, 5000)
	if t == ndi.FRAME_TYPE_VIDEO:
	print('Video data received (%dx%d).' % (v.xres, v.yres))
	frame = np.copy(v.data)
	cv.imshow('ndi image', frame)
	ndi.recv_free_video_v2(ndi_recv, v)

Suggest changing the C++ implementation, or changing the examples so that the other 2 frames are freed.

For anyone looking for a quick solution in their own code you can do something like:

(
    frame_type,
    video_frame,
    audio_frame,
    meta_frame,
) = ndi.recv_capture_v3(ndi_connection, 5000)

# As this implementation of NDI implicitly gives you ALL 3 frame types
# You must make sure you free them, else you will get a memory leak!

if frame_type == ndi.FRAME_TYPE_VIDEO:
    # do stuff
    ndi.recv_free_video_v2(ndi_connection, video_frame)

if frame_type == ndi.FRAME_TYPE_AUDIO:
     # do stuff
    ndi.recv_free_audio_v3(ndi_connection, audio_frame)

if frame_type == ndi.FRAME_TYPE_METADATA:
     # do stuff
    ndi.recv_free_metadata(ndi_connection, meta_frame)

[rsemihkoca]

even though i freed those pointers i think memory leak problem still continues.

with a new connection your memory usage increases but in case of disconnection, memory level stays in same level.

(NDI) ➜ NDI git:(main) pip list
Package Version

---

ndi-python 5.1.1.5
numpy 2.2.1
pip 24.2
psutil 6.1.1
PyQt6 6.8.0
PyQt6-Qt6 6.8.1
PyQt6_sip 13.9.1
setuptools 75.1.0
wheel 0.44.0

in order to reproduce you can run this code

import sys
import logging
import time
import psutil
import os
from typing import Dict, Set, Optional, List
import NDIlib as ndi
from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
                           QHBoxLayout, QListWidget, QListWidgetItem, QLabel)
from PyQt6.QtCore import Qt, QTimer, pyqtSlot
from PyQt6.QtGui import QImage, QPixmap

# Configure logging with more detailed format
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - [%(name)s] - %(message)s'
)
logger = logging.getLogger(__name__)

class ResourceMonitor:
    """Monitors system and application resources."""
    
    @staticmethod
    def get_memory_usage():
        """Get current process memory usage."""
        process = psutil.Process(os.getpid())
        return process.memory_info().rss / 1024 / 1024  # Convert to MB
    
    @staticmethod
    def get_cpu_usage():
        """Get current process CPU usage."""
        process = psutil.Process(os.getpid())
        return process.cpu_percent()

class NDIVideoProcessor:
    """Handles NDI video frame processing and conversion."""
    
    @staticmethod
    def process_frame(frame) -> Optional[QImage]:
        """Process NDI video frame and convert to QImage."""
        if not frame or not hasattr(frame, 'data') or frame.data is None:
            logger.warning("Received invalid frame for processing")
            return None
            
        try:
            height, width = frame.yres, frame.xres
            
            # Create a minimal copy of the data we need
            buffer = frame.data.reshape(height, width, 4)[:, :, :3]  # Only take BGR channels
            bytes_per_line = width * 3
            
            # Create image directly from the buffer
            image = QImage(buffer.tobytes(), width, height, 
                         bytes_per_line, QImage.Format.Format_BGR888).copy()
            
            # Clear buffer reference
            buffer = None
            
            return image
            
        except Exception as e:
            logger.error(f"Error processing frame: {e}")
            return None

class NDIReceiver(QMainWindow):
    def __init__(self):
        super().__init__()
        self.start_time = time.time()
        self.frame_count = 0
        self.resource_monitor = ResourceMonitor()
        self.pending_frames: Dict[str, List] = {}  # Store frames that need cleanup
        self.is_shutting_down = False
        self.source_frame_times: Dict[str, List[float]] = {}  # Track frame times for FPS calculation
        
        logger.info("Initializing NDI Receiver application")
        self._initialize_ui()
        self._initialize_ndi()
        self._setup_timers()
        
        # Log initial resource usage
        mem_usage = self.resource_monitor.get_memory_usage()
        logger.info(f"Initial memory usage: {mem_usage:.2f} MB")

    def _initialize_ui(self):
        """Initialize the UI components."""
        logger.info("Initializing UI components")
        self.setWindowTitle("NDI Receiver")
        self.setGeometry(100, 100, 1200, 800)
        
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QHBoxLayout(central_widget)
        
        # Left side - Source list
        left_widget = QWidget()
        left_layout = QVBoxLayout(left_widget)
        self.source_list = QListWidget()
        self.source_list.itemClicked.connect(self.source_selected)
        left_layout.addWidget(self.source_list)
        
        # Add "All" option
        self.source_list.insertItem(0, QListWidgetItem("All Sources"))
        
        # Right side - Video display
        self.video_widget = QWidget()
        self.video_layout = QVBoxLayout(self.video_widget)
        self.video_labels: Dict[str, (QLabel, QLabel)] = {}
        
        layout.addWidget(left_widget, 1)
        layout.addWidget(self.video_widget, 3)
        
        logger.info("UI components initialized successfully")

    def _initialize_ndi(self):
        """Initialize NDI components."""
        logger.info("Initializing NDI components")
        
        if not ndi.initialize():
            logger.error("Failed to initialize NDI")
            sys.exit(1)
        logger.info("NDI initialized successfully")
            
        self.finder = ndi.find_create_v2()
        if self.finder is None:
            logger.error("Failed to create NDI finder")
            sys.exit(1)
        logger.info("NDI finder created successfully")
            
        self.receivers: Dict[str, any] = {}
        self.active_sources: Set[str] = set()
        
        # Log NDI version info
        logger.info(f"NDI version: {ndi.version()}")

    def _cleanup_pending_frames(self, source_name: Optional[str] = None):
        """Clean up any pending frames for a source or all sources."""
        try:
            if source_name:
                # Clean up frames for specific source
                if source_name in self.pending_frames:
                    frames = self.pending_frames[source_name]
                    receiver = self.receivers.get(source_name)
                    if receiver:
                        for frame in frames:
                            if frame:
                                ndi.recv_free_video_v2(receiver, frame)
                        logger.debug(f"Cleaned up {len(frames)} pending frames for {source_name}")
                    del self.pending_frames[source_name]
            else:
                # Clean up frames for all sources
                for src_name, frames in self.pending_frames.items():
                    receiver = self.receivers.get(src_name)
                    if receiver:
                        for frame in frames:
                            if frame:
                                ndi.recv_free_video_v2(receiver, frame)
                        logger.debug(f"Cleaned up {len(frames)} pending frames for {src_name}")
                self.pending_frames.clear()
        except Exception as e:
            logger.error(f"Error cleaning up pending frames: {e}")

    def _process_source_frame(self, source_name: str, receiver):
        """Process a single source frame."""
        try:
            start_time = time.time()
            result = ndi.recv_capture_v3(receiver, 0)
            
            if not result:
                return
                
            frame_type, video_frame, audio_frame, metadata_frame = result
            
            try:
                if frame_type == ndi.FRAME_TYPE_VIDEO and video_frame:
                    if hasattr(video_frame, 'data') and video_frame.data is not None:
                        self._display_frame(source_name, video_frame)
            finally:
                # Always free all frame types, regardless of which type was received
                # This is necessary due to how the NDI Python binding works
                try:
                    if video_frame:
                        ndi.recv_free_video_v2(receiver, video_frame)
                except Exception as e:
                    logger.error(f"Error freeing video frame: {e}")
                    
                try:
                    if audio_frame:
                        ndi.recv_free_audio_v3(receiver, audio_frame)
                except Exception as e:
                    logger.error(f"Error freeing audio frame: {e}")
                    
                try:
                    if metadata_frame:
                        ndi.recv_free_metadata(receiver, metadata_frame)
                except Exception as e:
                    logger.error(f"Error freeing metadata frame: {e}")
                    
            processing_time = (time.time() - start_time) * 1000
            logger.debug(f"Frame processed for {source_name} in {processing_time:.2f}ms")
                    
        except Exception as e:
            logger.error(f"Error processing frame for {source_name}: {e}")

    def _add_new_source(self, source):
        """Add a new NDI source."""
        try:
            logger.info(f"Adding new NDI source: {source.ndi_name}")
            self.source_list.addItem(source.ndi_name)
            
            recv_settings = ndi.RecvCreateV3(source_to_connect_to=source)
            recv = ndi.recv_create_v3(recv_settings)
            
            if recv:
                self.receivers[source.ndi_name] = recv
                ndi.recv_connect(recv, source)
                logger.info(f"Successfully connected to source: {source.ndi_name}")
            else:
                logger.error(f"Failed to create receiver for source: {source.ndi_name}")
                
        except Exception as e:
            logger.error(f"Error adding source {source.ndi_name}: {e}")

    def _remove_disconnected_sources(self, sources):
        """Remove disconnected NDI sources."""
        active_source_names = {s.ndi_name for s in sources}
        
        for i in range(self.source_list.count() - 1, 0, -1):
            item_text = self.source_list.item(i).text()
            if item_text not in active_source_names:
                logger.info(f"Removing disconnected source: {item_text}")
                self._cleanup_source(item_text)
                self.source_list.takeItem(i)

    def update_frames(self):
        """Update video frames for active sources."""
        if self.is_shutting_down:
            return
            
        # Only increment frame count if we have active sources
        if self.receivers:
            self.frame_count += 1
            
        for source_name, receiver in self.receivers.items():
            if source_name in self.active_sources or "All Sources" in self.active_sources:
                self._process_source_frame(source_name, receiver)

    def _setup_timers(self):
        """Setup update timers for sources and frames."""
        logger.info("Setting up timers")
        
        self.source_timer = QTimer()
        self.source_timer.timeout.connect(self.update_sources)
        self.source_timer.start(1000)  # Check sources every second
        
        self.frame_timer = QTimer()
        self.frame_timer.timeout.connect(self.update_frames)
        self.frame_timer.start(33)  # ~30 FPS
        
        # Add resource monitoring timer
        self.resource_timer = QTimer()
        self.resource_timer.timeout.connect(self._log_resource_usage)
        self.resource_timer.start(5000)  # Log every 5 seconds
        
        logger.info("Timers initialized successfully")

    def _log_resource_usage(self):
        """Log current resource usage."""
        mem_usage = self.resource_monitor.get_memory_usage()
        cpu_usage = self.resource_monitor.get_cpu_usage()
        uptime = time.time() - self.start_time
        fps = self.frame_count / uptime if uptime > 0 else 0
        
        logger.info(
            f"Resource Usage - Memory: {mem_usage:.2f}MB, CPU: {cpu_usage:.1f}%, "
            f"Active Sources: {len(self.active_sources)}, "
            f"Average FPS: {fps:.2f}"
        )
        
    def update_sources(self):
        """Update NDI sources list."""
        if self.is_shutting_down:
            return
            
        try:
            sources = ndi.find_get_current_sources(self.finder)
            current_items = {self.source_list.item(i).text() 
                           for i in range(1, self.source_list.count())}
            
            logger.debug(f"Found {len(sources)} NDI sources")
            
            # Add new sources
            for source in sources:
                source_name = source.ndi_name
                if source_name not in current_items:
                    self._add_new_source(source)
            
            # Remove disconnected sources
            self._remove_disconnected_sources(sources)
                    
        except Exception as e:
            logger.error(f"Error updating sources: {e}")

    @pyqtSlot(QListWidgetItem)
    def source_selected(self, item):
        """Handle source selection."""
        selected_source = item.text()
        logger.info(f"Source selected: {selected_source}")
        
        for video_label, info_label in self.video_labels.values():
            video_label.hide()
            info_label.hide()
            
        self.active_sources = {"All Sources"} if selected_source == "All Sources" else {selected_source}
        
        if selected_source == "All Sources":
            logger.info("Showing all sources")
            for video_label, info_label in self.video_labels.values():
                video_label.show()
                info_label.show()
        elif selected_source in self.video_labels:
            logger.info(f"Showing single source: {selected_source}")
            video_label, info_label = self.video_labels[selected_source]
            video_label.show()
            info_label.show()

    def _display_frame(self, source_name: str, frame):
        """Display a video frame."""
        try:
            if source_name not in self.video_labels:
                logger.info(f"Creating new video label for source: {source_name}")
                container = QWidget()
                layout = QVBoxLayout(container)
                
                # Info label for source details
                info_label = QLabel()
                info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
                layout.addWidget(info_label)
                
                # Video label
                video_label = QLabel()
                video_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
                layout.addWidget(video_label)
                
                self.video_layout.addWidget(container)
                self.video_labels[source_name] = (video_label, info_label)
            
            # Calculate FPS
            current_time = time.time()
            if source_name not in self.source_frame_times:
                self.source_frame_times[source_name] = []
            
            frame_times = self.source_frame_times[source_name]
            frame_times.append(current_time)
            
            # Keep only last 30 frames and remove old ones
            while len(frame_times) > 30:
                frame_times.pop(0)
            
            fps = len(frame_times) / (frame_times[-1] - frame_times[0]) if len(frame_times) > 1 else 0
            
            video_label, info_label = self.video_labels[source_name]
            
            # Process frame and create pixmap in a controlled way
            try:
                image = NDIVideoProcessor.process_frame(frame)
                if image:
                    try:
                        # Clear old pixmap first
                        video_label.setPixmap(QPixmap())
                        
                        # Create and set new pixmap
                        pixmap = QPixmap.fromImage(image)
                        scaled_pixmap = pixmap.scaled(video_label.size(), 
                                                    Qt.AspectRatioMode.KeepAspectRatio)
                        video_label.setPixmap(scaled_pixmap)
                        
                        # Update source information
                        info_text = (f"Source: {source_name}\n"
                                   f"Resolution: {frame.xres}x{frame.yres}\n"
                                   f"Format: {frame.FourCC}\n"
                                   f"FPS: {fps:.1f}")
                        info_label.setText(info_text)
                    finally:
                        # Ensure image is cleared
                        image = None
                        pixmap = None
                        scaled_pixmap = None
                        
                        # Process events to ensure UI updates
                        QApplication.processEvents()
            except Exception as e:
                logger.error(f"Error processing image for {source_name}: {e}")
                
        except Exception as e:
            logger.error(f"Error displaying frame for {source_name}: {e}")

    def _cleanup_source(self, source_name: str):
        """Clean up resources for a disconnected source."""
        try:
            # Stop processing frames for this source
            if source_name in self.active_sources:
                self.active_sources.remove(source_name)
            
            # Clean up frame timing data
            if source_name in self.source_frame_times:
                del self.source_frame_times[source_name]
            
            # Clean up UI elements first
            if source_name in self.video_labels:
                logger.info(f"Removing video label for source: {source_name}")
                video_label, info_label = self.video_labels[source_name]
                
                try:
                    # Clear and delete video label
                    if video_label:
                        video_label.setPixmap(QPixmap())  # Clear the pixmap
                        video_label.clear()  # Clear any other content
                        video_label.deleteLater()
                    
                    # Clear and delete info label
                    if info_label:
                        info_label.clear()
                        info_label.deleteLater()
                    
                    # Get and delete the container widget
                    container = video_label.parent()
                    if container:
                        container.deleteLater()
                        
                    # Process events to ensure widgets are deleted
                    QApplication.processEvents()
                    
                finally:
                    del self.video_labels[source_name]
            
            # Clean up receiver
            if source_name in self.receivers:
                logger.info(f"Destroying receiver for source: {source_name}")
                receiver = self.receivers[source_name]
                
                if receiver:
                    # Flush any remaining frames with a timeout
                    try:
                        for _ in range(5):  # Try up to 5 times
                            result = ndi.recv_capture_v3(receiver, 100)  # 100ms timeout
                            if not result:
                                break
                            frame_type, video_frame, audio_frame, metadata_frame = result
                            try:
                                if video_frame:
                                    ndi.recv_free_video_v2(receiver, video_frame)
                                if audio_frame:
                                    ndi.recv_free_audio_v3(receiver, audio_frame)
                                if metadata_frame:
                                    ndi.recv_free_metadata(receiver, metadata_frame)
                            except Exception as e:
                                logger.warning(f"Error freeing frames during cleanup: {e}")
                    except Exception as e:
                        logger.warning(f"Error flushing frames during cleanup: {e}")
                    finally:
                        # Always destroy the receiver
                        ndi.recv_destroy(receiver)
                        logger.debug(f"Receiver destroyed for source: {source_name}")
                
                del self.receivers[source_name]
            
            # Reset frame count and timing data if no sources left
            if not self.receivers:
                self.frame_count = 0
                self.start_time = time.time()
                self.source_frame_times.clear()
            
            # Force garbage collection
            import gc
            gc.collect()
            
            # Process any pending events
            QApplication.processEvents()
                
            logger.info(f"Successfully cleaned up all resources for source: {source_name}")
                
        except Exception as e:
            logger.error(f"Error cleaning up source {source_name}: {e}")

    def closeEvent(self, event):
        """Clean up resources before closing."""
        logger.info("Application shutting down, cleaning up resources...")
        self.is_shutting_down = True
        
        try:
            # Stop timers first
            self.source_timer.stop()
            self.frame_timer.stop()
            self.resource_timer.stop()
            
            # Log final statistics
            uptime = time.time() - self.start_time
            avg_fps = self.frame_count / uptime if uptime > 0 else 0
            logger.info(
                f"Final Statistics - Uptime: {uptime:.1f}s, "
                f"Total Frames: {self.frame_count}, "
                f"Average FPS: {avg_fps:.2f}"
            )
            
            # Clean up all sources
            source_names = list(self.receivers.keys())
            for source_name in source_names:
                self._cleanup_source(source_name)
            
            # Clean up video widget and central widget
            if self.video_widget:
                self.video_widget.deleteLater()
            
            if self.centralWidget():
                self.centralWidget().deleteLater()
            
            # Process events to ensure widgets are deleted
            QApplication.processEvents()
            
            # Clear all data structures
            self.receivers.clear()
            self.active_sources.clear()
            self.source_frame_times.clear()
            self.video_labels.clear()
            
            # Final NDI cleanup
            if hasattr(self, 'finder') and self.finder:
                ndi.find_destroy(self.finder)
                self.finder = None
            
            ndi.destroy()
            
            # Force garbage collection
            import gc
            gc.collect()
            
            logger.info("NDI resources cleaned up successfully")
            
        except Exception as e:
            logger.error(f"Error during cleanup: {e}")
        finally:
            super().closeEvent(event)

if __name__ == "__main__":
    try:
        logger.info("Starting NDI Receiver application")
        app = QApplication(sys.argv)
        window = NDIReceiver()
        window.show()
        sys.exit(app.exec())
    except Exception as e:
        logger.critical(f"Application failed to start: {e}")
        sys.exit(1)

[kaine-bruce-dmt]

You've probably noticed this repo is abandoned.

If you have the time and you also wish to help others in the same boat, that you distill this down to an minimal reproducible example.

You might be right, but it's non-trivial to confirm with your own abstractions.

<3