Audio Analysis¶

def analyze_file(
    self,
    file_path: str | Path,
    calibration_levels: dict[str, float] | list[float] | None = None,
    resample: int | None = None,
) -> pd.DataFrame:
    """
    Analyze a single audio file using the current configuration.

    Parameters
    ----------
    file_path
        Path to the audio file to analyze.
    calibration_levels
        Dictionary containing calibration levels for left and right channels.
    resample
        Sampling rate to resample the audio to before analysis.

    Returns
    -------
    :
        DataFrame containing the analysis results.

    """
    file_path = ensure_input_path(file_path)

    logger.info(f"Analyzing file: {file_path}")
    return load_analyse_binaural(
        file_path,
        calibration_levels,
        self.settings,
        resample=resample,
    )

@logger.catch
def analyze_folder(
    self,
    folder_path: str | Path,
    calibration_file: str | Path | None = None,
    max_workers: int | None = None,
    resample: int | None = None,
) -> pd.DataFrame:
    """
    Analyze all audio files in a folder using parallel processing.

    Parameters
    ----------
    folder_path
        Path to the folder containing audio files.
    calibration_file
        Path to a JSON file containing calibration levels for each audio file.
    max_workers
        Maximum number of worker processes to use.
        If None, it will use the number of CPU cores.
    resample
        Sampling rate to resample the audio to before analysis.

    Returns
    -------
    :
        DataFrame containing the analysis results for all files.

    """
    folder_path = ensure_input_path(folder_path)
    audio_files = list(folder_path.glob("*.wav"))

    logger.info(
        f"Analyzing folder: {folder_path.name} of {len(audio_files)}"
        f"files in parallel (max_workers={max_workers})"
    ) if max_workers else logger.info(
        f"Analyzing folder: {folder_path}, {len(audio_files)} files"
    )

    calibration_levels = {}
    if calibration_file:
        calibration_file = ensure_input_path(calibration_file)
        with calibration_file.open() as f:
            calibration_levels = json.load(f)
        logger.debug(f"Loaded calibration levels from: {calibration_file}")

    all_results = []
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for file in audio_files:
            future = executor.submit(
                load_analyse_binaural,
                file,
                calibration_levels,
                self.settings,
                resample,
                parallel_mosqito=False,
            )
            futures.append(future)

        for future in tqdm(
            as_completed(futures), total=len(futures), desc="Analyzing files"
        ):
            try:
                result = future.result()
                all_results.append(result)
            except Exception as e:  # noqa: BLE001
                logger.error(f"Error processing file: {e!s}")

    combined_results = pd.concat(all_results)
    logger.info(
        f"Completed analysis for {len(audio_files)} files in folder: {folder_path}"
    )
    return combined_results

def save_results(self, results: pd.DataFrame, output_path: str | Path) -> None:
    """
    Save analysis results to a file.

    Parameters
    ----------
    results
        DataFrame containing the analysis results.
    output_path
        Path to save the results file.

    """
    output_path = ensure_path_type(
        output_path
    )  # If doesn't already exist, pandas will create the file.
    if output_path.suffix == ".csv":
        results.to_csv(output_path)
    elif output_path.suffix == ".xlsx":
        results.to_excel(output_path)
    else:
        msg = "Unsupported file format. Use .csv or .xlsx"
        raise ValueError(msg)
    logger.info(f"Results saved to: {output_path}")

def update_config(self, new_config: dict) -> "AudioAnalysis":
    """
    Update the current configuration.

    Parameters
    ----------
    new_config
        Dictionary containing the new configuration settings.

    """
    self.settings = self.config_manager.merge_configs(new_config)
    logger.info("Configuration updated")
    return self

def save_config(self, config_path: str | Path) -> None:
    """
    Save the current configuration to a file.

    Parameters
    ----------
    config_path
        Path to save the configuration file.

    """
    self.config_manager.save_config(config_path)
    logger.info(f"Configuration saved to: {config_path}")