Technical Reference¶

def prepare_df_comparison(
    responses_df_all: pd.DataFrame,
    responses_df_comparison: "pd.Series[str]",
    q: str,
    q_comparison: str,
    ordering: dict[str, list[str]],
) -> tuple[pd.DataFrame, dict[Hashable, int]]:
    """Compare groups of participants (determined by comparison_series).

    This function is for single-choice questions.

    The output dataframe contains the following columns:
        - q: The answer options
        - q_comparison: The groups
        - total: total number of participants in this group
        - count: number of participants (in this group) that gave this answer
        - proportion: share of participants (relative to "total") that gave this answ.

    Args:
        responses_df_all: DataFrame with answers for the base question
        responses_df_comparison: Answers for the intersecting question
        q: name of the output column for answer options
        q_comparison: name of the output column for groups
        ordering: Answer re-ordering dict, e.g. ORDER from `order/order2024.py`

    Returns:
        Tuple of [DataFrame, group size dict]. The latter is used as N in plots.
    """
    assert "id" not in responses_df_all.columns
    responses_joined = responses_df_all.join(responses_df_comparison)

    grouped_by_center = responses_joined.groupby(q_comparison, observed=False)[q]
    responses_df_counts = pd.concat(
        [
            grouped_by_center.value_counts(normalize=True).rename("proportion"),
            grouped_by_center.value_counts().rename("count"),
        ],
        axis=1,
    ).reset_index()

    # sort DF
    order_left = ordering.get(q)
    if order_left:
        responses_df_counts[q] = pd.Categorical(
            responses_df_counts[q], categories=order_left, ordered=True
        )
    order_right = ordering.get(q_comparison)
    if order_right:
        responses_df_counts[q_comparison] = pd.Categorical(
            responses_df_counts[q_comparison], categories=order_right, ordered=True
        )
    responses_df_counts_sorted = responses_df_counts.sort_values(by=[q_comparison, q])

    return responses_df_counts_sorted, grouped_by_center.count().to_dict()

def prepare_df_comparison_multiple(
    responses_df: pd.DataFrame,
    comparison_series: "pd.Series[str]",
    q: str,
    q_comparison: str,
    ordering: dict[str, list[str]],
) -> tuple[pd.DataFrame, dict[Hashable, int]]:
    """Compare groups of participants (determined by comparison_series).

    This function is for multiple-choice questions.

    The output dataframe contains the following columns:
        - q: The answer options
        - q_comparison: The groups
        - total: total number of participants in this group
        - count: number of participants (in this group) that gave this answer
        - proportion: share of participants (relative to "total") that gave this answ.

    Args:
        responses_df: The main DataFrame of answers
        comparison_series: Participant group (shares index with the main DF)
        q: name of the output column for answer options
        q_comparison: name of the output column for groups
        ordering: Answer re-ordering dict, e.g. ORDER from `order/order2024.py`

    Returns:
        Tuple of [DataFrame, group size dict]. The latter is used as N in plots.
    """
    # boolean value: participants who answered anything (summed up per group later)
    responses_df["total"] = responses_df.sum(axis="columns").gt(0)
    # melt into long form, merge with comparison
    responses_melt = pd.melt(
        responses_df.reset_index(),
        id_vars=["id", "total"],
        var_name=q,
        value_name="count",
    ).join(comparison_series, on="id")

    # for each subquestion, count `True` values, and normalize per group
    responses_counts = (
        responses_melt.groupby([q_comparison, q]).sum().drop(columns=["id"])
    )
    responses_counts["proportion"] = (
        responses_counts["count"] / responses_counts["total"]
    )

    # re-index for sorting
    responses_clean = responses_counts.reset_index()

    # get the number of participants per group in q_comparison
    participants = responses_clean.set_index(q_comparison)["total"].drop_duplicates()

    # ordering (copied from `prepare_df_comparison` above)
    order_left = ordering.get(q)
    if order_left:
        responses_clean[q] = pd.Categorical(
            responses_clean[q], categories=order_left, ordered=True
        )
    order_right = ordering.get(q_comparison)
    if order_right:
        responses_clean[q_comparison] = pd.Categorical(
            responses_clean[q_comparison], categories=order_right, ordered=True
        )
    responses_sort = responses_clean.sort_values(by=[q_comparison, q])
    # print(responses_sort)

    return responses_sort, participants.astype(int).to_dict()

def prepare_df_multiple(
    data: pd.DataFrame, q: str, ordering: dict[str, list[str]]
) -> tuple[pd.DataFrame, int]:
    """Count participants in the data. This function is for multiple-choice questions.

    The output dataframe contains the following columns:
        - q: The answer options
        - count: number of participants (in this group) that gave this answer
        - proportion: share of participants (relative to "total") that gave this answ.

    Args:
        data: The main DataFrame of answers
        q: name of the output column for answer options
        ordering: Answer re-ordering dict, e.g. ORDER from `order/order2024.py`

    Returns:
        Tuple of [DataFrame, participant number]. The latter is used as N in plots.
    """
    # boolean value: participants who answered anything (summed up later)
    data["total"] = data.sum(axis="columns").gt(0)
    # melt into long form
    responses_melted = pd.melt(data, id_vars=["total"], value_name="count", var_name=q)
    responses_counts = responses_melted.groupby(q).sum()

    # add percentages column
    responses_counts["proportion"] = (
        responses_counts["count"] / responses_counts["total"]
    )

    # re-index for sorting
    responses_clean = responses_counts.reset_index()
    # get number of participants
    participants = responses_clean["total"].drop_duplicates().iloc[0]

    # sort the DF
    orderlist = ordering.get(q)
    if orderlist:
        # sort with given order
        responses_clean[q] = pd.Categorical(
            responses_clean[q], categories=orderlist, ordered=True
        )
        responses_sorted = responses_clean.sort_values(by=q)
    else:
        # no order given, sort by descending values
        responses_sorted = responses_clean.sort_values(by="count", ascending=False)
        # TODO: sorting by value is unstable between centers. We probably want
        #       to define a fixed order for all questions.
        # print(q, responses_df_counts_sorted["variable"].to_list())

    return responses_sorted, participants

def prepare_df_single(
    data: pd.DataFrame, q: str, ordering: dict[str, list[str]]
) -> tuple[pd.DataFrame, int]:
    """Count participants in the data. This function is for single-choice questions.

    The output dataframe contains the following columns:
        - q: The answer options
        - count: number of participants (in this group) that gave this answer
        - proportion: share of participants (relative to "total") that gave this answ.

    Args:
        data: The main DataFrame of answers
        q: name of the output column for answer options
        ordering: Answer re-ordering dict, e.g. ORDER from `order/order2024.py`

    Returns:
        Tuple of [DataFrame, participant number]. The latter is used as N in plots.
    """
    assert "id" not in data.columns
    N_question = data.count().iloc[0]

    # need to reset the index, otherwise count returns an empty DF.
    data_q_counts = (
        data.reset_index()
        .groupby(q, observed=False)
        .count()
        .rename(columns={"id": "count"})
    )

    # sort the dataframe
    data_q_counts_sorted = data_q_counts.reset_index()
    orderlist = ordering.get(q)
    if orderlist:
        # sort with given order
        data_q_counts_sorted[q] = pd.Categorical(
            data_q_counts_sorted[q], categories=orderlist, ordered=True
        )
        data_q_counts_sorted = data_q_counts_sorted.sort_values(by=q)

    # add percentages column
    data_q_counts_sorted_percentages = data_q_counts_sorted
    data_q_counts_sorted_percentages["proportion"] = (
        data_q_counts_sorted_percentages["count"] / N_question
    )

    return data_q_counts_sorted_percentages, N_question

def filter_by_center(
    survey: LimeSurveyData, responses: pd.DataFrame, center_code: str
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Filter responses by center.

    Args:
        survey: The survey object
        responses: DataFrame with responses
        center_code: ID of the center to filter by (like 'A01')

    Returns:
        Tuple of filtered DataFrame and remainder DataFrame
    """
    # get IDs for the given center, then filter by the IDs
    centers = survey.get_responses(CENTER)
    center_students = centers[centers[CENTER] == center_code]
    if "id" in responses.columns:
        filtered = responses.loc[
            responses["id"].astype(int).isin(center_students.index)
        ]
        remainder = responses.loc[
            ~responses["id"].astype(int).isin(center_students.index)
        ]
    else:
        filtered = responses[responses.index.isin(center_students.index)]
        remainder = responses[~responses.index.isin(center_students.index)]

    assert len(filtered) == len(center_students)
    assert len(responses) == len(filtered) + len(remainder)
    assert type(filtered) is pd.DataFrame and type(remainder) is pd.DataFrame
    return filtered, remainder

def get_as_numeric(
    survey: LimeSurveyData, q_code: str, blocklist: list[str]
) -> "pd.Series[float]":
    """Get numeric answers for the requested question code.

    Raises:
        ValueError: if non-numeric answer codes are not in the given blocklist.

    Args:
        survey: The survey object
        q_code: The question ID to be queried
        blocklist: Answer codes to be excluded from the result

    Returns:
        Numeric Series
    """
    answers = survey.get_responses(q_code)
    filtered = answers.loc[~answers[q_code].isin(blocklist)]

    choices = survey.get_choices(q_code)
    mapped = filtered.map(lambda a_code: choices[a_code])
    numeric = cast("pd.Series[float]", mapped.apply(pd.to_numeric).squeeze())
    return numeric

def get_center_series(
    survey: LimeSurveyData, center_code: str
) -> tuple["pd.Series[str]", Sequence[str]]:
    """Get a series that contains the center name for every participant.

    All centers *other than* `center_code` are replaced by "Other Centers".
    The output of this function can be nicely used with the histogram plot.

    Args:
        survey: The survey object
        center_code: ID of the center to filter by (like 'A01')

    Returns:
        Tuple of the Series and a 2-element list for center ordering.
    """
    center_name = shorten_center_name(survey.get_choices(CENTER)[center_code])
    assert center_name is not None

    # get the center question data, replace all "other" centers
    centers = survey.get_responses(CENTER)[CENTER].astype(str).rename("Center")
    centers.loc[~centers.isin([center_code])] = "Other Centers"
    centers.loc[centers.isin([center_code])] = center_name

    return centers, [center_name, "Other Centers"]

def get_data_for_q(survey: LimeSurveyData, question_number: str) -> pd.DataFrame:
    """Returns a DataFrame with the responses for the given question ID.

    Contingent questions (free text fields, shown as "other") are removed.

    Deprecated:
        Use `LimeSurveyData.get_responses(drop_other=True)` instead.

    Args:
        survey: The LimeSurvey object
        question_number: The question ID (like 'A1')

    Returns:
        A DataFrame with all answers to the specified question
    """
    warn(
        "get_data_for_q is deprecated, use survey.get_responses instead",
        DeprecationWarning,
        stacklevel=2,
    )

    responses = survey.get_responses(question_number, drop_other=True)
    # change types of all columns to object
    responses = responses.astype("object")
    # make id column by resetting the index
    responses = responses.reset_index()
    # change id column to type string
    responses = responses.astype({"id": "string"})
    return responses

def get_phd_duration(
    survey: LimeSurveyData,
) -> tuple["pd.Series[int]", "pd.Series[int]"]:
    """Calculate relevant durations from questions A8 and A9.

    We calculate:
    * How long has the participant been a doctoral researcher [years]?
    * How long do they estimate their project to last *in total* [months]?

    Args:
        survey: The survey object

    Returns:
        Tuple of current phd year and total duration estimation.
    """
    Q_START = {"year": "A8", "month": "A8a"}
    Q_END = {"year": "A9", "month": "A9a"}

    # we collected the data in April / May 2024
    SURVEY_YEAR = 2024
    SURVEY_MONTH = 5

    # filter out "before 2015" / I don't know / IDWA
    startyear = get_as_numeric(survey, Q_START["year"], ["A8", "A9", "A13"])
    startmonth = get_as_numeric(survey, Q_START["month"], ["A13", "A14"])
    endyear = get_as_numeric(survey, Q_END["year"], ["A11", "A12", "A13", "A14", "A15"])
    endmonth = get_as_numeric(survey, Q_END["month"], ["A13", "A14", "A15"])

    # calculate phd year relative to survey time (0 means not yet started)
    phd_current_month = (
        startyear.rsub(SURVEY_YEAR)
        .mul(12)
        .add(
            startmonth.rsub(SURVEY_MONTH),
        )
        .dropna()
        .astype(int)
    )
    phd_current_year = phd_current_month.floordiv(12).add(1)
    phd_current_year.clip(0, 6, inplace=True)  # clamp after 6 years
    phd_current_year = phd_current_year[phd_current_year != 0]  # remove year 0

    # calculate estimated total phd duration in months
    # if end month is missing, assume December
    phd_estimation_months = (
        endyear.sub(startyear)
        .mul(12)
        .add(
            endmonth.sub(startmonth, fill_value=12),
        )
        .dropna()
        .astype(int)
    )

    return (phd_current_year, phd_estimation_months)

def rate_burnout(responses: pd.DataFrame) -> pd.DataFrame:
    """Calculate burnout scores from participants' answers.

    This uses the MBI-GS scale according to the Maslach Burnout Inventory (MBI)
    Manual, Fourth Edition.

    Args:
        responses: responses to question D3d (burnout)

    Returns:
        SUM scores for each `Scale` (3 ints) and a burnout `Profile` (1 string)
    """
    SCORE_MAP = {
        "A2": 0,  # "Never"
        "A3": 1,  # "A few times a year or less"
        "A4": 2,  # "Once a month or less"
        "A5": 3,  # "A few times a month"
        "A6": 4,  # "Once a week"
        "A7": 5,  # "A few times a week"
        "A8": 6,  # "Every day"
    }

    scales = [
        Scale.EX,  # I feel emotionally drained from my work.
        Scale.EX,  # I feel used up at the end of the workday.
        Scale.EX,  # I feel tired when I get up in the morning and have to ...
        Scale.EX,  # Working all day is really a strain for me.
        Scale.PE,  # I can effectively solve the problems that arise in my work.
        Scale.EX,  # I feel burned out from my work.
        Scale.PE,  # I feel I am making an effective contribution to what ...
        Scale.CY,  # I have become less interested in my work since I ...
        Scale.CY,  # I have become less enthusiastic about my work.
        Scale.PE,  # In my opinion, I am good at my job.
        Scale.PE,  # I feel exhilarated when I accomplish something at work.
        Scale.PE,  # I have accomplished many worthwhile things in this job.
        Scale.CY,  # I just want to do my job and not be bothered.
        Scale.CY,  # I have become more cynical about whether my work ...
        Scale.CY,  # I doubt the significance of my work.
        Scale.PE,  # At my work, I feel confident that I am effective at ...
    ]

    # make empty df with three columns
    df = pd.DataFrame(responses.index)
    df[Scale.EX], df[Scale.CY], df[Scale.PE] = 0, 0, 0

    for col, scale in zip(responses.columns, scales, strict=True):
        # sum up the score in the relevant category
        col_scores = responses[col].map(SCORE_MAP, na_action="ignore")
        df[scale] += col_scores

    # boolean classification according to Table 3 in the manual
    # for PE, critical == good, hence the ">" instead of ">="
    df["EX_critical"] = df[Scale.EX].div(5).map(lambda x: x >= 2.90, na_action="ignore")
    df["CY_critical"] = df[Scale.CY].div(5).map(lambda x: x >= 2.86, na_action="ignore")
    df["PE_critical"] = df[Scale.PE].div(6).map(lambda x: x > 4.30, na_action="ignore")

    def classify(row: "pd.Series[Any]") -> Profile:
        """Assign burnout profiles according to Table 1 in the manual.

        Args:
            row: a single participant

        Returns:
            burnout `Profile` of the participant
        """
        exhausted = row["EX_critical"]
        cynical = row["CY_critical"]
        effective = row["PE_critical"]
        assert isinstance(exhausted, bool)
        assert isinstance(cynical, bool)
        assert isinstance(effective, bool)

        if not exhausted and not cynical and effective:
            return Profile.ENGAGED
        elif not exhausted and not cynical and not effective:
            return Profile.INEFFECTIVE
        elif exhausted and not cynical:
            return Profile.OVEREXTENDED
        elif not exhausted and cynical:
            return Profile.DISENGAGED
        elif exhausted and cynical:
            return Profile.BURNOUT
        else:
            raise AssertionError("unreachable")

    df["Profile"] = df.dropna().apply(classify, axis=1, result_type="reduce")
    return df

def rate_mental_health(
    responses: pd.DataFrame,
    condition: Condition,
    keep_subscores: bool = False,
) -> pd.DataFrame:
    """Calculate State/Trait Anxiety or Depression score based on responses to question.

    Scoring is based on:
      * K. Kroenke, R. L. Spitzer, J. B. W. William, and B. Löwe., The Patient
        Health Questionnaire somatic, anxiety,and depressive symptom scales:
        a systematic review. General Hospital Psychiatry, 32(4):345-359, 2010.
      * T. M. Marteau and H. Bekker., The development of a six-item short-form
        of the state scale of the spielberger state-trait anxiety inventory
        (STAI). British Journal of Clinical Psychology, 31(3):301-306, 1992.

    Args:
        responses: DataFrame containing responses data
        condition: Which kind of mental health condition to rate
        keep_subscores: Whether to include scores from subquestions
            in the output DataFrame, or only total score and classification.
            Default False.

    Returns:
        Mental health condition ratings ("score") and classifications ("class").
    """
    # Set up condition-specific parameters
    match condition:
        case Condition.STATE_ANXIETY:
            num_subquestions = 6
            base_score = 10 / 3
            conversion = ["pos", "neg", "neg", "pos", "pos", "neg"]
            label = "state_anxiety"
            classification_boundaries = [20, 40, 60, 80]
            classes = ["no or low anxiety", "moderate anxiety", "high anxiety"]

        case Condition.TRAIT_ANXIETY:
            num_subquestions = 8
            base_score = 5 / 2
            conversion = [
                "pos",
                "neg",
                "neg",
                "pos",
                "neg",
                "neg",
                "pos",
                "neg",
            ]
            label = "trait_anxiety"
            classification_boundaries = [20, 40, 60, 80]
            classes = ["no or low anxiety", "moderate anxiety", "high anxiety"]

        case Condition.DEPRESSION:
            num_subquestions = 8
            base_score = 1
            conversion = ["freq" for _ in range(8)]
            label = "depression"
            classification_boundaries = [0, 4, 9, 14, 19, 24]
            classes = [
                "no to minimal depression",
                "mild depression",
                "moderate depression",
                "moderately severe depression",
                "severe depression",
            ]

        case _:
            raise AssertionError("unreachable")

    # sanity check
    q_code = responses.columns[0].split("_")[0]
    if q_code != condition:
        raise ValueError(f"expected question {condition}, got {q_code}")

    # Set up score conversion dicts
    pos_direction_scores = {
        "A1": 4 * base_score,  # "Not at all
        "A2": 3 * base_score,  # "Somewhat"
        "A3": 2 * base_score,  # "Moderately
        "A4": 1 * base_score,  # "Very much"
    }
    neg_direction_scores = {
        "A1": 1 * base_score,  # "Not at all"
        "A2": 2 * base_score,  # "Somewhat"
        "A3": 3 * base_score,  # "Moderately"
        "A4": 4 * base_score,  # "Very much"
    }
    frequency_scores = {
        "A1": 0 * base_score,  # "Not at all"
        "A2": 1 * base_score,  # "Several days"
        "A3": 2 * base_score,  # "More than half the days"
        "A4": 3 * base_score,  # "Nearly every day"
    }
    conversion_dicts = {
        "pos": pos_direction_scores,
        "neg": neg_direction_scores,
        "freq": frequency_scores,
    }

    # Map responses from code to score
    df = pd.DataFrame()
    for column, conv in zip(responses.columns, conversion, strict=True):
        df[f"{column}_score"] = responses[column].map(
            conversion_dicts[conv], na_action="ignore"
        )

    # Calculate total anxiety or depression scores
    # scaled by number of non-NaN responses
    # e.g. scale by 8/5 if 5/8 subquestions answered
    responses_counts = df.notna().sum(axis=1)
    df[f"{label}_score"] = (
        df.sum(axis=1, skipna=True).div(responses_counts).mul(num_subquestions)
    )

    # Suppress entries with less than half of all subquestions answered
    # TODO: we might want to be more strict here
    df.loc[responses_counts < num_subquestions / 2, f"{label}_score"] = None

    # Classify into categories
    df[f"{label}_class"] = pd.cut(
        df[f"{label}_score"],
        bins=classification_boundaries,
        include_lowest=True,
        labels=classes,
    )

    if not keep_subscores:
        df = df.drop(df.columns[:-2], axis=1)

    return df

def rate_satisfaction(
    responses: pd.DataFrame,
    calc_average: bool = True,
) -> pd.DataFrame:
    """Calculate satisfaction rating for each subquestion and calculate the average.

    Uses a numeric scale from 1 = very dissatisfied to 5 = very satisfied.

    Args:
        responses: DataFrame containing responses data
        calc_average: Whether to calculate average satisfaction. Default True.

    Returns:
        Satisfaction ratings for each component (and overall average)
    """
    # safety check: only questions where we verified that they use a scale from
    # A1 (very satisfied) to A5 (very dissatisfied) should be added to this list
    SATISFACTION_QUESTIONS = ["C1"]

    q_code = responses.columns[0].split("_")[0]
    if q_code not in SATISFACTION_QUESTIONS:
        raise ValueError(f"{q_code} is not a satisfaction-scale question")

    # Set up score conversion dicts for individual questions
    satisfaction_question_scores = {
        "A1": 5.0,  # "Very satisfied"
        "A2": 4.0,  # "Satisfied"
        "A3": 3.0,  # "Neither/nor"
        "A4": 2.0,  # "Dissatisfied"
        "A5": 1.0,  # "Very dissatisfied"
    }

    # Map responses to score
    df = pd.DataFrame()
    for column in responses.columns:
        df[f"{column}_score"] = responses[column].map(
            satisfaction_question_scores, na_action="ignore"
        )

    # Calculate mean rating and round (ignoring NaN)
    if calc_average:
        df[f"{q_code}_score"] = df.mean(axis=1, skipna=True).round()
        df[f"{q_code}_class"] = df[f"{q_code}_score"].map(
            {v: k for k, v in satisfaction_question_scores.items()}
        )

    return df