hackNY/match.py

#!/usr/bin/python3

import pandas
import prettytable
import queue
import random
import sys
from typing import List, Tuple


# misc constants
MAX_INTERVIEWS_PER_REVIEWER = 5
MIN_REVIEW_OVERALL_SCORE = 3.0
HELP_TEXT = """
I need two input files.
1. applicant_data.csv with headers: Name, Email, Interviewer, Review Overall Score: Mean, Interview Fit Score: Mean
2. interviewer_data.csv with headers: Name, Email, Confirmed?, #Committed
"""

# TODO: WRITE DATA TO CSVS AUTOMATICALLY?


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

    @property
    def name(self):
        return self.series["Name"]

    @property
    def email(self):
        return self.series["Email"]

    @property
    def reviewers(self):
        return self.series["Reviewers"].split("|")

    @property
    def review_overall_score(self):
        return self.series["Review Overall Score: Mean"]

    @property
    def is_candidate(self):
        return float(self.review_overall_score) >= MIN_REVIEW_OVERALL_SCORE

    @property
    def interviewer(self):
        return self.series["Interviewer"]

    @property
    def interview_fit_score(self):
        return self.series["Interview Fit Score: Mean"]


class Interviewer:
    def __init__(self, series: pandas.Series):
        self.candidates = []
        self.series = series

    @property
    def name(self):
        return self.series["Name"]

    @property
    def email(self):
        return self.series["Email"]

    @property
    def is_confirmed(self) -> bool:
        return (
            isinstance(self.series["Confirmed?"], str)
            and "Yes" in self.series["Confirmed?"]
        )

    @property
    def n_committed(self) -> int:
        return min(int(self.series["#Committed"]), MAX_INTERVIEWS_PER_REVIEWER)

    @property
    def n_actually_committed(self) -> int:
        return int(self.series["#Committed"])

    @property
    def n_conducted(self) -> int:
        return int(self.series["#Conducted"])


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


def register_interviewers(df: pandas.DataFrame) -> List[Interviewer]:
    interviewers = []
    for _row_id, series in df.iterrows():
        interviewer = Interviewer(series)
        if interviewer.is_confirmed:
            interviewers.append(interviewer)
    print(f"- Registered {len(interviewers)} interviewers.")
    return interviewers


def summarize_interviewer_progress(interviewers: List[Interviewer]):
    interviewers = sorted(interviewers, key=lambda i: i.n_actually_committed, reverse=True)
    table = prettytable.PrettyTable(["Name", "# Conducted", "# Matches", "# Committed"])
    for interviewer in interviewers:
        row = [
            interviewer.name,
            interviewer.n_conducted,
            len(interviewer.candidates),
            interviewer.n_actually_committed,
        ]
        table.add_row(row)
    print(table)


def generate_new_matches(
    todo_candidates: List[Applicant], interviewers: List[Interviewer]
) -> List[Tuple[Applicant, Interviewer]]:
    new_matches = []
    pq = queue.PriorityQueue()
    random.shuffle(todo_candidates)
    random.shuffle(interviewers)
    interviewer_index = {i.name: i for i in interviewers}
    for interviewer in interviewers:
        priority = get_priority(interviewer)
        pq.put((priority, interviewer.name))
    # initially, distribute based on interviewer capacity
    while todo_candidates:
        candidate = todo_candidates.pop()
        priority, interviewer_name = pq.get()
        interviewer = interviewer_index[interviewer_name]
        interviewer.candidates.append(candidate)
        new_matches.append((candidate, interviewer))
        new_priority = get_priority(interviewer)
        pq.put((new_priority, interviewer_name))
    # randomly redistribute if needed
    for i in range(len(new_matches)):
        candidate, interviewer = new_matches[i]
        if interviewer.name in candidate.reviewers:
            new_matches.pop(i)
            interviewer.candidates.pop(interviewer.candidates.index(candidate))
            valid_interviewers = [
                i for i in interviewers
                if i.name not in candidate.reviewers and len(i.candidates) < i.n_committed
            ]
            new_matches.append((candidate, random.choice(valid_interviewers)))
    return new_matches


# this function outputs something comparable
def get_priority(interviewer: Interviewer):
    n_existing_matches = len(interviewer.candidates)
    remaining_capacity = interviewer.n_committed - n_existing_matches
    candidate_saturation_ratio = n_existing_matches / interviewer.n_committed
    # apply a positive when you want the number to lean against new matches
    # apply a negative when you want the number to lean toward new matches
    return (candidate_saturation_ratio, n_existing_matches, -1 * remaining_capacity)


def main(applicant_df: pandas.DataFrame, interviewer_df: pandas.DataFrame):
    print("\nRegistering structs...")
    n_existing_matches = 0
    todo_candidates = []
    candidates = register_candidates(applicant_df)
    interviewers = register_interviewers(interviewer_df)
    interviewer_index = {i.name: i for i in interviewers}
    for candidate in candidates:
        if isinstance(candidate.interviewer, str):
            interviewer = interviewer_index[candidate.interviewer]
            interviewer.candidates.append(candidate.name)
            n_existing_matches += 1
        else:
            todo_candidates.append(candidate)
    print(f"- Registered {n_existing_matches} existing candidate::interviewer matches.")
    print("\nHere is a summary of current interviewer progress.")
    summarize_interviewer_progress(interviewers)
    if not todo_candidates:
        print("\nNo candidates need matches. Exiting.")
        return
    print(f"\nGenerating {len(todo_candidates)} matches...")
    new_matches = generate_new_matches(todo_candidates, interviewers)
    print("\nHere are the new matches. Assign them in Embark and email the relevant parties.")
    for candidate, interviewer in new_matches:
        print(f"- {candidate.name}: {interviewer.name}")
    print("\nHere is a summary of future interviewer progress (after the new matches).")
    summarize_interviewer_progress(interviewers)
    print("\nComplete.")


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(HELP_TEXT)
        sys.exit(1)
    applicant_df = pandas.read_csv(sys.argv[1])
    interviewer_df = pandas.read_csv(sys.argv[2])
    main(applicant_df, interviewer_df)

~/hackny/admissions/match
❯ cat match_interviewers.py
#!/usr/bin/python3

import pandas
import prettytable
import queue
import random
import sys
from typing import List, Tuple


# misc constants
MAX_INTERVIEWS_PER_REVIEWER = 5
MIN_REVIEW_OVERALL_SCORE = 3.0
HELP_TEXT = """
I need two input files.
1. applicant_data.csv with headers: Name, Email, Interviewer, Review Overall Score: Mean, Interview Fit Score: Mean
2. interviewer_data.csv with headers: Name, Email, Confirmed?, #Committed
"""

# TODO: WRITE DATA TO CSVS AUTOMATICALLY?


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

    @property
    def name(self):
        return self.series["Name"]

    @property
    def email(self):
        return self.series["Email"]

    @property
    def reviewers(self):
        return self.series["Reviewers"].split("|")

    @property
    def review_overall_score(self):
        return self.series["Review Overall Score: Mean"]

    @property
    def is_candidate(self):
        return float(self.review_overall_score) >= MIN_REVIEW_OVERALL_SCORE

    @property
    def interviewer(self):
        return self.series["Interviewer"]

    @property
    def interview_fit_score(self):
        return self.series["Interview Fit Score: Mean"]


class Interviewer:
    def __init__(self, series: pandas.Series):
        self.candidates = []
        self.series = series

    @property
    def name(self):
        return self.series["Name"]

    @property
    def email(self):
        return self.series["Email"]

    @property
    def is_confirmed(self) -> bool:
        return (
            isinstance(self.series["Confirmed?"], str)
            and "Yes" in self.series["Confirmed?"]
        )

    @property
    def n_committed(self) -> int:
        return min(int(self.series["#Committed"]), MAX_INTERVIEWS_PER_REVIEWER)

    @property
    def n_actually_committed(self) -> int:
        return int(self.series["#Committed"])

    @property
    def n_conducted(self) -> int:
        return int(self.series["#Conducted"])


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


def register_interviewers(df: pandas.DataFrame) -> List[Interviewer]:
    interviewers = []
    for _row_id, series in df.iterrows():
        interviewer = Interviewer(series)
        if interviewer.is_confirmed:
            interviewers.append(interviewer)
    print(f"- Registered {len(interviewers)} interviewers.")
    return interviewers


def summarize_interviewer_progress(interviewers: List[Interviewer]):
    interviewers = sorted(interviewers, key=lambda i: i.n_actually_committed, reverse=True)
    table = prettytable.PrettyTable(["Name", "# Conducted", "# Matches", "# Committed"])
    for interviewer in interviewers:
        row = [
            interviewer.name,
            interviewer.n_conducted,
            len(interviewer.candidates),
            interviewer.n_actually_committed,
        ]
        table.add_row(row)
    print(table)


def generate_new_matches(
    todo_candidates: List[Applicant], interviewers: List[Interviewer]
) -> List[Tuple[Applicant, Interviewer]]:
    new_matches = []
    pq = queue.PriorityQueue()
    random.shuffle(todo_candidates)
    random.shuffle(interviewers)
    interviewer_index = {i.name: i for i in interviewers}
    for interviewer in interviewers:
        priority = get_priority(interviewer)
        pq.put((priority, interviewer.name))
    # initially, distribute based on interviewer capacity
    while todo_candidates:
        candidate = todo_candidates.pop()
        priority, interviewer_name = pq.get()
        interviewer = interviewer_index[interviewer_name]
        interviewer.candidates.append(candidate)
        new_matches.append((candidate, interviewer))
        new_priority = get_priority(interviewer)
        pq.put((new_priority, interviewer_name))
    # randomly redistribute if needed
    for i in range(len(new_matches)):
        candidate, interviewer = new_matches[i]
        if interviewer.name in candidate.reviewers:
            new_matches.pop(i)
            interviewer.candidates.pop(interviewer.candidates.index(candidate))
            valid_interviewers = [
                i for i in interviewers
                if i.name not in candidate.reviewers and len(i.candidates) < i.n_committed
            ]
            new_matches.append((candidate, random.choice(valid_interviewers)))
    return new_matches


# this function outputs something comparable
def get_priority(interviewer: Interviewer):
    n_existing_matches = len(interviewer.candidates)
    remaining_capacity = interviewer.n_committed - n_existing_matches
    candidate_saturation_ratio = n_existing_matches / interviewer.n_committed
    # apply a positive when you want the number to lean against new matches
    # apply a negative when you want the number to lean toward new matches
    return (candidate_saturation_ratio, n_existing_matches, -1 * remaining_capacity)


def main(applicant_df: pandas.DataFrame, interviewer_df: pandas.DataFrame):
    print("\nRegistering structs...")
    n_existing_matches = 0
    todo_candidates = []
    candidates = register_candidates(applicant_df)
    interviewers = register_interviewers(interviewer_df)
    interviewer_index = {i.name: i for i in interviewers}
    for candidate in candidates:
        if isinstance(candidate.interviewer, str):
            interviewer = interviewer_index[candidate.interviewer]
            interviewer.candidates.append(candidate.name)
            n_existing_matches += 1
        else:
            todo_candidates.append(candidate)
    print(f"- Registered {n_existing_matches} existing candidate::interviewer matches.")
    print("\nHere is a summary of current interviewer progress.")
    summarize_interviewer_progress(interviewers)
    if not todo_candidates:
        print("\nNo candidates need matches. Exiting.")
        return
    print(f"\nGenerating {len(todo_candidates)} matches...")
    new_matches = generate_new_matches(todo_candidates, interviewers)
    print("\nHere are the new matches. Assign them in Embark and email the relevant parties.")
    for candidate, interviewer in new_matches:
        print(f"- {candidate.name}: {interviewer.name}")
    print("\nHere is a summary of future interviewer progress (after the new matches).")
    summarize_interviewer_progress(interviewers)
    print("\nComplete.")


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(HELP_TEXT)
        sys.exit(1)
    applicant_df = pandas.read_csv(sys.argv[1])
    interviewer_df = pandas.read_csv(sys.argv[2])
    main(applicant_df, interviewer_df)