import logging import re import subprocess from typing import Any, Dict, List, Optional def _parse_pactl_output(output: str) -> List[Dict[str, Any]]: """ Parses pactl list output (e.g., sink-inputs or sinks) into a list of property dictionaries. Each item in the list represents a block (e.g., a "Sink Input" or "Sink"). """ items = [] current_item = None # Regex to capture the start of a new block (e.g., "Sink Input #0" or "Sink #0") # and also to capture key-value pairs within a block. # Handles cases where property values span multiple lines (though simple for now) block_start_pattern = re.compile(r"^(Sink Input #|Sink #)(\d+)", re.IGNORECASE) # Property pattern: Catches indented lines, assumes property key until '=', value after. # Value can contain spaces and various characters. Strips leading/trailing quotes from value. property_pattern = re.compile(r"^\s+([\w\.\-]+)\s*=\s*\"?(.*?)\"?$", re.IGNORECASE) # More specific property pattern for keys that are known and simple simple_property_pattern = re.compile(r"^\s+([\w\.\s-]+):\s*(.*)", re.IGNORECASE) for line in output.splitlines(): block_match = block_start_pattern.match(line) if block_match: if current_item is not None: items.append(current_item) item_type = "sink_input" if block_match.group(1).lower() == "sink input #" else "sink" current_item = {"_type": item_type, "_index": int(block_match.group(2))} continue if current_item is None: continue # Skip lines until a block starts prop_match = property_pattern.match(line) if prop_match: key = prop_match.group(1).strip().replace(" ", "_").lower() # Normalize key value = prop_match.group(2).strip() current_item[key] = value else: simple_prop_match = simple_property_pattern.match(line) if simple_prop_match: key = simple_prop_match.group(1).strip().replace(" ", "_").lower().replace(":", "") value = simple_prop_match.group(2).strip() current_item[key] = value if current_item is not None: items.append(current_item) return items def _get_linux_app_monitor_source(target_app_names: List[str], logger: logging.Logger) -> Optional[str]: """ Attempts to find a PulseAudio monitor source associated with a running application. This function is specific to Linux systems using PulseAudio (`pactl`). It tries to find a sink input linked to one of the `target_app_names` (by checking application.name, application.process.binary, or media.name) and then finds the monitor source associated with that sink input's sink. Args: target_app_names: A list of application names or binary names to search for (e.g., ['zoom', 'chrome', 'firefox']). Case-insensitive substring match. logger: A logger instance for logging messages. Returns: The name of the monitor source device if found, otherwise None. """ if not sys.platform.startswith('linux'): logger.debug("Not a Linux system, skipping pactl audio detection.") return None try: # 1. List sink inputs to find the one associated with the target application pactl_sink_inputs_cmd = ["pactl", "list", "short", "sink-inputs"] # Use short form for easier parsing # Example short output: 0 116 122 protocol-native.c float32le 2ch 48000Hz RUNNING # The full 'pactl list sink-inputs' is more reliable for properties. pactl_sink_inputs_cmd_full = ["pactl", "list", "sink-inputs"] process = subprocess.run(pactl_sink_inputs_cmd_full, capture_output=True, text=True, check=True) sink_inputs_output = process.stdout # logger.debug(f"pactl list sink-inputs output:\n{sink_inputs_output}") parsed_sink_inputs = _parse_pactl_output(sink_inputs_output) # logger.debug(f"Parsed sink inputs: {parsed_sink_inputs}") target_sink_index_str: Optional[str] = None for si in parsed_sink_inputs: if si.get("_type") != "sink_input": continue app_name = si.get("properties", {}).get("application.name", "").lower() app_binary = si.get("properties", {}).get("application.process.binary", "").lower() media_name = si.get("properties", {}).get("media.name", "").lower() # Often useful, e.g., "Playback Stream" # Check against each target name for target_name_part in target_app_names: target_lower = target_name_part.lower() if (target_lower in app_name or target_lower in app_binary or (media_name and target_lower in media_name)): # Media name can be None # Simpler parsing for properties block from full output # Look for "Sink: " sink_prop = si.get("sink") # From simple_property_pattern if it worked if isinstance(sink_prop, str) and sink_prop.isdigit(): target_sink_index_str = sink_prop logger.info(f"Found matching application sink input #{si.get('_index')} for '{target_name_part}' on sink index {target_sink_index_str} (App: {app_name}, Binary: {app_binary}, Media: {media_name})") break # Found a match if target_sink_index_str: break if not target_sink_index_str: logger.info(f"No running sink input found matching target applications: {target_app_names}") return None # 2. List sinks to find the monitor source for the identified sink pactl_sinks_cmd = ["pactl", "list", "sinks"] process = subprocess.run(pactl_sinks_cmd, capture_output=True, text=True, check=True) sinks_output = process.stdout # logger.debug(f"pactl list sinks output:\n{sinks_output}") parsed_sinks = _parse_pactl_output(sinks_output) # logger.debug(f"Parsed sinks: {parsed_sinks}") for sink_info in parsed_sinks: if sink_info.get("_type") == "sink" and str(sink_info.get("_index")) == target_sink_index_str: monitor_source_name = sink_info.get("monitor_source") if monitor_source_name: logger.info(f"Found monitor source for sink index {target_sink_index_str}: '{monitor_source_name}'") return monitor_source_name else: logger.warning(f"Sink index {target_sink_index_str} found, but it has no 'Monitor Source' property. Properties: {sink_info}") return None logger.warning(f"Could not find sink details for index {target_sink_index_str} from application.") return None except FileNotFoundError: logger.warning("`pactl` command not found. Cannot auto-detect Linux application audio source. Please ensure PulseAudio utilities are installed.") return None except subprocess.CalledProcessError as e: logger.error(f"Error executing `pactl` command: {e}. Output: {e.stderr}") return None except Exception as e: logger.error(f"An unexpected error occurred during Linux audio source detection: {e}", exc_info=True) return None if __name__ == '__main__': # Basic test (requires a running audio application, e.g., Chrome playing YouTube) logging.basicConfig(level=logging.DEBUG) test_logger = logging.getLogger("LinuxAudioTest") # Test with common browser names browser_apps = ['chrome', 'firefox', 'msedge', 'chromium', 'opera'] monitor = _get_linux_app_monitor_source(browser_apps, test_logger) if monitor: test_logger.info(f"Test successful: Found monitor source for browsers: {monitor}") else: test_logger.warning("Test: No monitor source found for browsers. Ensure a browser is playing audio.") # Test with a specific application name (if you have one running, e.g., 'Spotify') # spotify_monitor = _get_linux_app_monitor_source(['spotify'], test_logger) # if spotify_monitor: # test_logger.info(f"Test successful: Found monitor source for Spotify: {spotify_monitor}") # else: # test_logger.warning("Test: No monitor source found for Spotify. Ensure Spotify is playing audio.") # Test with a non-existent app non_existent_monitor = _get_linux_app_monitor_source(['nonexistentapp123'], test_logger) if not non_existent_monitor: test_logger.info("Test successful: Correctly found no monitor for non-existent app.") else: test_logger.error(f"Test failed: Found monitor for non-existent app: {non_existent_monitor}") [end of atomic-docker/project/functions/agents/_linux_audio_utils.py]