Source code for modalysis.server.dmr

"""FastAPI routes for DMR operations."""

import logging

from fastapi import APIRouter

from modalysis.core import dmr as core_dmr
from modalysis.server.models import (
    DmrAnnotateRequest,
    DmrCommonGenesRequest,
    DmrFormatRequest,
    DmrGeneCountsRequest,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/dmr")




@router.post("/format")
def dmr_format(request: DmrFormatRequest) -> dict[str, str]:
    """Handle POST `/dmr/format` and run core DMR formatting."""
    logger.info("Server received dmr format request: %s", request)
    core_dmr.format(
        input_path=request.input_path,
        output_path=request.output_path,
        output_name=request.output_name,
        allowed_chromosomes=request.allowed_chromosomes,
        min_score=request.min_score,
        max_p_value=request.max_p_value,
        min_pct_a_samples=request.min_pct_a_samples,
        min_pct_b_samples=request.min_pct_b_samples,
        min_reads=request.min_reads,
    )
    return {"status": "success"}





@router.post("/annotate")
def dmr_annotate(request: DmrAnnotateRequest) -> dict[str, str]:
    """Handle POST `/dmr/annotate` and run core DMR interval annotation."""
    logger.info("Server received dmr annotate request: %s", request)
    core_dmr.annotate(
        dmr_path=request.dmr_path,
        gff_path=request.gff_path,
        output_path=request.output_path,
        output_name=request.output_name,
    )
    return {"status": "success"}





@router.post("/gene-counts")
def dmr_gene_counts(request: DmrGeneCountsRequest) -> dict[str, str]:
    """Handle POST `/dmr/gene-counts` and run gene-count aggregation."""
    logger.info("Server received dmr gene-counts request: %s", request)
    core_dmr.gene_counts(
        annotated_dmr_paths=request.annotated_dmr_paths,
        manifestations=request.manifestations,
        modifications=request.modifications,
        manifestation_labels=request.manifestation_labels,
        expression_labels=request.expression_labels,
        annotated_gff_path=request.annotated_gff_path,
        output_path=request.output_path,
        output_name=request.output_name,
        output_excel=request.output_excel,
    )
    return {"status": "success"}





@router.post("/common-genes")
def dmr_common_genes(request: DmrCommonGenesRequest) -> dict[str, str]:
    """Handle POST `/dmr/common-genes` and run overlap aggregation."""
    logger.info("Server received dmr common-genes request: %s", request)
    core_dmr.common_genes(
        annotated_dmr_paths=request.annotated_dmr_paths,
        manifestations=request.manifestations,
        modifications=request.modifications,
        manifestation_labels=request.manifestation_labels,
        expression_labels=request.expression_labels,
        modification_a=request.modification_a,
        modification_b=request.modification_b,
        annotated_gff_path=request.annotated_gff_path,
        output_path=request.output_path,
        output_name=request.output_name,
    )
    return {"status": "success"}