pipeline_philly/verify/connectors/source_file.py

"""Source-layer re-readers: pull values directly from the ingested files.

The ingested-file paths come from data/cities/philly/sources/<source_id>/<date>/.
For SDP PSSA we read inside the Zip (Acct or Actual variant).
For Future Ready we read the relevant xlsx sheet.
For SDP CSV sources (grad, attendance, etc.) we read the CSV.

The harness compares (school_code, year, subgroup) → value cells between
served DB and source. Each function returns {(key, year): {value, n, suppressed}}.
"""
from __future__ import annotations

import io
import zipfile
import pathlib

import pandas as pd

from pipeline_philly.verify import common


# Subgroup string maps for SDP CSV files
SDP_GROUP_MAP = {
    "All Students": "ALL",
    "American Indian/Alaskan Native": "AMER_INDIAN_AK_NATIVE",
    "Asian": "ASIAN",
    "Native Hawaiian/Pacific Islander": "HAWAIIAN_PAC_ISL",
    "Black/African American": "BLACK",
    "Hispanic/Latino": "HISPANIC",
    "White": "WHITE",
    "Multi Racial/Other": "TWO_OR_MORE_RACES",
    "Economically Disadvantaged": "ECON_DISADV",
    "Not Economically Disadvantaged": "NOT_ECON_DISADV",
    "EL": "ELL",
    "Non-EL": "NOT_ELL",
    "Has IEP": "IEP",
    "Does Not Have IEP": "NOT_IEP",
    "Female": "FEMALE",
    "Male": "MALE",
    "Non-Binary": "NON_BINARY",
}

# Future Ready subgroup column suffix → enum
FRPI_GROUP_MAP = {
    "AllStudent": "ALL",
    "AmericanIndianAlaskaNative": "AMER_INDIAN_AK_NATIVE",
    "Asian": "ASIAN",
    "HawaiianPacificIslander": "HAWAIIAN_PAC_ISL",
    "Black": "BLACK",
    "Hispanic": "HISPANIC",
    "White": "WHITE",
    "2orMoreRaces": "TWO_OR_MORE_RACES",
    "EconomicallyDisadvantaged": "ECON_DISADV",
    "EnglishLearner": "ELL",
    "StudentswithDisabilities": "IEP",
    "CombinedEthnicity": "COMBINED_ETHNICITY",
}


def _to_number(v) -> float | None:
    if v is None:
        return None
    s = str(v).strip()
    if s in {"", "*", "S", "IS", "n.a.", "i.s."}:
        return None
    try:
        return float(s)
    except ValueError:
        return None


def read_sdp_pssa_zip_schools(
    cut: str,
    year: str = "2024-25",
    testname: str = "PSSA",
    subject: str = "ELA",
    grade: str = "Grades 3-8",
    subgroup: str = "ALL",
) -> dict[tuple[str, str], dict]:
    """Read the SDP PSSA Zip and return one cell per (id_eos == ulcs, year).

    Filters server-side by testname/subject/grade/subgroup so the caller can
    reconcile one metric at a time. cut="acct" or "actual" picks the variant.
    """
    source_id = "sdp_pssa_keystone_2024-25"
    zip_path = common.latest_source_path(source_id)
    csv_name = f"PSSA_Keystone_Schools_{'Acct' if cut == 'acct' else 'Actual'}_Suppressed.csv"

    with zipfile.ZipFile(zip_path) as z:
        with z.open(csv_name) as f:
            df = pd.read_csv(f, low_memory=False)

    # invert the group map so we can filter
    pretty = next((k for k, v in SDP_GROUP_MAP.items() if v == subgroup), None)
    if pretty is None:
        raise SystemExit(f"unknown subgroup: {subgroup}")

    mask = (
        (df["testname"].astype(str).str.upper() == testname.upper())
        & (df["subject"].astype(str).str.upper() == subject.upper())
        & (df["grade"].astype(str) == grade)
        & (df["group"].astype(str) == pretty)
    )
    out: dict[tuple[str, str], dict] = {}
    for _, r in df[mask].iterrows():
        ulcs = str(r["id_eos"]).strip()
        v = _to_number(r["profadv_score"])
        n = _to_number(r["denom"])
        out[(ulcs, year)] = {
            "value": v,
            "n": n,
            "suppressed": v is None,
        }
    return out


def read_frpi_performance_school_wide(
    metric_column_prefix: str,
    subgroup: str = "ALL",
    sheet: str = "State Assessment Measures",
    year: str = "2024-25",
) -> dict[tuple[str, str], dict]:
    """Read Future Ready Performance file. Returns {(aun-paCode, year): value}.

    `metric_column_prefix` is the bare measure name before the underscore-
    separated subgroup suffix (e.g. "PercentProficientorAdvancedonELALiterature").
    """
    source_id = "futurereadypa_performance_2024-25"
    path = common.latest_source_path(source_id)
    df = pd.read_excel(path, sheet_name=sheet)

    sg_suffix = next((k for k, v in FRPI_GROUP_MAP.items() if v == subgroup), None)
    if sg_suffix is None:
        raise SystemExit(f"unknown subgroup: {subgroup}")
    col = f"{metric_column_prefix}_{sg_suffix}"
    if col not in df.columns:
        raise SystemExit(f"missing column: {col}")

    out: dict[tuple[str, str], dict] = {}
    for _, r in df.iterrows():
        aun = str(int(r["AUN"])).strip() if pd.notna(r["AUN"]) else ""
        schl = str(int(r["Schl"])).strip() if pd.notna(r["Schl"]) else ""
        if not aun or not schl:
            continue
        v = _to_number(r[col])
        out[(f"{aun}-{schl}", year)] = {
            "value": v,
            "n": None,
            "suppressed": v is None,
        }
    return out


def read_sdp_csv_long(
    source_id: str,
    metric_value_col: str,
    *,
    school_col: str = "ULCS Code",
    year_col: str = "School Year",
    group_col: str = "Group",
    subgroup: str = "ALL",
    extra_filter: dict[str, str] | None = None,
) -> dict[tuple[str, str], dict]:
    """Generic long-format CSV reader for SDP outcome files."""
    path = common.latest_source_path(source_id)
    df = pd.read_csv(path, low_memory=False)

    pretty = next((k for k, v in SDP_GROUP_MAP.items() if v == subgroup), None)
    if pretty is None:
        raise SystemExit(f"unknown subgroup: {subgroup}")

    mask = df[group_col].astype(str) == pretty
    if extra_filter:
        for k, v in extra_filter.items():
            mask &= df[k].astype(str) == v

    out: dict[tuple[str, str], dict] = {}
    for _, r in df[mask].iterrows():
        ulcs = str(r[school_col]).strip()
        year_raw = str(r[year_col]).strip()
        # "2020-2021" → "2020-21"
        if len(year_raw) == 9 and year_raw[4] == "-":
            year_can = f"{year_raw[:4]}-{year_raw[7:]}"
        else:
            year_can = year_raw
        v = _to_number(r[metric_value_col])
        out[(ulcs, year_can)] = {
            "value": v,
            "n": None,
            "suppressed": v is None,
        }
    return out