Function apply_scoring_rules

scoring/src/scoring/scoring_rules.py:1646–1743

Apply a list of ScoringRules to note inputs and return noteStats augmented with scoring results. This function applies a list of ScoringRules in order. Once each rule has run a final ratingStatus is set for each note. An additional column is added to capture which rules acted on the note and

(
  noteStats: pd.DataFrame,
  rules: List[ScoringRule],
  statusColumn: str,
  ruleColumn: str,
  decidedByColumn: Optional[str] = None,
)

def apply_scoring_rules(
  noteStats: pd.DataFrame,
  rules: List[ScoringRule],
  statusColumn: str,
  ruleColumn: str,
  decidedByColumn: Optional[str] = None,
) -> pd.DataFrame:
  """Apply a list of ScoringRules to note inputs and return noteStats augmented with scoring results.

  This function applies a list of ScoringRules in order.  Once each rule has run
  a final ratingStatus is set for each note. An additional column is added to capture
  which rules acted on the note and any additional columns generated by the ScoringRules
  are merged with the scored notes to generate the final return value.

  Args:
    noteStats: attributes, aggregates and raw scoring signals for each note.
    rules: ScoringRules which will be applied in the order given.
    statusColumn: str indicating column where status should be assigned.
    ruleColumn: str indicating column where active rules should be stored.
    decidedByColumn: None or str indicating column where the last rule to act on a note is stored.

  Returns:
    noteStats with additional columns representing scoring results.
  """
  # Initialize empty dataframes to store labels for each note and which rules impacted
  # scoring for each note.
  noteLabels = pd.DataFrame.from_dict(
    {c.noteIdKey: pd.Series([], dtype=np.int64), statusColumn: pd.Series([], dtype=object)}
  )
  noteRules = pd.DataFrame.from_dict(
    {c.noteIdKey: pd.Series([], dtype=np.int64), ruleColumn: pd.Series([], dtype=object)}
  )
  noteColumns = pd.DataFrame.from_dict({c.noteIdKey: pd.Series([], dtype=np.int64)})

  # Establish state to enforce rule dependencies.
  ruleIDs: Set[RuleID] = set()

  # Successively apply each rule
  for rule in rules:
    with c.time_block(f"Applying scoring rule: {rule.get_name()}"):
      logger.info(f"Applying scoring rule: {rule.get_name()}")
      rule.check_dependencies(ruleIDs)
      assert rule.get_rule_id() not in ruleIDs, f"repeat ruleID: {rule.get_name()}"
      ruleIDs.add(rule.get_rule_id())
      with c.time_block(f"Calling score_notes: {rule.get_name()}"):
        noteStatusUpdates, additionalColumns = rule.score_notes(noteStats, noteLabels, statusColumn)
      if (
        additionalColumns is not None
        # This rule updates both status and NmrDueToStableCrhTime (in additional column), they can
        # be on different rows.
        and rule.get_rule_id() != RuleID.NMR_DUE_TO_MIN_STABLE_CRH_TIME
      ):
        assert set(noteStatusUpdates[c.noteIdKey]) == set(additionalColumns[c.noteIdKey])

      # Update noteLabels, which will always hold at most one label per note.
      noteLabels = pd.concat([noteLabels, noteStatusUpdates]).groupby(c.noteIdKey).tail(1)
      # Update note rules to have one row per rule which was active for a note
      noteRules = pd.concat(