hackNY/analyze.py

#!/usr/bin/python3

import csv
import math
import numpy
import pandas
import statistics
import sys
from collections import defaultdict
from typing import DefaultDict, Dict, List, Optional, Tuple, Union


# recommendation related constants
MAX_RECOMMENDATIONS_PER_APPLICANT = 3
# review related constants
REVIEW_ROUND_ID = 1
MAX_REVIEWERS_PER_APPLICANT = 5
MIN_REVIEWERS_PER_APPLICANT = 2
# interview related constants
INTERVIEW_ROUND_ID = 2
MAX_INTERVIEWERS_PER_APPLICANT = 1
MIN_INTERVIEWERS_PER_APPLICANT = 1
# relative file paths
APPLICANT_DATA_CSV_PATH = "output/applicant_data.csv"
APPLICANT_INSIGHTS_GENDER_CSV_PATH = "output/applicant_insights_gender.csv"
APPLICANT_INSIGHTS_GRAD_YEAR_CSV_PATH = "output/applicant_insights_grad_year.csv"
APPLICANT_INSIGHTS_SCHOOL_CSV_PATH = "output/applicant_insights_school.csv"
REVIEWER_DATA_CSV_PATH = "output/reviewer_data.csv"
INTERVIEWER_DATA_CSV_PATH = "output/interviewer_data.csv"
# misc constants
GENDER_ALIASES = {
    "Male": ["m", "male", "man"],
    "Female": ["f", "female", "woman"],
    "Nonbinary": ["nb", "non-binary", "nb", numpy.nan],
}
# invert gender_aliases dict
ALIASES_GENDER = {}
for gender, aliases in GENDER_ALIASES.items():
    for alias in aliases:
        ALIASES_GENDER[alias] = gender
HELP_TEXT = """
I need one input file.
1. applicants.tsv with headers: all visible columns from the export view on Embark
"""
# question constants
INTERNSHIP_EXISTS_QUESTION = "If you have already secured a summer internship, please specify the compan
y name and the start/end dates of your internship here."
INTERNSHIP_PLAN_QUESTION = "If you have NOT already secured a summer internship, please explain your cur
rent plan to secure one here."


def normalize_score(score: Union[float, str]) -> float:
    # scores could be float:numpy.nan or str:"3: Good"
    if isinstance(score, str):
        return float(score[0])
    return score


def get_stdev(variance: float) -> float:
    # could be -1.0 (no score exists)
    if variance < 0:
        return -1.0
    return math.sqrt(variance)


def get_mean(values: List[float]) -> float:
    if values and len(values) > 1:
        return statistics.mean(values)
    elif values:
        return values[0]
    else:
        return numpy.nan


def get_variance(values: List[float]) -> float:
    if values and len(values) > 1:
        return statistics.variance(values)
    else:
        return numpy.nan


class Recommendation:
    def __init__(self, applicant: str, recommender_id: int, series: pandas.Series):
        prefix = f"Rec {recommender_id}"
        self.applicant = applicant
        self.recommender = series[f"{prefix} Full Name"]

    @property
    def exists(self) -> bool:
        return isinstance(self.recommender, str)


class Review:
    def __init__(self, applicant: str, reviewer_id: int, series: pandas.Series):
        prefix = f"Rd {REVIEW_ROUND_ID} R{reviewer_id}"
        self.applicant = applicant
        self.reviewer = series[f"{prefix} Name"]
        self.status = series[f"{prefix} Status"]
        self.technical_score = normalize_score(series[f"{prefix} Technical Competency Score"])
        self.community_score = normalize_score(series[f"{prefix} Community Involvement Score"])

    @property
    def exists(self) -> bool:
        return isinstance(self.reviewer, str)

    @property
    def is_complete(self) -> bool:
        return bool(
            self.status == "Completed"
            and self.technical_score is not numpy.nan
            and self.community_score is not numpy.nan
        )


class Interview:
    def __init__(self, applicant: str, interviewer_id: int, series: pandas.Series):
        prefix = f"Rd {INTERVIEW_ROUND_ID} R{interviewer_id}"
        self.applicant = applicant
        self.interviewer = series[f"{prefix} Name"]
        self.status = series[f"{prefix} Status"]
        self.fit_score = normalize_score(series[f"{prefix} Candidate Fit Score"])
        self.fit_confidence = normalize_score(series[f"{prefix} Candidate Fit Confidence Rating"])

    @property
    def exists(self) -> bool:
        return isinstance(self.interviewer, str)

    @property
    def is_complete(self) -> bool:
        return bool(
            self.status == "Completed"
            and self.fit_score is not numpy.nan
            and self.fit_confidence is not numpy.nan
        )


class Applicant:
    def __init__(self, series: pandas.Series):
        self.series = series

    @property
    def name(self):
        first_name = self.series["First Name"]
        last_name = self.series["Last Name"]
        return f"{first_name} {last_name}"

    @property
    def internship_response(self):
        internship_exists_response = self.series[INTERNSHIP_EXISTS_QUESTION]
        internship_plan_response = self.series[INTERNSHIP_PLAN_QUESTION]
        if internship_exists_response is numpy.nan:
            return internship_plan_response
        else:
            return internship_exists_response

    @property
    def email(self):
        return self.series["E-mail address"]

    @property
    def school(self):
        return self.series["University/College"]

    @property
    def gender(self):
        return (
            self.series["Gender"].lower()
            if isinstance(self.series["Gender"], str)
            else numpy.nan
        )

    @property
    def race_ethnicity(self):
        return self.series["Race / Ethnicity"]

    @property
    def graduation_year(self):
        # exported format is XX-XX-20XX
        graduation_date = self.series["Graduation Date"]
        return graduation_date[-4:]

    @property
    def reviewers(self):
        return [
            " ".join(reversed(str(self.series[f"Rd 1 R{i} Name"]).split(", ")))
            for i in range(1, 1 + MAX_REVIEWERS_PER_APPLICANT)
        ]

    @property
    def recommendations(self) -> List[Recommendation]:
        recommendations = [
            Recommendation(self.name, i + 1, self.series)
            for i in range(MAX_RECOMMENDATIONS_PER_APPLICANT)
        ]
        return [r for r in recommendations if r.exists]


    @property
    def reviews(self) -> List[Review]:
        reviews = [
            Review(self.name, i + 1, self.series)
            for i in range(MAX_REVIEWERS_PER_APPLICANT)
        ]
        return [r for r in reviews if r.exists]

    @property
    def interviews(self) -> List[Interview]:
        interviews = [
            Interview(self.name, i + 1, self.series)
            for i in range(MAX_INTERVIEWERS_PER_APPLICANT)
        ]
        return [i for i in interviews if i.exists]

    @property
    def complete_reviews(self) -> List[Review]:
        return [review for review in self.reviews if review.is_complete]

    @property
    def complete_interviews(self) -> List[Interview]:
        return [interview for interview in self.interviews if interview.is_complete]

    @property
    def technical_scores(self) -> List[float]:
        return [review.technical_score for review in self.complete_reviews]

    @property
    def community_scores(self) -> List[float]:
        return [review.community_score for review in self.complete_reviews]

    @property
    def technical_moments(self) -> Tuple[float, float]:
        if not self.is_reviewed:
            return (-1.0, -1.0)
        return (
            get_mean(self.technical_scores),
            get_variance(self.technical_scores),
        )

    @property
    def community_moments(self) -> Tuple[float, float]:
        if not self.is_reviewed:
            return (-1.0, -1.0)
        return (
            get_mean(self.community_scores),
            get_variance(self.community_scores),
        )

    @property
    def overall_moments(self) -> Tuple[float, float]:
        if not self.is_reviewed:
            return (-1.0, -1.0)
        return numpy.add(self.technical_moments, self.community_moments) / 2


    @property
    def is_reviewed(self) -> bool:
        return len(self.complete_reviews) >= MIN_REVIEWERS_PER_APPLICANT

    @property
    def fit_scores(self) -> List[float]:
        return [interview.fit_score for interview in self.complete_interviews]

    @property
    def fit_confidences(self) -> List[float]:
        return [interview.fit_confidence for interview in self.complete_interviews]

    @property
    def fit_score_moments(self) -> List[float]:
        if not self.is_interviewed:
            return (-1.0, -1.0)
        return (
            get_mean(self.fit_scores),
            get_variance(self.fit_confidences),
        )

    @property
    def fit_confidence_moments(self) -> List[float]:
        if not self.is_interviewed:
            return (-1.0, -1.0)
        return (
            get_mean(self.fit_confidences),
            get_variance(self.fit_confidences),
        )

    @property
    def is_interviewed(self) -> bool:
        return len(self.complete_interviews) >= MIN_INTERVIEWERS_PER_APPLICANT

    def to_dict(self) -> Dict:
        technical_mean, technical_variance = self.technical_moments
        community_mean, community_variance = self.community_moments
        fit_score_mean, fit_score_variance = self.fit_score_moments
        fit_confidence_mean, fit_confidence_variance = self.fit_confidence_moments
        return {
            "Name": self.name,
            "Email": self.email,
            "Gender": self.gender,
            "Race/Ethnicity": self.race_ethnicity,
            "Graduation Year": self.graduation_year,
            "Internship Response": self.internship_response,
            "# Recommenders": len(self.recommendations),
            "# Complete Reviews": len(self.complete_reviews),
            "# Complete Interviews": len(self.complete_interviews),
            "Reviewers": "|".join(self.reviewers),
            "Review Technical Score: Mean": technical_mean,
            "Review Technical Score: Std Dev": get_stdev(technical_variance),
            "Review Community Score: Mean": community_mean,
            "Review Community Score: Std Dev": get_stdev(community_variance),
            "Review Overall Score: Mean": (technical_mean + community_mean) / 2,
            "Review Overall Score: Std Dev": get_stdev((technical_variance + community_variance) / 2),
            "Interviewer": self.interviews[0].interviewer if self.interviews else numpy.nan,
            "Interview Status": self.interviews[0].status if self.interviews else numpy.nan,
            "Interview Fit Score: Mean": fit_score_mean,
            "Interview Fit Score: Std Dev": get_stdev(fit_score_variance),
            "Interview Fit Confidence: Mean": fit_confidence_mean,
            "Interview Fit Confidence: Std Dev": get_stdev(fit_confidence_variance),
        }


class Recommender:
    def __init__(self, name: str, recommendations: List[Recommendation]):
        self.name = name
        self.recommendations = recommendations


class Reviewer:
    def __init__(self, name: str, reviews: List[Review]):
        self.name = name
        self.reviews = reviews

    @property
    def complete_reviews(self) -> List[Review]:
        return [review for review in self.reviews if review.is_complete]

    @property
    def incomplete_reviews(self) -> List[Review]:
        return [review for review in self.reviews if not review.is_complete]

    @property
    def technical_scores(self) -> List[float]:
        return [review.technical_score for review in self.complete_reviews]

    @property
    def community_scores(self) -> List[float]:
        return [review.community_score for review in self.complete_reviews]

    @property
    def technical_moments(self) -> Tuple[float, float]:
        if not len(self.technical_scores):
            return (-1.0, -1.0)
        return (
            get_mean(self.technical_scores),
            get_variance(self.technical_scores),
        )

    @property
    def community_moments(self) -> Tuple[float, float]:
        if not len(self.community_scores):
            return (-1.0, -1.0)
        return (
            get_mean(self.community_scores),
            get_variance(self.community_scores),
        )

    def to_dict(self) -> Dict:
        technical_mean, technical_variance = self.technical_moments
        community_mean, community_variance = self.community_moments
        return {
            "Name": self.name,
            "# Complete Reviews": len(self.complete_reviews),
            "# Incomplete Reviews": len(self.incomplete_reviews),
            "Technical Score: Mean": technical_mean,
            "Technical Score: Std Dev": get_stdev(technical_variance),
            "Community Score: Mean": community_mean,
            "Community Score: Std Dev": get_stdev(community_variance),
            "Overall Score: Mean": (technical_mean + community_mean) / 2,
            "Overall Score: Std Dev": get_stdev((technical_variance + community_variance) / 2),
        }


class Interviewer:
    def __init__(self, name: str, interviews: List[Interview]):
        self.name = name
        self.interviews = interviews

    @property
    def complete_interviews(self) -> List[Interview]:
        return [interview for interview in self.interviews if interview.is_complete]

    @property
    def incomplete_interviews(self) -> List[Review]:
        return [interview for interview in self.interviews if not interview.is_complete]

    @property
    def fit_scores(self) -> List[float]:
        return [interview.fit_score for interview in self.complete_interviews]

    @property
    def fit_confidences(self) -> List[float]:
        return [interview.fit_confidence for interview in self.complete_interviews]

    @property
    def fit_score_moments(self) -> Tuple[float, float]:
        if not len(self.fit_scores):
            return (-1.0, -1.0)
        return (
            get_mean(self.fit_scores),
            get_variance(self.fit_scores),
        )

    @property
    def fit_confidence_moments(self) -> Tuple[float, float]:
        if not len(self.fit_confidences):
            return (-1.0, -1.0)
        return (
            get_mean(self.fit_confidences),
            get_variance(self.fit_confidences),
        )

    def to_dict(self) -> Dict:
        fit_score_mean, fit_score_variance = self.fit_score_moments
        fit_confidence_mean, fit_confidence_variance = self.fit_confidence_moments
        return {
            "Name": self.name,
            "# Complete Interviews": len(self.complete_interviews),
            "# Incomplete Interviews": len(self.incomplete_interviews),
            "Fit Score: Mean": fit_score_mean,
            "Fit Score: Std Dev": get_stdev(fit_score_variance),
            "Fit Confidence: Mean": fit_confidence_mean,
            "Fit Confidence: Std Dev": get_stdev(fit_confidence_variance),
        }


def register_applicants(df: pandas.DataFrame) -> List[Applicant]:
    applicants = []
    for _row_id, series in df.iterrows():
        applicant = Applicant(series)
        applicants.append(applicant)
    print(f"- Registered {len(applicants)} applicants.")
    return applicants


def register_recommenders(applicants: List[Applicant]) -> List[Recommender]:
    recommenders = []
    recommender_recommendations: DefaultDict[str, List[Recommendation]] = defaultdict(list)
    for applicant in applicants:
        for recommendation in applicant.recommendations:
            recommender_recommendations[recommendation.recommender].append(recommendation)
    for recommender_name, recommendations in recommender_recommendations.items():
        recommender = Recommender(recommender_name, recommendations)
        recommenders.append(recommender)
    print(f"- Registered {len(recommenders)} recommenders.")
    return recommenders


def register_reviewers(applicants: List[Applicant]) -> List[Reviewer]:
    reviewers = []
    reviewer_reviews: DefaultDict[str, List[Review]] = defaultdict(list)
    for applicant in applicants:
        for review in applicant.reviews:
            reviewer_reviews[review.reviewer].append(review)
    for reviewer_name, reviews in reviewer_reviews.items():
        reviewer = Reviewer(reviewer_name, reviews)
        reviewers.append(reviewer)
    print(f"- Registered {len(reviewers)} reviewers.")
    return reviewers


def register_interviewers(applicants: List[Applicant]) -> List[Interviewer]:
    interviewers = []
    interviewer_interviews: DefaultDict[str, List[Interview]] = defaultdict(list)
    for applicant in applicants:
        for interview in applicant.interviews:
            interviewer_interviews[interview.interviewer].append(interview)
    for interviewer_name, interviews in interviewer_interviews.items():
        interviewer = Interviewer(interviewer_name, interviews)
        interviewers.append(interviewer)
    print(f"- Registered {len(interviewers)} interviewers.")
    return interviewers


def generate_applicant_key_insights(
    applicants: List[Applicant], key_name: str, alias_key_lookup: Optional[Dict] = None
) -> Dict[str, Dict[str, float]]:
    applicant_key_insights = {}
    # bucket all review/interview scores by key (school, grad year, gender, etc.)
    # key must be a property of the Applicant class
    key_freq = defaultdict(int)
    technical_scores = defaultdict(list)
    community_scores = defaultdict(list)
    fit_scores = defaultdict(list)
    fit_confidences = defaultdict(list)
    for applicant in applicants:
        key = getattr(applicant, key_name)
        if alias_key_lookup:
            key = alias_key_lookup.get(key)
        key_freq[key] += 1
        technical_scores[key].extend(applicant.technical_scores)
        community_scores[key].extend(applicant.community_scores)
        fit_scores[key].extend(applicant.fit_scores)
        fit_confidences[key].extend(applicant.fit_confidences)
    # generate insights by key
    for key in key_freq.keys():
        review_technical_mean = get_mean(technical_scores[key])
        review_technical_variance = get_variance(technical_scores[key])
        review_community_mean = get_mean(community_scores[key])
        review_community_variance = get_variance(community_scores[key])
        review_overall_mean = (review_technical_mean + review_community_mean) / 2
        review_overall_variance = (review_technical_variance + review_community_variance) / 2
        interview_fit_score_mean = get_mean(fit_scores[key]) if fit_scores[key] else -1.0
        interview_fit_score_variance = (
            get_variance(fit_scores[key]) if fit_scores[key] else -1.0
        )
        interview_fit_confidence_mean = (
            get_mean(fit_confidences[key]) if fit_confidences[key] else -1.0
        )
        interview_fit_confidence_variance = (
            get_variance(fit_confidences[key]) if fit_confidences[key] else -1.0
        )
        applicant_key_insights[key] = {
            "Review Technical Score: Mean": review_technical_mean,
            "Review Technical Score: Std Dev": get_stdev(review_technical_variance),
            "Review Community Score: Mean": review_community_mean,
            "Review Community Score: Std Dev": get_stdev(review_community_variance),
            "Review Overall Score: Mean": review_overall_mean,
            "Review Overall Score: Std Dev": get_stdev(review_overall_variance),
            "Interview Fit Score: Mean": interview_fit_score_mean,
            "Interview Fit Score: Std Dev": get_stdev(interview_fit_score_variance),
            "Interview Fit Confidence: Mean": interview_fit_confidence_mean,
            "Interview Fit Confidence: Std Dev": get_stdev(interview_fit_confidence_variance),
        }
    return [
        format_dict({key_name.capitalize(): key, "Count": key_freq[key], **insights})
        for key, insights in applicant_key_insights.items()
    ]


def generate_applicant_insights(applicants: List[Applicant]) -> List[Dict]:
    return [format_dict(a.to_dict()) for a in applicants]


def generate_reviewer_insights(reviewers: List[Reviewer]) -> List[Dict]:
    return [format_dict(r.to_dict()) for r in reviewers]


def generate_interviewer_insights(interviewers: List[Interviewer]) -> List[Dict]:
    return [format_dict(r.to_dict()) for r in interviewers]


def write_to_csv(
    filename: str, entry_dicts: List[Dict], sort_headers: Optional[List[str]] = None
) -> None:
    if not entry_dicts:
        print(f"- Cannot write to {filename}: entry_dicts is empty.")
        return
    if sort_headers:
        entry_dicts = sorted(
            entry_dicts, reverse=True, key=lambda d: tuple(d.get(h) for h in sort_headers)
        )
    with open(filename, 'w', newline='') as csvfile:
        headers = list(entry_dicts[0])
        writer = csv.DictWriter(csvfile, fieldnames=headers)
        writer.writeheader()
        for entry_dict in entry_dicts:
            writer.writerow(entry_dict)


def format_dict(d: Dict) -> Dict:
    formatted_d = {}
    for key, value in d.items():
        if isinstance(value, bool):
            formatted_d[key] = str(value)
        elif isinstance(value, float):
            formatted_d[key] = f"{value:.2f}"
        else:
            formatted_d[key] = value
    return formatted_d


def main(df: pandas.DataFrame):
    print("\nRegistering structs...")
    applicants = register_applicants(df)
    recommenders = register_recommenders(applicants)
    reviewers = register_reviewers(applicants)
    interviewers = register_interviewers(applicants)
    print("\nGenerating data + insights...")
    applicant_dicts = generate_applicant_insights(applicants)
    applicant_insights_gender = generate_applicant_key_insights(
        applicants, "gender", alias_key_lookup=ALIASES_GENDER
    )
    applicant_insights_grad_year = generate_applicant_key_insights(applicants, "graduation_year")
    applicant_insights_school = generate_applicant_key_insights(applicants, "school")
    reviewer_dicts = generate_reviewer_insights(reviewers)
    interviewer_dicts = generate_interviewer_insights(interviewers)
    print("\nWriting to CSVs...")
    write_to_csv(
        APPLICANT_DATA_CSV_PATH,
        applicant_dicts,
        sort_headers=["Review Overall Score: Mean"],
    )
    write_to_csv(
        APPLICANT_INSIGHTS_GENDER_CSV_PATH,
        applicant_insights_gender,
        sort_headers=["Review Overall Score: Mean"]
    )
    write_to_csv(
        APPLICANT_INSIGHTS_GRAD_YEAR_CSV_PATH,
        applicant_insights_grad_year,
        sort_headers=["Review Overall Score: Mean"]
    )
    write_to_csv(
        APPLICANT_INSIGHTS_SCHOOL_CSV_PATH,
        applicant_insights_school,
        sort_headers=["Review Overall Score: Mean"],
    )
    write_to_csv(
        REVIEWER_DATA_CSV_PATH,
        reviewer_dicts,
        sort_headers=["Overall Score: Mean"],
    )
    write_to_csv(
        INTERVIEWER_DATA_CSV_PATH,
        interviewer_dicts,
        sort_headers=["Fit Score: Mean"],
    )
    print("\nComplete.")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(HELP_TEXT)
        sys.exit(1)
    df = pandas.read_csv(sys.argv[1], sep="\t")
    main(df)