Source code for modalysis.client.dmr

"""HTTP client calls for DMR endpoints."""

import logging

import requests
from modalysis.constants import DEFAULT_BASE_URL

logger = logging.getLogger(__name__)




def dmr_format(
    input_path: str,
    output_path: str,
    output_name: str,
    allowed_chromosomes: list[str],
    min_score: float = 5,
    max_p_value: float = 0.05,
    min_pct_a_samples: float = 50.0,
    min_pct_b_samples: float = 50.0,
    min_reads: int = 5,
    base_url: str = DEFAULT_BASE_URL,
) -> dict[str, str]:
    """POST a DMR format request to the server."""
    url = f"{base_url}/dmr/format"
    payload = {
        "input_path": input_path,
        "output_path": output_path,
        "output_name": output_name,
        "allowed_chromosomes": allowed_chromosomes,
        "min_score": min_score,
        "max_p_value": max_p_value,
        "min_pct_a_samples": min_pct_a_samples,
        "min_pct_b_samples": min_pct_b_samples,
        "min_reads": min_reads,
    }
    logger.info("Client sending dmr format request to %s", url)
    response = requests.post(url, json=payload)
    response.raise_for_status()
    return response.json()





def dmr_annotate(
    dmr_path: str,
    gff_path: str,
    output_path: str,
    output_name: str,
    base_url: str = DEFAULT_BASE_URL,
) -> dict[str, str]:
    """POST a DMR annotate request to the server."""
    url = f"{base_url}/dmr/annotate"
    payload = {
        "dmr_path": dmr_path,
        "gff_path": gff_path,
        "output_path": output_path,
        "output_name": output_name,
    }
    logger.info("Client sending dmr annotate request to %s", url)
    response = requests.post(url, json=payload)
    response.raise_for_status()
    return response.json()





def dmr_gene_counts(
    annotated_dmr_paths: list[str],
    manifestations: list[str],
    modifications: list[str],
    manifestation_labels: list[str],
    expression_labels: list[str],
    annotated_gff_path: str,
    output_path: str,
    output_name: str,
    output_excel: bool = False,
    base_url: str = DEFAULT_BASE_URL,
) -> dict[str, str]:
    """POST a DMR gene-count aggregation request to the server."""
    url = f"{base_url}/dmr/gene-counts"
    payload = {
        "annotated_dmr_paths": annotated_dmr_paths,
        "manifestations": manifestations,
        "modifications": modifications,
        "manifestation_labels": manifestation_labels,
        "expression_labels": expression_labels,
        "annotated_gff_path": annotated_gff_path,
        "output_path": output_path,
        "output_name": output_name,
        "output_excel": output_excel,
    }
    logger.info("Client sending dmr gene-counts request to %s", url)
    response = requests.post(url, json=payload)
    response.raise_for_status()
    return response.json()





def dmr_common_genes(
    annotated_dmr_paths: list[str],
    manifestations: list[str],
    modifications: list[str],
    manifestation_labels: list[str],
    expression_labels: list[str],
    modification_a: str,
    modification_b: str,
    annotated_gff_path: str,
    output_path: str,
    output_name: str,
    base_url: str = DEFAULT_BASE_URL,
) -> dict[str, str]:
    """POST a DMR common-genes aggregation request to the server."""
    url = f"{base_url}/dmr/common-genes"
    payload = {
        "annotated_dmr_paths": annotated_dmr_paths,
        "manifestations": manifestations,
        "modifications": modifications,
        "manifestation_labels": manifestation_labels,
        "expression_labels": expression_labels,
        "modification_a": modification_a,
        "modification_b": modification_b,
        "annotated_gff_path": annotated_gff_path,
        "output_path": output_path,
        "output_name": output_name,
    }
    logger.info("Client sending dmr common-genes request to %s", url)
    response = requests.post(url, json=payload)
    response.raise_for_status()
    return response.json()