dpacman/data_tasks/download/remap.py

from omegaconf import DictConfig
from pathlib import Path
import logging
import subprocess
import os

import rootutils

root = rootutils.setup_root(__file__, indicator=".project-root", pythonpath=True)

logger = logging.getLogger(__name__)


def recap(result):
    """
    Print out info about the success of the download job.
    """
    logger.info("STDOUT:\n" + result.stdout)
    if result.stderr:
        logger.warning("STDERR:\n" + result.stderr)

    if result.returncode != 0:
        logger.error(f"Download script exited with code {result.returncode}")
    else:
        logger.info("Download completed successfully.")


def main(cfg: DictConfig):
    """
    Download IntAct file, which is one zip file
    """

    script_path = root / "dpacman/data_tasks/download/download_unzip.sh"
    delete_zip = str(cfg.data_task.get("delete_zip", False)).lower()
    assert delete_zip in ("true", "false")

    nr_url = cfg.data_task.nr_url
    nr_output_dir = root / cfg.data_task.nr_output_dir
    nr_filename = cfg.data_task.nr_filename

    logger.info(f"Running {cfg.data_task.type} for {cfg.data_task.name}")

    ##### Non-redundant peaks download #####
    logger.info(f"Script: {script_path}")
    logger.info(f"Non-Redundant Peaks - URL: {nr_url}")
    logger.info(f"Non-Redundant Peaks - Output: {nr_output_dir / nr_filename}")

    os.makedirs(nr_output_dir, exist_ok=True)

    # Run the download.sh script as a subproces
    result = subprocess.run(
        ["bash", str(script_path), nr_url, str(nr_output_dir), nr_filename, delete_zip],
        capture_output=True,
        text=True,
    )

    recap(result)

    ##### CRMs download #####
    crm_url = cfg.data_task.crm_url
    crm_output_dir = root / cfg.data_task.crm_output_dir
    crm_filename = cfg.data_task.crm_filename

    logger.info(f"CRMs - URL: {crm_url}")
    logger.info(f"CRMs - Output: {crm_output_dir / crm_filename}")

    os.makedirs(crm_output_dir, exist_ok=True)

    # Run the download.sh script as a subproces
    result = subprocess.run(
        [
            "bash",
            str(script_path),
            crm_url,
            str(crm_output_dir),
            crm_filename,
            delete_zip,
        ],
        capture_output=True,
        text=True,
    )

    recap(result)


if __name__ == "__main__":
    main()