The Cost of Virtuosity

import os
import requests
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

API = "https://api.data.gov/ed/collegescorecard/v1/schools"
API_KEY = os.getenv("SCORECARD_API_KEY")

# DOE Unit Identification Numbers
UNITIDS = [164748, 192110, 167057, 192712, 211893]
YEARS = range(2012, 2022 + 1)

NAME_MAP = {
    164748: "Berklee College of Music",
    192110: "The Juilliard School",
    167057: "New England Conservatory",
    192712: "Manhattan School of Music",
    211893: "Curtis Institute of Music",
}

FIELD_MAP = {
    "enrollment_total": "student.size",
    "admission_rate": "admissions.admission_rate.overall",
    "retention_rate_ft": "student.retention_rate.four_year.full_time",
    "grad_rate_150": "completion.completion_rate_4yr_150nt",

    # Published tuition + required fees only (no room/board) 
    "tuition_fees": "cost.tuition.in_state",

    # What families actually pay:
    # (Tuition + Room/Board + Books + Personal) - (Federal + State + Institutional grants)
    "avg_net_price": "cost.avg_net_price.private",
}


# Convert decimal values (0-1) to percentages (0-100) and round to 2 decimal places
# If values are already in percentage format (>1 or <0), just round them
def normalize_percentages(df: pd.DataFrame, percentage_fields=None) -> pd.DataFrame:
    if percentage_fields is None:
        percentage_fields = ["admission_rate", "retention_rate_ft", "grad_rate_150"]

    # c = column
    # s = series
    out = df.copy()
    for c in percentage_fields:
        if c in out.columns:
            s = pd.to_numeric(out[c], errors="coerce")
            if s.notna().any():
                valid = s[s.notna()]
                if len(valid) > 0 and valid.max() <= 1.0 and valid.min() >= 0:
                    # Convert decimals to percentages
                    out[c] = (s * 100).round(2)
                else:
                    # Already in percentage format, just round
                    out[c] = s.round(2)
    return out


# Build a filename with year range and timestamp
# Accepts optional datetime for testing
def build_filename(years, now=None) -> str:
    if now is None:
        now = datetime.now()
    year_range = f"{min(years)}_{max(years)}"
    timestamp = now.strftime("%Y%m%d")
    return f"music_school_data_{year_range}_{timestamp}.csv"


def fetch_year(institution_ids, year):
    fields = ["id", "school.name"]
    fields += [f"{year}.{f}" for f in FIELD_MAP.values()]
    params = {
        "api_key": API_KEY,
        "id__in": ",".join(map(str, institution_ids)),
        "fields": ",".join(fields),
        "per_page": 100,
    }
    
    # res = response
    # ex = exception
    try:
        res = requests.get(API, params=params, timeout=30)
        res.raise_for_status()
        js = res.json()
    except requests.exceptions.RequestException as ex:
        print(f"Error fetching data for year {year}: {ex}")
        return pd.DataFrame()
    except ValueError as ex:
        print(f"Error parsing JSON response for year {year}: {ex}")
        return pd.DataFrame()
    
    results = js.get("results", [])
    if not results:
        print(f"Warning: No data returned for year {year}")
        return pd.DataFrame()
    
    rows = []
    for item in results:
        row_id = item.get("id")
        if not row_id:
            continue
            
        row = {
            "institution": NAME_MAP.get(row_id, item.get("school.name", "Unknown")),
            "unitid": row_id,
            "year": year,
        }

        for metric, field_suffix in FIELD_MAP.items():
            row[metric] = item.get(f"{year}.{field_suffix}")
        rows.append(row)
    return pd.DataFrame(rows)


def main():
    frames = []
    failed_years = []
    
    for y in YEARS:
        df = fetch_year(UNITIDS, y)
        if not df.empty:
            frames.append(df)
        else:
            failed_years.append(y)
    
    if not frames:
        raise SystemExit("Error: No data retrieved for any years. Check API key and network connection.")
    
    if failed_years:
        print(f"Warning: Failed to retrieve data for years: {failed_years}")
    
    long_df = pd.concat(frames, ignore_index=True)
    long_df = normalize_percentages(long_df)

    sorted_df = long_df.sort_values(['institution', 'year']).reset_index(drop=True)
    filename = build_filename(YEARS)
    
    sorted_df.to_csv(filename, index=False)
    print(f"Saved data: {filename} ({len(sorted_df)} rows, {len(sorted_df['institution'].unique())} institutions)")
    return sorted_df

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nOperation cancelled by user.")
    except Exception as ex:
        print("Error: Operation failed. Check your API key and network connection.")
        raise SystemExit(1)
                        

import os
import sqlite3
import requests
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

API = "https://api.data.gov/ed/collegescorecard/v1/schools"
API_KEY = os.getenv("SCORECARD_API_KEY")

# DOE Unit Identification Numbers
UNITIDS = [164748, 192110, 167057, 192712, 211893]
YEARS = range(2012, 2022 + 1)

NAME_MAP = {
    164748: "Berklee College of Music",
    192110: "The Juilliard School",
    167057: "New England Conservatory",
    192712: "Manhattan School of Music",
    211893: "Curtis Institute of Music",
}

FIELD_MAP = {
    "enrollment_total": "student.size",
    "admission_rate": "admissions.admission_rate.overall",
    "retention_rate_ft": "student.retention_rate.four_year.full_time",
    "grad_rate_150": "completion.completion_rate_4yr_150nt",

    # Published tuition + required fees only (no room/board) 
    "tuition_fees": "cost.tuition.in_state",

    # What families actually pay:
    # (Tuition + Room/Board + Books + Personal) - (Federal + State + Institutional grants)
    "avg_net_price": "cost.avg_net_price.private",
}


# Convert decimal values (0-1) to percentages (0-100) and round to 2 decimal places
# If values are already in percentage format (>1 or <0), just round them
def normalize_percentages(df: pd.DataFrame, percentage_fields=None) -> pd.DataFrame:
    if percentage_fields is None:
        percentage_fields = ["admission_rate", "retention_rate_ft", "grad_rate_150"]

    # c = column
    # s = series
    out = df.copy()
    for c in percentage_fields:
        if c in out.columns:
            s = pd.to_numeric(out[c], errors="coerce")
            if s.notna().any():
                valid = s[s.notna()]
                if len(valid) > 0 and valid.max() <= 1.0 and valid.min() >= 0:
                    # Convert decimals to percentages
                    out[c] = (s * 100).round(2)
                else:
                    # Already in percentage format, just round
                    out[c] = s.round(2)
    return out

# Build a database filename with timestamp
# Accepts optional datetime for testing
def build_db_filename(now=None) -> str:
    if now is None:
        now = datetime.now()
    timestamp = now.strftime("%Y%m%d")
    return f"music_schools_{timestamp}.db"


def create_database(db_path="music_schools.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Schools dimension table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS schools (
            school_id INTEGER PRIMARY KEY AUTOINCREMENT,
            unitid INTEGER UNIQUE NOT NULL,
            institution_name TEXT NOT NULL)""")

    # Main metrics table (long format)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS school_metrics (
            school_id INTEGER NOT NULL,
            year INTEGER NOT NULL,
            enrollment_total INTEGER,
            admission_rate REAL,
            retention_rate_ft REAL,
            grad_rate_150 REAL,
            tuition_fees REAL,
            avg_net_price REAL,
            PRIMARY KEY (school_id, year),
            FOREIGN KEY (school_id) REFERENCES schools(school_id))""")

    # Indexes for common query patterns
    cursor.execute(
        "CREATE INDEX IF NOT EXISTS idx_metrics_year ON school_metrics(year)")

    cursor.execute(
        "CREATE INDEX IF NOT EXISTS idx_metrics_school ON school_metrics(school_id)")

    # Schools & Metrics View
    cursor.execute("""
        CREATE VIEW IF NOT EXISTS v_school_metrics AS
        SELECT s.institution_name,
            s.unitid,
            m.school_id,
            m.year,
            m.enrollment_total,
            m.admission_rate,
            m.retention_rate_ft,
            m.grad_rate_150,
            m.tuition_fees,
            m.avg_net_price
        FROM school_metrics m
        JOIN schools s ON m.school_id = s.school_id""")

    # YoY Comparisons View
    cursor.execute("""
        CREATE VIEW IF NOT EXISTS v_metrics_yoy AS
        SELECT s.institution_name, s.unitid, m.school_id, m.year,
        
        m.enrollment_total,
        m.enrollment_total - LAG(m.enrollment_total)
        OVER (PARTITION BY m.school_id ORDER BY m.year) AS enrollment_change,
        
        m.tuition_fees,
        m.tuition_fees - LAG(m.tuition_fees)
        OVER (PARTITION BY m.school_id ORDER BY m.year) AS tuition_change,
        
        m.avg_net_price,
        m.avg_net_price - LAG(m.avg_net_price)
        OVER (PARTITION BY m.school_id ORDER BY m.year) AS net_price_change,

        m.admission_rate, m.retention_rate_ft, m.grad_rate_150
        
        FROM school_metrics m
        JOIN schools s ON m.school_id = s.school_id""")

    # School Stats Summary View
    cursor.execute("""
        CREATE VIEW IF NOT EXISTS v_school_summary AS

        SELECT s.institution_name, s.unitid,
            COUNT(m.year) AS years_of_data,
            MIN(m.year) AS first_year,
            MAX(m.year) AS last_year,
            ROUND(AVG(m.enrollment_total), 0) AS avg_enrollment,
            ROUND(AVG(m.admission_rate), 2) AS avg_admission_rate,
            ROUND(AVG(m.retention_rate_ft), 2) AS avg_retention_rate,
            ROUND(AVG(m.grad_rate_150), 2) AS avg_grad_rate,
            ROUND(AVG(m.tuition_fees), 2) AS avg_tuition,
            ROUND(AVG(m.avg_net_price), 2) AS avg_net_price
        
        FROM schools s
        LEFT JOIN school_metrics m ON s.school_id = m.school_id
        GROUP BY s.school_id, s.institution_name, s.unitid""")

    conn.commit()
    return conn

def insert_schools(conn):
    cursor = conn.cursor()

    for unitid, name in NAME_MAP.items():
        cursor.execute("""
            INSERT OR IGNORE INTO schools (unitid, institution_name)
            VALUES (?, ?)""",
            (unitid, name),)

    conn.commit()

def get_school_id(conn, unitid):
    cursor = conn.cursor()
    cursor.execute("SELECT school_id FROM schools WHERE unitid = ?", (unitid,))
    result = cursor.fetchone()
    return result[0] if result else None

def fetch_year(institution_ids, year):
    fields = ["id", "school.name"]
    fields += [f"{year}.{f}" for f in FIELD_MAP.values()]
    params = {
        "api_key": API_KEY,
        "id__in": ",".join(map(str, institution_ids)),
        "fields": ",".join(fields),
        "per_page": 100,
    }

    # res = response
    # ex = exception
    try:
        res = requests.get(API, params=params, timeout=30)
        res.raise_for_status()
        js = res.json()
    except requests.exceptions.RequestException as ex:
        print(f"Error fetching data for year {year}: {ex}")
        return pd.DataFrame()
    except ValueError as ex:
        print(f"Error parsing JSON response for year {year}: {ex}")
        return pd.DataFrame()

    results = js.get("results", [])
    if not results:
        print(f"Warning: No data returned for year {year}")
        return pd.DataFrame()

    rows = []
    for item in results:
        row_id = item.get("id")
        if not row_id:
            continue

        row = {
            "institution": NAME_MAP.get(row_id, item.get("school.name", "Unknown")),
            "unitid": row_id,
            "year": year,
        }

        for metric, field_suffix in FIELD_MAP.items():
            row[metric] = item.get(f"{year}.{field_suffix}")
        rows.append(row)
    return pd.DataFrame(rows)


# Insert metrics data into the school_metrics table
def insert_metrics(conn, df):
    cursor = conn.cursor()

    # Add school_id to dataframe for sorting
    df["school_id"] = df["unitid"].map(lambda x: get_school_id(conn, x))
    df_sorted = df.sort_values(["school_id", "year"])

    for _, row in df_sorted.iterrows():
        school_id = row["school_id"]
        if not school_id:
            continue

        try:
            cursor.execute("""
                INSERT OR REPLACE INTO school_metrics 
                (school_id, year, enrollment_total, admission_rate, retention_rate_ft,
                 grad_rate_150, tuition_fees, avg_net_price)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    school_id,
                    row["year"],
                    row.get("enrollment_total"),
                    row.get("admission_rate"),
                    row.get("retention_rate_ft"),
                    row.get("grad_rate_150"),
                    row.get("tuition_fees"),
                    row.get("avg_net_price"),
                ),
            )
        except sqlite3.Error as err:
            print(
                f"Error inserting data for school_id {school_id}, year {row['year']}: {err}"
            )

    conn.commit()

def main():
    print("Creating Database...")
    print("=" * 60)

    db_path = build_db_filename()
    conn = create_database(db_path)
    insert_schools(conn)
    failed_years = []
    all_data = []

    for year in YEARS:
        print(f"\nFetching data for {year}...")
        df = fetch_year(UNITIDS, year)

        if not df.empty:
            df = normalize_percentages(df)
            all_data.append(df)
        else:
            failed_years.append(year)
            print(f"  No data available for {year}")

    if all_data:
        combined_df = pd.concat(all_data, ignore_index=True)
        insert_metrics(conn, combined_df)
        print(f"\nInserted {len(combined_df)} records into school_metrics table")

    if failed_years:
        print(f"\nWarning: Failed to retrieve data for years: {failed_years}")

    conn.close()
    print(f"\nDatabase saved as: {db_path}")
    print("\nAvailable views for easy querying:")
    print("  - v_school_metrics    : All data with school names")
    print("  - v_metrics_yoy       : Year-over-year changes")
    print("  - v_school_summary    : Summary statistics per school")
    print(f"\nExample query:")
    print(f'  sqlite3 -header -column {db_path} "SELECT * FROM v_school_metrics;"')

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nOperation cancelled by user.")
    except Exception as ex:
        print(f"Error: {ex}")
        raise SystemExit(1)