Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
cable / usr / lib / python3 / dist-packages / cables / port_manager.py
Size: Mime:
# cables/port_manager.py
"""
Manages port discovery, filtering, and population of Audio/MIDI tree widgets.
"""
import jack
from PyQt6.QtCore import Qt
from cables.jack_service import get_jack_service
from cables.utils.sort_utils import natural_sort_key_for_full_port_name

import logging
from typing import TYPE_CHECKING, Optional, List, Dict, Any, Tuple, Set

if TYPE_CHECKING:
    from cables.connection_manager import JackConnectionManager
    from cables.features.node_visibility_manager import NodeVisibilityManager
    from cables.ui.port_tree_widget import PortTreeWidget
    from PyQt6.QtWidgets import QLineEdit, QTreeWidget

logger = logging.getLogger(__name__)

class PortManager:
    """Manages fetching, sorting, and filtering of JACK ports."""

    def __init__(self, connection_manager: 'JackConnectionManager', jack_client: Optional[jack.Client], input_filter_edit: Optional['QLineEdit'], output_filter_edit: Optional['QLineEdit']) -> None:
        """
        Initialize the PortManager. Trees are set later via set_trees().

        Args:
            connection_manager: The main JackConnectionManager instance.
            jack_client: The jack.Client instance.
            input_filter_edit: The QLineEdit for input filtering.
            output_filter_edit: The QLineEdit for output filtering.
        """
        self.connection_manager = connection_manager
        self.jack_client = jack_client
        self.input_filter_edit = input_filter_edit
        self.output_filter_edit = output_filter_edit
        self._jack_service = get_jack_service()
        
        # Node visibility manager will be set by JackConnectionManager
        self.node_visibility_manager: Optional['NodeVisibilityManager'] = None

        # Initialize trees as None, they will be set by set_trees()
        self.input_tree: Optional['PortTreeWidget'] = None
        self.output_tree: Optional['PortTreeWidget'] = None
        self.midi_input_tree: Optional['PortTreeWidget'] = None
        self.midi_output_tree: Optional['PortTreeWidget'] = None

        # Do NOT connect filter signals here yet

    def set_trees(self, input_tree: 'PortTreeWidget', output_tree: 'PortTreeWidget', midi_input_tree: 'PortTreeWidget', midi_output_tree: 'PortTreeWidget') -> None:
        """
        Set the tree widgets and connect filter signals. Called after trees are created.

        Args:
            input_tree: The QTreeWidget for audio input ports.
            output_tree: The QTreeWidget for audio output ports.
            midi_input_tree: The QTreeWidget for MIDI input ports.
            midi_output_tree: The QTreeWidget for MIDI output ports.
        """
        self.input_tree = input_tree
        self.output_tree = output_tree
        self.midi_input_tree = midi_input_tree
        self.midi_output_tree = midi_output_tree

        # Connect filter signals now that trees and filters exist
        if self.input_filter_edit:
            # Style to match Graph tab's filter
            self.input_filter_edit.setFrame(False)
            self.input_filter_edit.setStyleSheet("""
                QLineEdit {
                    border: 1px solid rgba(0, 0, 0, 0.1);
                    border-radius: 3px;
                    padding: 2px 5px;
                }
                QLineEdit:focus {
                    border: 1px solid rgba(0, 85, 255, 0.3);
                }
            """)
            # Ensure no duplicate connections if called multiple times
            try: self.input_filter_edit.textChanged.disconnect(self._handle_filter_change)
            except TypeError: pass
            self.input_filter_edit.textChanged.connect(self._handle_filter_change)

        if self.output_filter_edit:
            # Style to match Graph tab's filter
            self.output_filter_edit.setFrame(False)
            self.output_filter_edit.setStyleSheet("""
                QLineEdit {
                    border: 1px solid rgba(0, 0, 0, 0.1);
                    border-radius: 3px;
                    padding: 2px 5px;
                }
                QLineEdit:focus {
                    border: 1px solid rgba(0, 85, 255, 0.3);
                }
            """)
            try: self.output_filter_edit.textChanged.disconnect(self._handle_filter_change)
            except TypeError: pass
            self.output_filter_edit.textChanged.connect(self._handle_filter_change)

    def set_node_visibility_manager(self, node_visibility_manager: 'NodeVisibilityManager') -> None:
        """
        Set the node visibility manager.
        
        Args:
            node_visibility_manager: The NodeVisibilityManager instance
        """
        self.node_visibility_manager = node_visibility_manager

    def _get_ports(self, is_midi_tab: bool) -> Tuple[List[str], List[str]]:
        """
        Get the input and output ports using jack_utils.

        Args:
            is_midi_tab: Whether to get MIDI ports (for MIDI tab) or Audio ports (for Audio tab)

        Returns:
            tuple: A tuple containing the sorted input and output port names
        """
        input_port_names = []
        output_port_names = []

        if not self.jack_client:
            return input_port_names, output_port_names

        try:
            if is_midi_tab:
                # For MIDI tab, get MIDI ports
                input_port_objects = self._jack_service.get_ports(is_input=True, is_midi=True)
                output_port_objects = self._jack_service.get_ports(is_output=True, is_midi=True)
            else:
                # For Audio tab, get Audio ports (explicitly not MIDI)
                input_port_objects = self._jack_service.get_ports(is_input=True, is_audio=True)
                output_port_objects = self._jack_service.get_ports(is_output=True, is_audio=True)

            # Filter ports by visibility if node_visibility_manager is available
            if self.node_visibility_manager:
                # Determine tab type based on is_midi_tab parameter
                tab_type = "midi" if is_midi_tab else "audio"
                
                # For input ports, only check input visibility
                input_port_names = []
                for p in input_port_objects:
                    client_name = p.name.split(':', 1)[0] if ':' in p.name else p.name
                    if self.node_visibility_manager.is_input_visible(client_name, is_midi=is_midi_tab, tab_type=tab_type):
                        input_port_names.append(p.name)
                
                # For output ports, only check output visibility
                output_port_names = []
                for p in output_port_objects:
                    client_name = p.name.split(':', 1)[0] if ':' in p.name else p.name
                    if self.node_visibility_manager.is_output_visible(client_name, is_midi=is_midi_tab, tab_type=tab_type):
                        output_port_names.append(p.name)
            else:
                input_port_names = [p.name for p in input_port_objects]
                output_port_names = [p.name for p in output_port_objects]

            input_port_names = self._sort_ports(input_port_names)
            output_port_names = self._sort_ports(output_port_names)

        except jack.JackError as e: # This might be redundant if jack_utils handles it, but good for safety.
            logger.error(f"Error getting ports via jack_utils: {e}")
            # jack_utils functions return [] on JackError, so lists will be empty.
            pass
        
        return input_port_names, output_port_names

    def _sort_ports(self, port_names: List[str]) -> List[str]:
        """
        Sort port names in a logical order, grouping by base name.
        
        For example, ports like:
        - Equalizer:input_FL
        - Equalizer:input_FL-448
        - Equalizer:input_FL-458
        - Equalizer:input_FR
        - Equalizer:input_FR-449
        - Equalizer:input_FR-459
        
        Will be sorted as:
        - Equalizer:input_FL
        - Equalizer:input_FR
        - Equalizer:input_FL-448
        - Equalizer:input_FR-449
        - Equalizer:input_FL-458
        - Equalizer:input_FR-459

        Args:
            port_names: The port names to sort

        Returns:
            list: The sorted port names
        """
        return sorted(port_names, key=natural_sort_key_for_full_port_name)

    def filter_ports(self, tree_widget: Optional['QTreeWidget'], filter_text: str) -> None:
        """
        Filters the items in the specified tree widget based on the filter text.

        Args:
            tree_widget: The tree widget to filter
            filter_text: The filter text
        """
        if not tree_widget: # Guard against None tree during initialization phases
             return

        filter_text_lower = filter_text.lower()
        terms = filter_text_lower.split()
        include_terms = [term for term in terms if not term.startswith('-')]
        exclude_terms = [term[1:] for term in terms if term.startswith('-') and len(term) > 1]  # Remove '-'

        # Iterate through all top-level items (groups)
        for i in range(tree_widget.topLevelItemCount()):
            group_item = tree_widget.topLevelItem(i)
            group_visible = False  # Assume group is hidden unless a child matches

            # Iterate through children (ports) of the group
            for j in range(group_item.childCount()):
                port_item = group_item.child(j)
                port_name = port_item.data(0, Qt.ItemDataRole.UserRole)  # Get full port name
                if not port_name:  # Skip if port name is invalid
                    port_item.setHidden(True)
                    continue

                port_name_lower = port_name.lower()

                # 1. Check exclusion terms
                excluded = False
                for term in exclude_terms:
                    if term in port_name_lower:
                        excluded = True
                        break
                if excluded:
                    port_item.setHidden(True)
                    continue  # Skip to next port if excluded

                # 2. Check inclusion terms (all must match)
                included = True
                if include_terms:  # Only check if there are inclusion terms
                    for term in include_terms:
                        if term not in port_name_lower:
                            included = False
                            break

                if included:
                    port_item.setHidden(False)
                    group_visible = True  # Make group visible if this port is visible
                else:
                    port_item.setHidden(True)

            # Set the visibility of the group item
            group_item.setHidden(not group_visible)

        # After filtering, we need to refresh the connection visualization
        # because hidden items might affect line drawing positions.
        # Call the method on the connection_manager instance
        self.connection_manager.refresh_visualizations()

    def _handle_filter_change(self) -> None:
        """Handles text changes in the shared filter boxes."""
        # Access tab_widget through ui_manager
        tab_widget = self.connection_manager.ui_manager.tab_widget
        current_index = tab_widget.currentIndex()
        current_tab_text = tab_widget.tabText(current_index)
        input_text = self.input_filter_edit.text()
        output_text = self.output_filter_edit.text()

        # Use tab name to determine which trees to filter (handles integrated mode)
        if current_tab_text == "Audio":
            self.filter_ports(self.input_tree, input_text)
            self.filter_ports(self.output_tree, output_text)
        elif current_tab_text == "MIDI":
            self.filter_ports(self.midi_input_tree, input_text)
            self.filter_ports(self.midi_output_tree, output_text)