def meta_score(
  scorers: Dict[Scorers, List[Scorer]],
  scoredNotes: pd.DataFrame,
  auxiliaryNoteInfo: pd.DataFrame,
  noteStatusHistory: pd.DataFrame,
  enabledScorers: Optional[Set[Scorers]],
  enableNmrDueToMinStableCrhTime: bool = True,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
  """Determine final note status based on individual scoring results.

  This function applies logic merging the results of individual scorers to determine
  the final rating status of each note.  As part of determining the final rating status,
  we also apply scoring logic which exists independent of individual scorers (specifically
  note tag assignment and status locking).

  Args:
    scoredNotes: pd.DataFrame containing all scored note results.
    auxiliaryNoteInfo: pd.DataFrame containing tag aggregates
    noteStatusHistory: pd.DataFrame containing {noteId, lockedStatus, timestampMillisOfNmrDueToMinStableCrhTime} for all notes
    enabledScorers: if not None, set of which scorers should be instantiated and enabled

  Returns:
    Tuple[pd.DataFrame, pd.DataFrame]:
      scoredNotesCols pd.DataFrame: one row per note contained note scores and parameters.
      auxiliaryNoteInfoCols pd.DataFrame: one row per note containing adjusted and ratio tag values
  """
  # Temporarily merge helpfulness tag aggregates into scoredNotes so we can run InsufficientExplanation
  with c.time_block("Post-scorers: Meta Score: Setup"):
    assert len(scoredNotes) == len(auxiliaryNoteInfo)
    scoredNotes = scoredNotes.merge(
      auxiliaryNoteInfo[
        [c.noteIdKey, c.currentLabelKey] + c.helpfulTagsTSVOrder + c.notHelpfulTagsTSVOrder
      ],
      on=c.noteIdKey,
    )
    scoredNotes = scoredNotes.merge(
      noteStatusHistory[
        [
          c.noteIdKey,
          c.timestampMillisOfNmrDueToMinStableCrhTimeKey,
          c.firstNonNMRLabelKey,
        ]
      ],
      on=c.noteIdKey,
    )
    assert len(scoredNotes) == len(auxiliaryNoteInfo)
    rules: List[scoring_rules.ScoringRule] = [
      scoring_rules.DefaultRule(RuleID.META_INITIAL_NMR, set(), c.needsMoreRatings)
    ]
    # Only attach meta-scoring rules for models which actually run.
    if enabledScorers is None or Scorers.MFExpansionPlusScorer in enabledScorers:
      # The MFExpansionPlusScorer should score a disjoint set of notes from MFExpansionScorer
      # and MFCoreScorer because it should score notes by EXPANSION_PLUS writers and should be
      # the only model to score notes by EXPANSION_PLUS writers.  This ordering is safe, where if
      # there is any bug and a note is scored by MFExpansionPlusScorer and another scorer, then
      # MFExpansionPlusScorer will have the lowest priority.
      rules.append(
        scoring_rules.ApplyModelResult(
          RuleID.EXPANSION_PLUS_MODEL,
          {RuleID.META_INITIAL_NMR},
          c.expansionPlusRatingStatusKey,
        )
      )
    if enabledScorers is None or Scorers.MFExpansionScorer in enabledScorers:
      rules.append(
        scoring_rules.ApplyModelResult(
          RuleID.EXPANSION_MODEL,
          {RuleID.META_INITIAL_NMR},
          c.expansionRatingStatusKey,
        )
      )
    if enabledScorers is None or Scorers.MFCoreWithTopicsScorer in enabledScorers:
      rules.append(
        scoring_rules.ApplyModelResult(
          RuleID.CORE_WITH_TOPICS_MODEL,
          {RuleID.META_INITIAL_NMR},
          c.coreWithTopicsRatingStatusKey,
        )
      )
    if enabledScorers is None or Scorers.MFCoreScorer in enabledScorers:
      rules.append(
        scoring_rules.ApplyModelResult(
          RuleID.CORE_MODEL,
          {RuleID.META_INITIAL_NMR},
          c.coreRatingStatusKey,
        )
      )

    if enabledScorers is None or Scorers.MFMultiGroupScorer in enabledScorers:
      rules.append(
        scoring_rules.ApplyModelResult(
          RuleID["MULTI_GROUP_MODEL_1"],
          {RuleID.CORE_MODEL},
          c.multiGroupRatingStatusKey,
          checkFirmReject=True,
          filterColumnPairs=[(c.modelingMultiGroupKey, 1)],
        )
      )
    if enabledScorers is None or Scorers.MFGroupScorer in enabledScorers:
      # TODO: modify this code to work when MFExpansionScorer is disabled by the system test
      assert len(scorers[Scorers.MFCoreScorer]) == 1
      assert len(scorers[Scorers.MFExpansionScorer]) == 1
      coreScorer = scorers[Scorers.MFCoreScorer][0]
      assert isinstance(coreScorer, MFCoreScorer)
      expansionScorer = scorers[Scorers.MFExpansionScorer][0]
      assert isinstance(expansionScorer, MFExpansionScorer)
      coreCrhThreshold = coreScorer.get_crh_threshold()
      expansionCrhThreshold = expansionScorer.get_crh_threshold()
      # consecutive group scorers
      for i in range(1, groupScorerCount + 1):
        if i != trialScoringGroup:
          rules.append(
            scoring_rules.ApplyGroupModelResult(
              RuleID[f"GROUP_MODEL_{i}"],
              {RuleID.EXPANSION_MODEL, RuleID.CORE_MODEL},
              i,
              coreCrhThreshold,
              expansionCrhThreshold,
            )
          )
        else:
          rules.append(
            scoring_rules.ApplyGroupModelResult(
              RuleID[f"GROUP_MODEL_{i}"],
              {RuleID.EXPANSION_MODEL, RuleID.CORE_MODEL},
              i,
              None,
              None,
              minSafeguardThreshold=0.25,
            )
          )
      rules.append(
        scoring_rules.ApplyGroupModelResult(
          RuleID[f"GROUP_MODEL_{nmrScoringGroup}"],
          {RuleID.EXPANSION_MODEL, RuleID.CORE_MODEL},
          nmrScoringGroup,
          None,
          None,
          minSafeguardThreshold=0.25,
        )
      )
      # nmr group scorer
      rules.append(
        scoring_rules.ApplyNMRGroupModelResult(
          RuleID[f"GROUP_MODEL_{nmrScoringGroup}_NMR"],
          {RuleID.EXPANSION_MODEL, RuleID.CORE_MODEL},
          nmrScoringGroup,
        )
      )
    if enabledScorers is None or Scorers.MFTopicScorer in enabledScorers:
      for topic in Topics:
        if topic == Topics.Unassigned:
          continue
        rules.append(
          scoring_rules.ApplyTopicModelResult(
            RuleID[f"TOPIC_MODEL_{topic.value}"],
            {RuleID.EXPANSION_PLUS_MODEL, RuleID.EXPANSION_MODEL, RuleID.CORE_MODEL},
            topic,
          )
        )
    if enableNmrDueToMinStableCrhTime:
      rules.append(
        scoring_rules.NmrDueToMinStableCrhTime(
          RuleID.NMR_DUE_TO_MIN_STABLE_CRH_TIME,
          {RuleID.CORE_MODEL},
        )
      )
    rules.extend(
      [
        scoring_rules.ScoringDriftGuard(
          RuleID.SCORING_DRIFT_GUARD, {RuleID.CORE_MODEL}, noteStatusHistory
        ),
        # TODO: The rule below both sets tags for notes which are CRH / CRNH and unsets status for
        # any notes which are CRH / CRNH but don't have enough ratings to assign two tags.  The later
        # behavior can lead to unsetting locked status.  We should refactor this code to (1) remove
        # the behavior which unsets status (instead tags will be assigned on a best effort basis) and
        # (2) set tags in logic which is not run as a ScoringRule (since ScoringRules function to
        # update note status).
        scoring_rules.InsufficientExplanation(
          RuleID.INSUFFICIENT_EXPLANATION,
          {RuleID.CORE_MODEL},
          c.needsMoreRatings,
          c.minRatingsToGetTag,
          c.minTagsNeededForStatus,
        ),
      ]
    )
    scoredNotes[c.firstTagKey] = np.nan
    scoredNotes[c.secondTagKey] = np.nan

  with c.time_block("Post-scorers: Meta Score: Apply Scoring Rules"):
    scoringResult = scoring_rules.apply_scoring_rules(
      scoredNotes,
      rules,
      c.finalRatingStatusKey,
      c.metaScorerActiveRulesKey,
      decidedByColumn=c.decidedByKey,
    )
    if not enableNmrDueToMinStableCrhTime:
      scoringResult[c.updatedTimestampMillisOfNmrDueToMinStableCrhTimeKey] = np.nan
      scoringResult[c.preStabilizationRatingStatusKey] = np.nan
    # Validate that nothing that was a FIRM_REJECT or CRNH from Core or Expansion is rated CRH
    coreRejects = scoringResult[c.coreRatingStatusKey].isin(
      {c.firmReject, c.currentlyRatedNotHelpful}
    )
    expansionRejects = scoringResult[c.expansionRatingStatusKey].isin(
      {c.firmReject, c.currentlyRatedNotHelpful}
    )
    blockedRows = coreRejects | (scoringResult[c.coreRatingStatusKey].isna() & expansionRejects)
    crhRows = scoringResult[c.finalRatingStatusKey] == c.currentlyRatedHelpful
    logger.info("Summary of blocked and CRH rows:")
    # TODO: validate that these are all due to ScoringDriftGuard and change to an assert
    logger.info(
      scoringResult[blockedRows & crhRows][c.metaScorerActiveRulesKey].value_counts(dropna=False)
    )
    logger.info(scoringResult[blockedRows & crhRows][c.decidedByKey].value_counts(dropna=False))
  with c.time_block("Post-scorers: Meta Score: Preparing Return Values"):
    scoredNotesCols = scoringResult[
      [
        c.noteIdKey,
        c.finalRatingStatusKey,
        c.metaScorerActiveRulesKey,
        c.firstTagKey,
        c.secondTagKey,
        c.decidedByKey,
        c.updatedTimestampMillisOfNmrDueToMinStableCrhTimeKey,
      ]
    ]
    auxiliaryNoteInfoCols = scoringResult[
      [
        c.noteIdKey,
        c.currentlyRatedHelpfulBoolKey,
        c.currentlyRatedNotHelpfulBoolKey,
        c.awaitingMoreRatingsBoolKey,
        c.unlockedRatingStatusKey,
        c.preStabilizationRatingStatusKey,
      ]
    ]
  return scoredNotesCols, auxiliaryNoteInfoCols
Determine final note status based on individual scoring results.

  This function applies logic merging the results of individual scorers to determine
  the final rating status of each note.  As part of determining the final rating status,
  we also apply scoring logic which exists independent of individual scorers (specifically
  note tag assignment and status locking).

  Args:
    scoredNotes: pd.DataFrame containing all scored note results.
    auxiliaryNoteInfo: pd.DataFrame containing tag aggregates
    noteStatusHistory: pd.DataFrame containing {noteId, lockedStatus, timestampMillisOfNmrDueToMinStableCrhTime} for all notes
    enabledScorers: if not None, set of which scorers should be instantiated and enabled

  Returns:
    Tuple[pd.DataFrame, pd.DataFrame]:
      scoredNotesCols pd.DataFrame: one row per note contained note scores and parameters.
      auxiliaryNoteInfoCols pd.DataFrame: one row per note containing adjusted and ratio tag values
  
sourcecode/scoring/run_scoring.py