Agentic Programming with Pandas

import pandas as pd
students = pd.read_csv("input/student_marks.csv")
subjects = ["Maths", "Science", "English"]
passed_mask = students[subjects].ge(40).all(axis=1)
passed = students.loc[passed_mask]
failed = students.loc[~passed_mask]
print("Passed:", len(passed), "Failed:", len(failed))

"""Program 1: read student marks and produce passed.csv + failed.csv.

Rule: a student passes only if EVERY subject mark is >= 40.
Invalid data is rejected rather than being treated as a failure.
"""
from pathlib import Path
import pandas as pd

BASE = Path(__file__).resolve().parent
SUBJECTS = ("Maths", "Science", "English")
COLUMNS = ("StudentID", "Name", "City", *SUBJECTS)


def split_students(source: pd.DataFrame):
    """Return independent passed/failed DataFrames, without mutating source."""
    missing = sorted(set(COLUMNS) - set(source.columns))
    if missing:
        raise ValueError("Missing columns: " + ", ".join(missing))
    df = source.loc[:, list(COLUMNS)].copy()
    for col in ("StudentID", "Name", "City"):
        df[col] = df[col].astype("string").str.strip()
        if df[col].isna().any() or df[col].eq("").any():
            raise ValueError(f"{col} cannot be missing or blank")
    if df["StudentID"].duplicated().any():
        raise ValueError("Duplicate StudentID: review the input file")
    for subject in SUBJECTS:
        df[subject] = pd.to_numeric(df[subject], errors="coerce")
        if df[subject].isna().any():
            raise ValueError(f"{subject} has missing/non-numeric marks")
        if not df[subject].between(0, 100).all():
            raise ValueError(f"{subject} must be between 0 and 100")
        if not df[subject].mod(1).eq(0).all():
            raise ValueError(f"{subject} marks must be whole numbers")
        df[subject] = df[subject].astype(int)

    passed_mask = df[list(SUBJECTS)].ge(40).all(axis=1)
    passed = df.loc[passed_mask].reset_index(drop=True)
    failed = df.loc[~passed_mask].reset_index(drop=True)
    assert len(passed) + len(failed) == len(df)
    return passed, failed


def main(base=None):
    base = Path(base) if base is not None else BASE
    input_file = base / "input" / "student_marks.csv"
    students = pd.read_csv(
        input_file, dtype={"StudentID": "string", "Name": "string", "City": "string"}
    )
    passed, failed = split_students(students)
    output = base / "output"
    output.mkdir(parents=True, exist_ok=True)
    passed.to_csv(output / "passed.csv", index=False)
    failed.to_csv(output / "failed.csv", index=False)
    print(f"Program 1: {len(students)} students → {len(passed)} passed, {len(failed)} failed")
    return passed, failed


if __name__ == "__main__":
    main()

"""Program 2: read passed.csv, calculate scores and publish meritlist.csv.

Tied totals receive the same competition rank (1, 2, 2, 4).
Names and student IDs resolve display order within a tied rank.
"""
from pathlib import Path
import pandas as pd

BASE = Path(__file__).resolve().parent
SUBJECTS = ("Maths", "Science", "English")


def build_meritlist(source: pd.DataFrame) -> pd.DataFrame:
    required = {"StudentID", "Name", *SUBJECTS}
    if missing := sorted(required - set(source.columns)):
        raise ValueError("passed.csv is missing: " + ", ".join(missing))
    df = source.copy()
    if df["StudentID"].isna().any() or df["StudentID"].duplicated().any():
        raise ValueError("passed.csv has invalid StudentID values")
    for subject in SUBJECTS:
        df[subject] = pd.to_numeric(df[subject], errors="coerce")
        if df[subject].isna().any() or not df[subject].between(40, 100).all():
            raise ValueError("passed.csv contains incomplete or failing marks")
    df["Total"] = df[list(SUBJECTS)].sum(axis=1)
    df["Percentage"] = (df["Total"] / len(SUBJECTS)).round(2)
    df["Rank"] = df["Total"].rank(ascending=False, method="min").astype(int)
    return df.sort_values(
        ["Total", "Name", "StudentID"],
        ascending=[False, True, True], kind="mergesort",
    ).reset_index(drop=True)


def main(base=None):
    base = Path(base) if base is not None else BASE
    passed = pd.read_csv(
        base / "output" / "passed.csv",
        dtype={"StudentID": "string"},
    )
    merit = build_meritlist(passed)
    merit.to_csv(base / "output" / "meritlist.csv", index=False)
    print(f"Program 2: {len(merit)} passed students → meritlist.csv")
    return merit


if __name__ == "__main__":
    main()

"""Program 3: read meritlist.csv and produce exactly ten rows at most.

If the 10th position is tied, we keep exactly 10 records and break
ties alphabetically by Name then StudentID (not all tied candidates).
"""
from pathlib import Path
import pandas as pd

BASE = Path(__file__).resolve().parent


def choose_top_ten(source: pd.DataFrame) -> pd.DataFrame:
    required = {"StudentID", "Name", "Total", "Rank"}
    if missing := sorted(required - set(source.columns)):
        raise ValueError("meritlist.csv is missing: " + ", ".join(missing))
    df = source.copy()
    if df["StudentID"].isna().any() or df["StudentID"].duplicated().any():
        raise ValueError("Invalid meritlist StudentID")
    df["Total"] = pd.to_numeric(df["Total"], errors="coerce")
    df["Rank"] = pd.to_numeric(df["Rank"], errors="coerce")
    if df["Total"].isna().any() or df["Rank"].isna().any():
        raise ValueError("Merit rank or total is missing or invalid")
    expected_ranks = df["Total"].rank(ascending=False, method="min")
    if not df["Rank"].eq(expected_ranks).all():
        raise ValueError("Merit ranks are inconsistent with total marks")
    return df.sort_values(
        ["Total", "Name", "StudentID"], ascending=[False, True, True],
        kind="mergesort"
    ).head(10).reset_index(drop=True)


def main(base=None):
    base = Path(base) if base is not None else BASE
    merit = pd.read_csv(
        base / "output" / "meritlist.csv", dtype={"StudentID": "string"}
    )
    top = choose_top_ten(merit)
    top.to_csv(base / "output" / "top_ten.csv", index=False)
    print(f"Program 3: top {len(top)} students → top_ten.csv")
    return top


if __name__ == "__main__":
    main()

"""Program 4: read failed.csv and create one CSV per failed subject.

A student failing two or three subjects intentionally appears in two
or three subject CSVs. Each CSV includes shortfall = 40 - subject mark.
"""
from pathlib import Path
import pandas as pd

BASE = Path(__file__).resolve().parent
SUBJECTS = ("Maths", "Science", "English")


def failures_by_subject(source: pd.DataFrame):
    missing = sorted(set(SUBJECTS) - set(source.columns))
    if missing:
        raise ValueError("failed.csv is missing: " + ", ".join(missing))
    df = source.copy()
    for subject in SUBJECTS:
        df[subject] = pd.to_numeric(df[subject], errors="coerce")
        if df[subject].isna().any() or not df[subject].between(0, 100).all():
            raise ValueError("failed.csv contains invalid marks")
    if not df[list(SUBJECTS)].lt(40).any(axis=1).all():
        raise ValueError("failed.csv contains a student who passed all subjects")
    result = {}
    for subject in SUBJECTS:
        part = df.loc[df[subject].lt(40)].copy()
        part["Shortfall"] = 40 - part[subject]
        part = part.sort_values(
            ["Shortfall", "Name", "StudentID"],
            ascending=[False, True, True], kind="mergesort",
        ).reset_index(drop=True)
        result[subject] = part
    return result


def main(base=None):
    base = Path(base) if base is not None else BASE
    failed = pd.read_csv(
        base / "output" / "failed.csv", dtype={"StudentID": "string"}
    )
    reports = failures_by_subject(failed)
    for subject, report in reports.items():
        name = f"failed-in{subject.lower()}.csv"
        report.to_csv(base / "output" / name, index=False)
        print(f"Program 4: {subject} → {name} ({len(report)} students)")
    return reports


if __name__ == "__main__":
    main()

from program1_split import main as split
from program2_merit import main as merit
from program3_top_ten import main as top_ten
from program4_subject_failures import main as failures

passed, failed = split()       # one input → two branches
meritlist = merit()             # passed.csv → meritlist.csv
top = top_ten()                 # meritlist.csv → top_ten.csv
subject_lists = failures()      # failed.csv → three CSVs
print("Pipeline complete")

python -m pip install pandas
cd python/pandas/projects/csv-pipelines/agentic-student-merit

# Run each program separately:
python program1_split.py
python program2_merit.py
python program3_top_ten.py
python program4_subject_failures.py

# Or let the coordinator run all four:
python run_all.py

# Verify the results:
python -m unittest discover -s . -p "test_*.py"