Reporte de placa

[project]
name = "reporte"
version = "0.1.0"
description = "Proyecto 2: reporte de placa qPCR a partir del export crudo"
requires-python = ">=3.12"
dependencies = [
    "pandas>=2.2",
    "openpyxl>=3.1",
    "matplotlib>=3.9",
    "typer>=0.12",
]

[project.scripts]
reporte = "reporte.cli:app"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/reporte"]

[tool.pytest.ini_options]
testpaths = ["tests"]

[tool.ruff]
line-length = 100

Well, Sample ,Target,Ct
A1,Control,GAPDH,18.02
A2,Control,GAPDH,18.10
A3,Control,IL6,24.51
A4,Control,IL6,24.47
B1,Tratada ,GAPDH,18.20
B2,Tratada,GAPDH,18.12
B3,Tratada,IL6,22.03
B4,Tratada,IL6,Undetermined

"""Errores propios del reporte. Un nombre por causa: el mensaje dice qué arreglar."""


class ReporteError(Exception):
    """Base de todos los errores del reporte."""


class EsquemaInvalido(ReporteError):
    """El archivo no tiene las columnas que el reporte necesita."""


class ReferenciaNoEncontrada(ReporteError):
    """El gen de referencia no aparece en la placa (o falta en alguna muestra)."""


class ControlNoEncontrado(ReporteError):
    """La muestra control no aparece en la placa."""

from pathlib import Path

import pytest

from reporte.errors import EsquemaInvalido
from reporte.load import cargar_placa, resumen_carga

DATA = Path(__file__).parent / "data" / "placa_min.csv"


def test_carga_limpia_espacios_y_tipos():
    df = cargar_placa(DATA)
    assert list(df.columns) == ["Well", "Sample", "Target", "Ct"]
    assert set(df["Sample"]) == {"Control", "Tratada"}  # " Tratada " quedó limpio
    assert df["Ct"].dtype == "float64"


def test_undetermined_queda_como_nan_no_como_cero():
    df = cargar_placa(DATA)
    fila = df[(df["Sample"] == "Tratada") & (df["Well"] == "B4")]
    assert fila["Ct"].isna().all()


def test_columna_faltante_nombra_la_columna(tmp_path):
    malo = tmp_path / "malo.csv"
    malo.write_text("Well,Sample,Ct\nA1,Control,18.0\n")
    with pytest.raises(EsquemaInvalido, match="Target"):
        cargar_placa(malo)


def test_archivo_inexistente():
    with pytest.raises(FileNotFoundError):
        cargar_placa(Path("no-existe.csv"))


def test_resumen_carga():
    r = resumen_carga(cargar_placa(DATA))
    assert r == {"pocillos": 8, "sin_ct": 1, "muestras": 2, "targets": 2}


def test_no_modifica_el_crudo(tmp_path):
    copia = tmp_path / "placa.csv"
    copia.write_bytes(DATA.read_bytes())
    antes = copia.read_bytes()
    cargar_placa(copia)
    assert copia.read_bytes() == antes

"""Capa 1: leer el archivo crudo y validarlo. Aquí termina la desconfianza.

Todo lo que sale de `cargar_placa` cumple el esquema; las otras capas ya no validan.
"""

from pathlib import Path

import pandas as pd

from reporte.errors import EsquemaInvalido

COLUMNAS = ["Well", "Sample", "Target", "Ct"]
SIN_CT = {"undetermined", "undet", "n/a", "na", ""}


def cargar_placa(ruta: Path) -> pd.DataFrame:
    """Lee un export de placa qPCR (CSV o Excel) y devuelve un DataFrame limpio.

    - Columnas exactas: Well, Sample, Target, Ct (ignora mayúsculas y espacios en el nombre).
    - Sample y Target sin espacios sobrantes.
    - Ct numérico; "Undetermined" y vacíos quedan como NaN (no se inventan valores).
    Lanza EsquemaInvalido si falta una columna, nombrándola.
    """
    ruta = Path(ruta)
    if not ruta.exists():
        raise FileNotFoundError(f"no existe el archivo: {ruta}")
    if ruta.suffix.lower() in {".xlsx", ".xls"}:
        crudo = pd.read_excel(ruta)
    else:
        crudo = pd.read_csv(ruta)

    # Normaliza nombres de columna: " sample " -> "Sample"
    crudo = crudo.rename(columns={c: str(c).strip().title() for c in crudo.columns})
    faltan = [c for c in COLUMNAS if c not in crudo.columns]
    if faltan:
        raise EsquemaInvalido(f"faltan columnas {faltan}; el archivo tiene {list(crudo.columns)}")

    df = crudo[COLUMNAS].copy()  # copia: el crudo no se toca
    df["Sample"] = df["Sample"].astype(str).str.strip()
    df["Target"] = df["Target"].astype(str).str.strip()
    ct = df["Ct"].astype(str).str.strip()
    ct = ct.where(~ct.str.lower().isin(SIN_CT), other=None)
    df["Ct"] = pd.to_numeric(ct, errors="raise")
    return df


def resumen_carga(df: pd.DataFrame) -> dict[str, int]:
    """Números que conviene mirar antes de calcular nada."""
    return {
        "pocillos": int(len(df)),
        "sin_ct": int(df["Ct"].isna().sum()),
        "muestras": int(df["Sample"].nunique()),
        "targets": int(df["Target"].nunique()),
    }

import pandas as pd
import pytest

from reporte.errors import ControlNoEncontrado, ReferenciaNoEncontrada
from reporte.transform import analizar, delta_ct, promedio_por_replica


def placa() -> pd.DataFrame:
    # 2 muestras x 2 targets x 2 réplicas, números fáciles de verificar a mano
    return pd.DataFrame(
        {
            "Well": ["A1", "A2", "A3", "A4", "B1", "B2", "B3", "B4"],
            "Sample": ["Control"] * 4 + ["Tratada"] * 4,
            "Target": ["GAPDH", "GAPDH", "IL6", "IL6"] * 2,
            "Ct": [18.0, 18.0, 24.0, 24.0, 18.0, 18.0, 22.0, 22.0],
        }
    )


def test_promedio_ignora_nan_y_cuenta_n():
    df = placa()
    df.loc[7, "Ct"] = None
    prom = promedio_por_replica(df)
    fila = prom[(prom["Sample"] == "Tratada") & (prom["Target"] == "IL6")].iloc[0]
    assert fila["ct_mean"] == 22.0
    assert fila["n"] == 1


def test_delta_ct_calculado_a_mano():
    dct = delta_ct(promedio_por_replica(placa()), ref="GAPDH")
    control = dct[dct["Sample"] == "Control"].iloc[0]
    assert control["dct"] == pytest.approx(6.0)  # 24 - 18


def test_ddct_y_fold_a_mano():
    res = analizar(placa(), ref="GAPDH", control="Control")
    tratada = res[res["Sample"] == "Tratada"].iloc[0]
    assert tratada["ddct"] == pytest.approx(-2.0)  # (22-18) - (24-18)
    assert tratada["fold"] == pytest.approx(4.0)  # 2^2


def test_referencia_ausente_lanza_error_claro():
    with pytest.raises(ReferenciaNoEncontrada, match="ACTB"):
        delta_ct(promedio_por_replica(placa()), ref="ACTB")


def test_control_ausente_lanza_error_claro():
    with pytest.raises(ControlNoEncontrado, match="Mock"):
        analizar(placa(), ref="GAPDH", control="Mock")

"""Capa 2: cálculo. Funciones puras: entra un DataFrame, sale otro. Nada de archivos."""

import pandas as pd

from reporte.errors import ControlNoEncontrado, ReferenciaNoEncontrada


def promedio_por_replica(df: pd.DataFrame) -> pd.DataFrame:
    """Promedio y desviación estándar de Ct por (Sample, Target).

    Los pocillos sin Ct (NaN) no entran al promedio; `n` dice cuántos sí entraron.
    """
    agrupado = df.groupby(["Sample", "Target"], sort=True)["Ct"]
    return agrupado.agg(ct_mean="mean", ct_sd="std", n="count").reset_index()


def delta_ct(prom: pd.DataFrame, ref: str = "GAPDH") -> pd.DataFrame:
    """ΔCt = Ct(target) - Ct(ref) dentro de cada muestra. Excluye el propio gen de referencia."""
    if ref not in set(prom["Target"]):
        raise ReferenciaNoEncontrada(
            f"el gen de referencia {ref!r} no está en la placa; "
            f"targets: {sorted(prom['Target'].unique())}"
        )
    referencia = prom.loc[prom["Target"] == ref, ["Sample", "ct_mean"]].rename(
        columns={"ct_mean": "ct_ref"}
    )
    sin_ref = sorted(set(prom["Sample"]) - set(referencia["Sample"]))
    if sin_ref:
        raise ReferenciaNoEncontrada(f"muestras sin {ref!r}: {sin_ref}")
    otros = prom[prom["Target"] != ref]
    res = otros.merge(referencia, on="Sample", how="left")
    res["dct"] = res["ct_mean"] - res["ct_ref"]
    return res


def delta_delta_ct(dct: pd.DataFrame, control: str = "Control") -> pd.DataFrame:
    """ΔΔCt = ΔCt(muestra) - ΔCt(control) por target, y fold = 2^-ΔΔCt."""
    if control not in set(dct["Sample"]):
        raise ControlNoEncontrado(
            f"la muestra control {control!r} no está en la placa; "
            f"muestras: {sorted(dct['Sample'].unique())}"
        )
    base = dct.loc[dct["Sample"] == control, ["Target", "dct"]].rename(
        columns={"dct": "dct_control"}
    )
    res = dct.merge(base, on="Target", how="left")
    res["ddct"] = res["dct"] - res["dct_control"]
    res["fold"] = 2.0 ** (-res["ddct"])
    return res


def analizar(df: pd.DataFrame, ref: str = "GAPDH", control: str = "Control") -> pd.DataFrame:
    """Las tres etapas en orden. Es la única función que la CLI necesita llamar."""
    return delta_delta_ct(delta_ct(promedio_por_replica(df), ref=ref), control=control)

"""Capa 3: escribir resultados. Recibe DataFrames ya calculados; no calcula nada."""

from pathlib import Path

import matplotlib
import pandas as pd

matplotlib.use("Agg")  # sin ventana: solo archivos
import matplotlib.pyplot as plt  # noqa: E402

COLUMNAS_SALIDA = ["Sample", "Target", "n", "ct_mean", "ct_sd", "dct", "ddct", "fold"]


def _fmt(x: float) -> str:
    return "—" if pd.isna(x) else f"{x:.2f}"


def escribir_markdown(res: pd.DataFrame, ruta: Path, resumen: dict[str, int]) -> Path:
    lineas = ["# Reporte de placa qPCR", ""]
    lineas.append(
        f"Pocillos: {resumen['pocillos']} · sin Ct: {resumen['sin_ct']} · "
        f"muestras: {resumen['muestras']} · targets: {resumen['targets']}"
    )
    lineas += [
        "",
        "| Muestra | Target | n | Ct medio | SD | ΔCt | ΔΔCt | Fold (2^-ΔΔCt) |",
        "|---|---|---|---|---|---|---|---|",
    ]
    for _, r in res.sort_values(["Target", "Sample"]).iterrows():
        lineas.append(
            f"| {r.Sample} | {r.Target} | {int(r.n)} | {_fmt(r.ct_mean)} | {_fmt(r.ct_sd)} | "
            f"{_fmt(r.dct)} | {_fmt(r.ddct)} | {_fmt(r.fold)} |"
        )
    ruta.write_text("\n".join(lineas) + "\n", encoding="utf-8")
    return ruta


def escribir_xlsx(res: pd.DataFrame, ruta: Path) -> Path:
    res[COLUMNAS_SALIDA].to_excel(ruta, index=False, sheet_name="resultados")
    return ruta


def graficar_fold(res: pd.DataFrame, ruta: Path) -> Path:
    tabla = res.pivot(index="Sample", columns="Target", values="fold")
    ax = tabla.plot(kind="bar", figsize=(8, 4.5), width=0.8)
    ax.axhline(1.0, color="gray", linewidth=0.8, linestyle="--")
    ax.set_ylabel("Expresión relativa (2^-ΔΔCt)")
    ax.set_xlabel("")
    ax.set_title("Expresión relativa por muestra y target")
    plt.xticks(rotation=0)
    plt.tight_layout()
    plt.savefig(ruta, dpi=150)
    plt.close()
    return ruta


def generar_reporte(res: pd.DataFrame, resumen: dict[str, int], carpeta: Path) -> list[Path]:
    carpeta = Path(carpeta)
    carpeta.mkdir(parents=True, exist_ok=True)
    return [
        escribir_markdown(res, carpeta / "reporte.md", resumen),
        escribir_xlsx(res, carpeta / "resultados.xlsx"),
        graficar_fold(res, carpeta / "fold.png"),
    ]

"""Entrada por línea de comandos. Delgada: parsea argumentos, llama a las capas, informa."""

from pathlib import Path
from typing import Annotated

import typer

from reporte import load, report, transform
from reporte.errors import ReporteError

app = typer.Typer(help="Reporte ΔΔCt a partir del export de una placa qPCR.", no_args_is_help=True)


@app.command()
def generar(
    archivo: Annotated[Path, typer.Argument(help="CSV o xlsx exportado del equipo")],
    ref: Annotated[str, typer.Option(help="Gen de referencia")] = "GAPDH",
    control: Annotated[str, typer.Option(help="Muestra control (calibrador)")] = "Control",
    out: Annotated[Path, typer.Option(help="Carpeta de salida")] = Path("salida"),
) -> None:
    """Lee la placa, calcula ΔCt/ΔΔCt y escribe reporte.md, resultados.xlsx y fold.png."""
    try:
        df = load.cargar_placa(archivo)
        resumen = load.resumen_carga(df)
        res = transform.analizar(df, ref=ref, control=control)
        rutas = report.generar_reporte(res, resumen, out)
    except (ReporteError, FileNotFoundError) as e:
        typer.secho(f"error: {e}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=1) from e
    typer.echo(
        f"{resumen['pocillos']} pocillos, {resumen['sin_ct']} sin Ct, "
        f"{resumen['muestras']} muestras, {resumen['targets']} targets"
    )
    for r in rutas:
        typer.echo(f"  escrito {r}")


if __name__ == "__main__":
    app()

from pathlib import Path

from typer.testing import CliRunner

from reporte.cli import app

DATA = Path(__file__).parent / "data" / "placa_min.csv"
runner = CliRunner()


def test_generar_escribe_los_tres_archivos(tmp_path):
    out = tmp_path / "rep"
    r = runner.invoke(app, [str(DATA), "--ref", "GAPDH", "--control", "Control", "--out", str(out)])
    assert r.exit_code == 0, r.output
    assert (out / "reporte.md").exists()
    assert (out / "resultados.xlsx").exists()
    assert (out / "fold.png").exists()


def test_referencia_mala_sale_con_codigo_1(tmp_path):
    r = runner.invoke(app, [str(DATA), "--ref", "ACTB", "--out", str(tmp_path)])
    assert r.exit_code == 1
    assert "ACTB" in r.output

name: ci
on:
  pull_request:
  push:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
        with:
          python-version: "3.12"
      - run: uv sync
      - run: uv run ruff check .
      - run: uv run ruff format --check .
      - run: uv run pytest

# Imagen base: Python 3.12 mínimo, sobre Debian
FROM python:3.12-slim

# Copiamos el binario de uv desde su imagen oficial
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

WORKDIR /app

# Primero las dependencias (cambian poco → la capa se cachea)
COPY pyproject.toml uv.lock ./
# Luego el código y los tests
COPY src ./src
COPY tests ./tests

# Instala exactamente lo que dice uv.lock, nada más
RUN uv sync --frozen

# El reporte se genera sobre /datos, que se monta al correr
ENTRYPOINT ["uv", "run", "reporte"]
CMD ["--help"]