diff --git a/medcat/medcat/components/linking/context_based_linker.py b/medcat/medcat/components/linking/context_based_linker.py index 2813d1744..736f78a00 100644 --- a/medcat/medcat/components/linking/context_based_linker.py +++ b/medcat/medcat/components/linking/context_based_linker.py @@ -169,6 +169,7 @@ def _process_entity_inference( # Check does it have a detected concepts cuis = entity.link_candidates if not cuis: + logger.debug("No candidates") return # Check does it have a detected name name = entity.detected_name diff --git a/medcat/medcat/stats/common.py b/medcat/medcat/stats/common.py new file mode 100644 index 000000000..3a496c712 --- /dev/null +++ b/medcat/medcat/stats/common.py @@ -0,0 +1,37 @@ +from typing import TypedDict + +from medcat.tokenizing.tokens import MutableEntity + + +NER_DETECTED_ENTITY_TAG = "DETECTED_ENTITY" + + +class GoldAnnotation(TypedDict): + """Validated gold annotation payload after CUI filtering.""" + + start: int + end: int + cuis: list[str] + cui: str + text: str + raw: object + document_id: int + document_name: str + # context is 60 chars before + after the entity + context: str + + +class PredictedAnnotation(TypedDict): + """Predicted entity payload used for scoring and metrics.""" + + start: int + end: int + cui: str + text: str + confidence: float + raw: MutableEntity + no_tokens: int + document_id: int + document_name: str + # context is 60 chars before + after the entity + context: str diff --git a/medcat/medcat/stats/failuremodes.py b/medcat/medcat/stats/failuremodes.py new file mode 100644 index 000000000..95ac9ebe1 --- /dev/null +++ b/medcat/medcat/stats/failuremodes.py @@ -0,0 +1,604 @@ +from enum import Enum +from pprint import pprint +from typing import Callable, Collection, Literal, TypedDict +from itertools import product +import logging +from collections import Counter, defaultdict + +from medcat.cat import CAT +from medcat.cdb.concepts import CUIInfo, NameInfo +from medcat.config import LinkingFilters +from medcat.tokenizing.tokens import MutableDocument, MutableToken +from medcat.stats.common import NER_DETECTED_ENTITY_TAG, PredictedAnnotation +from medcat.utils.cdb_utils import reverse_pt2ch + + +logger = logging.getLogger(__name__) + + +class FailureMode(str, Enum): + """Why a gold-standard annotation was not predicted (false negative). + + Each FN is assigned exactly ONE mode. Evaluated in this order; the first + matching mode wins: + + 0. gold sanity (SPAN_INCORRECT_FOR_VALUE) + 1. concept / name lookup (CUI_NOT_IN_CDB .. NAME_NOT_LINKED_TO_CUI) + 2. NER (NER_NO_SPAN) + 3. filters (FILTERED_*) + 4. disambiguation (WRONG_CONCEPT_*) + 5. span (SPAN_*) + 6. thresholding (BELOW_THRESHOLD) + 7. fallback (UNKNOWN) + + Filters are checked before disambiguation because a filtered-out gold + concept is removed from the candidates, which would otherwise show up as + a (misleading) disambiguation failure. + + Hierarchy-based modes need a concept hierarchy (e.g. SNOMED is-a). With + no hierarchy (custom ontology) use WRONG_CONCEPT_UNRELATED. + """ + + # --- 0. Gold sanity ----------------------------------------------- + SPAN_INCORRECT_FOR_VALUE = "span_incorrect_for_value" + """The span start/end index did not match the value specified by the + annotation. This is normally due to some preprocessing somewhere. Not a + model failure, and nothing else can be said about such an annotation.""" + + # --- 1. Concept / name lookup ------------------------------------- + CUI_NOT_IN_CDB = "cui_not_in_cdb" + """The gold concept does not exist in the CDB at all, so no name can + reach it (e.g. CDB built from a subset of the ontology).""" + + NAME_UNKNOWN = "name_unknown" + """The surface text (after normalisation) is not a known name in the CDB + for any concept.""" + + NAME_NOT_LINKED_TO_CUI = "name_not_linked_to_cui" + """The name is in the CDB, but only maps to other concepts. The gold + concept was never a candidate, so disambiguation could not select it.""" + + # --- 2. NER --------------------------------------------------------- + NER_NO_SPAN = "ner_no_span" + """The name is in the CDB and maps to the gold concept, but the NER step + never proposed a span. Typical causes: tokenisation boundary mismatch, + normalisation, case sensitivity (e.g. abbreviations), min name length, + or too many skipped tokens in a multi-word name.""" + + # --- 3. Filters ----------------------------------------------------- + FILTERED_DISALLOW_LIST = "filtered_disallow_list" + """The gold concept is explicitly excluded by a disallow list / exclude + filter.""" + + FILTERED_NOT_IN_ALLOW_LIST = "filtered_not_in_allow_list" + """An allow list / include filter is active and the gold concept is not + in it.""" + + # --- 4. Disambiguation (gold concept was a candidate, other chosen) -- + WRONG_CONCEPT_DIRECT_PARENT = "wrong_concept_direct_parent" + """The direct parent of the concept was predicted instead of the expected + (more specific) gold concept.""" + WRONG_CONCEPT_ANCESTOR = "wrong_concept_ancestor" + """A more general concept (ancestor of the gold concept) was predicted + instead. Record the hierarchy distance as metadata.""" + + WRONG_CONCEPT_DIRECT_CHILD = "wrong_concept_direct_child" + """The direct child of the concept was predicted instead of the expected + (more general) gold concept.""" + + WRONG_CONCEPT_DESCENDANT = "wrong_concept_descendant" + """A more specific concept (descendant of the gold concept) was predicted + instead. Record the hierarchy distance as metadata.""" + + WRONG_CONCEPT_SIBLING = "wrong_concept_sibling" + """A concept sharing a direct parent with the gold concept was predicted + instead.""" + + WRONG_CONCEPT_UNRELATED = "wrong_concept_unrelated" + """A concept with no known hierarchical relation to the gold concept was + predicted instead (or no hierarchy is available).""" + + # --- 5. Span -------------------------------------------------------- + SPAN_PRED_CONTAINS_GOLD = "span_pred_contains_gold" + """Right concept, but the predicted span is longer: it fully contains the + gold span.""" + + SPAN_GOLD_CONTAINS_PRED = "span_gold_contains_pred" + """Right concept, but the predicted span is shorter: it lies fully inside + the gold span.""" + + SPAN_PARTIAL_OVERLAP = "span_partial_overlap" + """Right concept, but the predicted span is shifted: it overlaps the gold + span without either containing the other.""" + + # --- 6. Thresholding ----------------------------------------------- + BELOW_THRESHOLD = "below_threshold" + """The gold concept was not chosen due to its linking score being below + the acceptance threshold. We cannot tell for certain if there were other + concepts that were higher in linking score at this stage. Store the score + margin and the concept's training count as metadata to tell 'needs more + training data' apart from 'close to the threshold'.""" + + # --- 7. Fallback ---------------------------------------------------- + UNKNOWN = "unknown" + """No mode above explains the miss (includes possible gold-annotation + errors such as bad offsets or labels).""" + + def __str__(self) -> str: + return f"FM.{self.name}" + + def __repr__(self) -> str: + return f"<{str(self)}>" + + +def get_ner_candidates( + doc: MutableDocument, + tkns: list[MutableToken], +) -> list[tuple[int, int, Collection[str]]]: + start_char_index = tkns[0].base.char_index + end_char_index = tkns[-1].base.char_index + len(tkns[-1].base.text) + return [ + (ent.base.start_char_index, ent.base.end_char_index, + ent.link_candidates) + for ent in doc.ner_ents + if start_char_index <= ent.base.start_char_index < end_char_index + ] + + +def _get_local_span( + context: str, + start: int, end: int, + source_value: str, + window_size: int, +) -> tuple[int, int]: + span_len = end - start + # if in the middle of some text + if len(context) == window_size + span_len + window_size: + return window_size, window_size + span_len + # harder case where we're not in the middle of text so the span + # we're looking for could be at the start or end of the context + l_context = context.lower() + l_source = source_value.lower() + occurances = l_context.count(l_source) + if occurances == 1: + cstart = l_context.find(l_source) + return cstart, cstart + span_len + elif occurances == 0: + # does not exist in here + # NOTE: this can detract from SPAN_INCORRECT_FOR_VALUE + # but only in cases where the span is at the start + # or end of the text, otherwise the length based + # check should have caught it anyway + return -2, -1 + # even harder since there are multiple places in there + # I will just get the one that's closest to the middle of + # the context... + logger.info( + "Having trouble getting local span detials for %s due to the text " + "having %d instances of the string within it. We will try to pick " + "the one that's closest to the centre of the context, but that is " + "not guaranteed to be correct either." + ) + cur_start = 0 + spans: list[tuple[int, int]] = [] + while (new_start := l_context.find(l_source, cur_start)) != -1: + cur_end = new_start + len(l_source) + spans.append((new_start, cur_end)) + cur_start = cur_end + midpoint = len(l_context) + return min(spans, key=lambda item: abs(sum(item) - midpoint)) + + +# TODO: if/when #622 is merged, use method from there +def _build_opts(tkns: list[MutableToken], separator: str) -> set[str]: + per_tkn_opts: list[list[str]] = [] + for tkn in tkns: + name_versions = tkn.base.text_versions + # NOTE: I want to preserve order since the text versions returns in + # the correct / expected order in which to check + unique_versions = [] + # NOTE: we want to avoid duplicate names so we only keep uinque ones + # otherwise the option set can become massive + for version in name_versions: + # NOTE: we're checking contents of a list, not ideal + # but we should only ever have 0 or 1 items in the list + # at time of check so shouldn't be too bad + if version not in unique_versions: + unique_versions.append(version) + per_tkn_opts.append(unique_versions) + opt_set = set(product(*per_tkn_opts)) + return set(map(separator.join, opt_set)) + + +def _get_failure_mode_for_partial_span( + gold_cui: str, gold_start: int, gold_end: int, + pred_cui: str, pred_start: int, pred_end: int, +) -> FailureMode | None: + if gold_cui != pred_cui: + return None + if pred_end <= gold_start or gold_end <= pred_start: + return None + if pred_start <= gold_start and gold_end <= pred_end: + # NOTE: identical spans would be a TP, so never reach here as FN + return FailureMode.SPAN_PRED_CONTAINS_GOLD + if gold_start <= pred_start and pred_end <= gold_end: + return FailureMode.SPAN_GOLD_CONTAINS_PRED + return FailureMode.SPAN_PARTIAL_OVERLAP + + +def _get_partially_overlapping_spans( + example: dict, + predictions: list[PredictedAnnotation], +) -> list[tuple[PredictedAnnotation, FailureMode]]: + gold_cui = example['cui'] + gold_start, gold_end = example['start'], example['end'] + return [ + (pred, fm) for pred in predictions + if (fm := _get_failure_mode_for_partial_span( + gold_cui, gold_start, gold_end, + pred['cui'], pred['start'], pred['end'])) + ] + + +class FailureModeFinder: + + def __init__( + self, + cui2info: dict[str, CUIInfo], + name2info: dict[str, NameInfo], + pt2ch: dict[str, list[str]], + linking_filters: LinkingFilters, + token_separator: str, + ) -> None: + self.cui2info = cui2info + self.name2info = name2info + self.pt2ch = pt2ch + self._ch2pt: dict[str, list[str]] | None = None + self.linking_filters = linking_filters + self.token_separator = token_separator + + @property + def ch2pt(self) -> dict[str, list[str]]: + if self._ch2pt is None: + self._ch2pt = reverse_pt2ch(self.pt2ch) + return self._ch2pt + + @classmethod + def from_cat(cls, cat: CAT) -> 'FailureModeFinder': + return cls( + cat.cdb.cui2info, cat.cdb.name2info, + cat.cdb.addl_info['pt2ch'], cat.config.components.linking.filters, + cat.config.general.separator + ) + + def _ancestors(self, cui: str) -> set[str]: + """All ancestors of a concept. Safe against cycles.""" + seen: set[str] = set() + stack = list(self.ch2pt.get(cui, [])) + while stack: + cur = stack.pop() + if cur in seen: + continue + seen.add(cur) + stack.extend(self.ch2pt.get(cur, [])) + return seen + + def _get_disamb_failure_mode( + self, + gold_cui: str, + pred_cui: str, + ) -> FailureMode: + """Relation of the predicted concept to the gold one. + + Only needs the child -> parents map. Walks UPWARDS from both concepts + (parents are few; descendants of a general concept can be huge). + """ + gold_parents = set(self.ch2pt.get(gold_cui, [])) + pred_parents = set(self.ch2pt.get(pred_cui, [])) + if pred_cui in gold_parents: + return FailureMode.WRONG_CONCEPT_DIRECT_PARENT + if gold_cui in pred_parents: + return FailureMode.WRONG_CONCEPT_DIRECT_CHILD + if gold_parents & pred_parents: + return FailureMode.WRONG_CONCEPT_SIBLING + if pred_cui in self._ancestors(gold_cui): + return FailureMode.WRONG_CONCEPT_ANCESTOR + if gold_cui in self._ancestors(pred_cui): + return FailureMode.WRONG_CONCEPT_DESCENDANT + return FailureMode.WRONG_CONCEPT_UNRELATED + + def _step_0_gold_sanity( + self, context: str, start: int, end: int, + source_val: str, + ) -> FailureMode | None: + if context[start: end] != source_val: + return FailureMode.SPAN_INCORRECT_FOR_VALUE + return None + + def _step_1_concept_and_name_lookup( + self, + tkns: list[MutableToken], + gold_cui: str, + ) -> FailureMode | None: + name_versions = _build_opts(tkns, self.token_separator) + suitable_names = [ + name for name in name_versions if name in self.name2info] + # any name variant linking to the gold concept is enough + candidates: set[str] = set() + for name in suitable_names: + candidates.update(self.name2info[name]['per_cui_status']) + # NOTE: for NER-only stats the concept ID is a special one + # since the assumption is that the perfect linker knows + # every concept + if gold_cui not in self.cui2info and gold_cui != NER_DETECTED_ENTITY_TAG: + return FailureMode.CUI_NOT_IN_CDB + if not suitable_names: + return FailureMode.NAME_UNKNOWN + # NOTE: for NER-only stats the perfect-linker should have all + # names and concepts linked so we ignore this failure mode + if gold_cui not in candidates and gold_cui != NER_DETECTED_ENTITY_TAG: + return FailureMode.NAME_NOT_LINKED_TO_CUI + return None + + def _step_2_ner( + self, mut_doc: MutableDocument, tkns: list[MutableToken], + ) -> FailureMode | None: + if not get_ner_candidates( + mut_doc, tkns): + return FailureMode.NER_NO_SPAN + return None + + def _step_3_filters( + self, gold_cui: str, + ) -> FailureMode | None: + allow_filter = self.linking_filters.cuis + disallow_filter = self.linking_filters.cuis_exclude + if gold_cui == NER_DETECTED_ENTITY_TAG: + # NOTE: for NER terms there is no filter + return None + if allow_filter and gold_cui not in allow_filter: + return FailureMode.FILTERED_NOT_IN_ALLOW_LIST + if gold_cui in disallow_filter: + return FailureMode.FILTERED_DISALLOW_LIST + return None + + def _step_4_disambiguation( + self, span_predictions: list[PredictedAnnotation], + gold_cui: str, source_val: str, + ) -> FailureMode | None: + if span_predictions: + modes = { + self._get_disamb_failure_mode(gold_cui, pred["cui"]) + for pred in span_predictions + } + picked = min(modes, key=lambda m: m.name) + if len(modes) > 1: + logger.warning( + "Unable to determine a single point of failure for %s " + "(%s). Available: %s. Picking the first alphabetically " + "(%s)", + gold_cui, source_val, [fm.name for fm in modes], + picked.name, + ) + return picked + return None + + def _step_5_partial_overlap( + self, example: dict, all_predictions: list[PredictedAnnotation], + ) -> FailureMode | None: + partially_matching_spans = _get_partially_overlapping_spans( + example, all_predictions) + fm_priority = [ + FailureMode.SPAN_PARTIAL_OVERLAP, + FailureMode.SPAN_GOLD_CONTAINS_PRED, + FailureMode.SPAN_PRED_CONTAINS_GOLD, + ] + if partially_matching_spans: + cntr: Counter[FailureMode] = Counter( + fm for _, fm in partially_matching_spans) + if len(cntr) > 1: + if example['cui'] == NER_DETECTED_ENTITY_TAG: + # NOTE: all overlapping entities will match due to + # all of them having the same CUI so we will + # prioritise the least "accurate" here since + # we cannot fully be sure + most_common = sorted(cntr.keys(), key=fm_priority.index)[0] + else: + most_common = cntr.most_common(1)[0][0] + logger.warning( + "Got multiple types of partial failure modes. " + "Using most common (%s), available: %s", + most_common, cntr + ) + return most_common + return next(iter(cntr.keys())) + return None + + def get_failure_mode( + self, + mut_doc: MutableDocument, + example: dict, + span_predictions: list[PredictedAnnotation], + all_predictions: list[PredictedAnnotation], + window_size: int = 60, + ) -> FailureMode: + """Get the failure mode of a specific example. + + Args: + example (dict): The example to consider. + span_predictions (list[PredictedAnnotation]): Predicted + annotations for the same span (i.e disambiguation mistakes) + all_predictions (list[PredictedAnnotation]): All predicted + annotations for the text. + window_size (int, optional): The contex window size. + Defaults to 60. + + Returns: + FailureMode: The resulting failure mode. + """ + context = mut_doc.base.text + gold_cui = example['cui'] + start = example['start'] + end = example['end'] + source_val = example['source_value'] + + tkns = [ + tkn for tkn in mut_doc + if (not tkn.to_skip and + example['start'] <= tkn.base.char_index < example['end'] + ) + ] + + return ( + self._step_0_gold_sanity(context, start, end, source_val) + or self._step_1_concept_and_name_lookup(tkns, gold_cui) + or self._step_2_ner(mut_doc, tkns) + or self._step_3_filters(gold_cui) + or self._step_4_disambiguation( + span_predictions, gold_cui, source_val) + or self._step_5_partial_overlap(example, all_predictions) + # 6. thresholding: everything that got here had a valid name->cui + # mapping, NER proposed the span, and nothing else was + # annotated, so the linker must have rejected it. + or FailureMode.BELOW_THRESHOLD + ) + + +class ExampleDescription(TypedDict): + project_id: str + project_name: str + document_id: str + document_name: str + context: str + source_val: str + + +FailureCountOrDescription = dict[FailureMode, int | list[ExampleDescription]] +PrintMode = Literal['none', 'totals', 'per_type_id', 'per_cui', 'all'] + + +class FailureModeSummary(TypedDict): + per_type_id_counts: dict[str, FailureCountOrDescription] + per_cui_counts: dict[str, FailureCountOrDescription] + total_counts: FailureCountOrDescription + + +def summarise_failure_modes( + cat: CAT, + examples: dict[str, dict[str, list[dict]]], + print_mode: PrintMode = 'none', + include_details_for_examples_max: int = 0, +): + """Summarise failure modes for all examples. + + This allows one to get a summary of all the failure modes in examples. + + NOTE: Only false negative examples (`fns`) are used here. Those are the + only ones for which the current setup makes sense. + + Args: + cat (CAT): The model pack. This is used for getting Type IDs. + examples (dict[str, dict[str, list[dict]]]): All examples + print_mode (PrintMode): Determines what to print out. + Defaults to 'none'. + include_details_for_examples_max (int): Whether to include details + about each example up to a certain maximum. Defaults to 0. + """ + def get_cui_type_ids(cui: str) -> set[str]: + if cui not in cat.cdb.cui2info: + return set() + return cat.cdb.cui2info[cui]['type_ids'] + return _summarise_failure_modes( + examples, + get_cui_type_ids, + print_mode=print_mode, + include_details_for_examples_max=include_details_for_examples_max, + ) + + +def _to_description(example: dict) -> ExampleDescription: + context = example['text'] + source_val = example['source_value'] + start, end = example['start'], example['end'] + start, end = _get_local_span(context, start, end, source_val, 60) + context = f"{context[:start]}>>>{context[start:end]}<<<{context[end:]}" + return { + 'project_id': example['project_id'], + 'project_name': example['project_name'], + 'document_id': example['document_id'], + 'document_name': example['document_name'], + 'context': context, + 'source_val': source_val, + } + + +def _collapse( + by_mode: dict[FailureMode, list[dict]], + max_details: int, +) -> FailureCountOrDescription: + # with max_details=0 this always yields counts, as before + return { + mode: ([_to_description(ex) for ex in exs] + if len(exs) <= max_details else len(exs)) + for mode, exs in by_mode.items() + } + + +def _summarise_failure_modes( + examples: dict[str, dict[str, list[dict]]], + cui2type_ids: dict[str, set[str]] | Callable[[str], set[str]], + print_mode: Literal[ + 'none', 'totals', 'per_type_id', 'per_cui', 'all'] = 'none', + include_details_for_examples_max: int = 0, +) -> FailureModeSummary: + def cui2type_ids_call(cui: str) -> set[str]: + if not callable(cui2type_ids): + return cui2type_ids.get(cui, set()) + return cui2type_ids(cui) + fns = examples['fn'] + + per_cui_examples: dict[str, dict[FailureMode, list[dict]]] = {} + for cui, cui_examples in fns.items(): + by_mode: defaultdict[FailureMode, list[dict]] = defaultdict(list) + for example in cui_examples: + by_mode[example['failure_mode']].append(example) + per_cui_examples[cui] = by_mode + + total_dd: defaultdict[FailureMode, list[dict]] = defaultdict(list) + per_type_id_dd: defaultdict[ + str, defaultdict[FailureMode, list[dict]] + ] = defaultdict[str, defaultdict[FailureMode, list]]( + lambda: defaultdict(list)) + for cui, cur_cui_examples in per_cui_examples.items(): + type_ids = cui2type_ids_call(cui) + for mode, exs in cur_cui_examples.items(): + total_dd[mode].extend(exs) + for type_id in type_ids: + per_type_id_dd[type_id][mode].extend(exs) + + max_d = include_details_for_examples_max + per_cui_counts = { + cui: _collapse(by_mode, max_d) + for cui, by_mode in per_cui_examples.items() + } + per_type_id_counts = { + tid: _collapse(by_mode, max_d) + for tid, by_mode in per_type_id_dd.items() + } + total_counts = _collapse(total_dd, max_d) + + # do prints if required + if print_mode in ('per_type_id', 'all'): + print("Per type ID failure modes") + pprint(per_cui_counts, width=120) + if print_mode in ('per_cui', 'all'): + print("Per CUI failure modes") + pprint(per_cui_counts, width=120) + if print_mode in ('totals', 'all'): + print("All failure modes") + pprint(total_counts, width=120) + return { + "per_type_id_counts": per_type_id_counts, + "per_cui_counts": per_cui_counts, + "total_counts": total_counts, + } diff --git a/medcat/medcat/stats/stats.py b/medcat/medcat/stats/stats.py index ae75d4d42..739c5e4e1 100644 --- a/medcat/medcat/stats/stats.py +++ b/medcat/medcat/stats/stats.py @@ -1,21 +1,30 @@ -from typing import Optional, Callable, TextIO, TypedDict, Any +from typing import Optional, Callable, TextIO, Any +import logging from tqdm import tqdm from medcat.cat import CAT +from medcat.stats.common import ( + GoldAnnotation, PredictedAnnotation, NER_DETECTED_ENTITY_TAG) +from medcat.stats.failuremodes import FailureModeFinder from medcat.utils.filters import project_filters from medcat.data.mctexport import ( MedCATTrainerExport, MedCATTrainerExportProject, MedCATTrainerExportDocument) from medcat.config.config import LinkingFilters from medcat.cdb.concepts import CUIInfo, get_new_cui_info -from medcat.tokenizing.tokens import MutableEntity, UNTOKENIZABLE_ENTITY_ID +from medcat.tokenizing.tokens import ( + MutableDocument, MutableEntity, UNTOKENIZABLE_ENTITY_ID) from medcat.components.types import CoreComponentType from medcat.utils.training_utils import dataset_aware_component from collections import defaultdict from pydantic import BaseModel, Field from enum import Enum + +logger = logging.getLogger(__name__) + + class MetricMode(str, Enum): """Supported evaluation modes for statistics collection.""" @@ -35,8 +44,8 @@ class RawStats(BaseModel): # i.e. entity at chars 100-103, token is 100-104. no_tokens: int = 0 - # per document IoU metrics, summed over all documents and - # averaged later via char_docs, which counts the number of + # per document IoU metrics, summed over all documents and + # averaged later via char_docs, which counts the number of # documents that have entities that have been processed. iou_sum: float = 0.0 giou_sum: float = 0.0 @@ -50,7 +59,7 @@ class RawStats(BaseModel): # gold counts is the number of labels for that CUI cui_gold_counts: dict[str, int] = Field(default_factory=dict) cui_no_tokens: dict[str, int] = Field(default_factory=dict) - + examples: dict[str, dict[str, list]] = { 'tp': {}, 'fp': {}, 'fn': {}} @@ -71,10 +80,10 @@ class OverallMetrics(BaseModel): precision: float = 0.0 recall: float = 0.0 f1: float = 0.0 - + # Number of labels where it is not possible to generate an entity no_tokens: int = 0 - # Number of labels in entire project where it is not + # Number of labels in entire project where it is not # possible to generate an entity no_tokens_ratio: float = 0.0 @@ -113,45 +122,15 @@ class Metrics(BaseModel): per_cui: dict[str, CUIMetrics] = Field(default_factory=dict) -class GoldAnnotation(TypedDict): - """Validated gold annotation payload after CUI filtering.""" - - start: int - end: int - cuis: list[str] - cui: str - text: str - raw: object - document_id: int - document_name: str - # context is 60 chars before + after the entity - context: str - - -class PredictedAnnotation(TypedDict): - """Predicted entity payload used for scoring and metrics.""" - - start: int - end: int - cui: str - text: str - confidence: float - raw: MutableEntity - no_tokens: int - document_id: int - document_name: str - # context is 60 chars before + after the entity - context: str - - class ModeStats(BaseModel): """Accumulated state and calculated metrics for one evaluation mode.""" stats: RawStats = Field(default_factory=RawStats) metrics: Metrics | None = None - + + class ProjectStats(BaseModel): - """Accumulated state and calculated metrics for one project + """Accumulated state and calculated metrics for one project or all projects.""" full_pipeline: ModeStats = Field( default_factory=ModeStats @@ -162,7 +141,7 @@ class ProjectStats(BaseModel): def get_mode(self, mode: MetricMode) -> ModeStats | None: normalized = MetricMode(mode) return getattr(self, normalized.value) - + @classmethod def create( cls, @@ -173,7 +152,8 @@ def create( ner=ModeStats() if ner else None, linking=ModeStats() if linking else None, ) - + + class StatsCollection(BaseModel): """Accumulated state and calculated metrics for all projects.""" all_projects: ProjectStats = Field( @@ -211,7 +191,8 @@ def create( for i in range(num_projects) }, ) - + + class StatsCalculator: """Calculates statistics for entity linking.""" @@ -221,16 +202,18 @@ def __init__(self, num_projects: int, ner_performance: bool = False, linking_performance: bool = False, + failure_mode_finder: FailureModeFinder | None = None ) -> None: self.filters = filters self.cui2info = cui2info - self.reset(num_projects, + self.reset(num_projects, ner_performance, linking_performance) + self.failure_mode_finder = failure_mode_finder - def reset(self, - num_projects: int, - ner_performance: bool = False, + def reset(self, + num_projects: int, + ner_performance: bool = False, linking_performance: bool = False) -> None: self.ner_performance = ner_performance self.linking_performance = linking_performance @@ -281,7 +264,7 @@ def _extract_gold_annotations( 'raw': ann }) return gold_anns - + def _extract_predictions( self, predictions: list[MutableEntity], @@ -320,14 +303,14 @@ def _count_gold_annotations( """Count gold annotations for a project and all-projects aggregate.""" project_stats = self.stats.get_project_stats(project_index) aggregate_stats = self.stats.get_aggregate_stats() - + for stats in (project_stats, aggregate_stats): mode_stats = stats.get_mode(mode) if mode_stats is None: continue state = mode_stats.stats if mode == MetricMode.NER: - key = "DETECTED_ENTITY" + key = NER_DETECTED_ENTITY_TAG state.cui_gold_counts[key] = ( state.cui_gold_counts.get(key, 0) + len(gold_anns) @@ -343,7 +326,7 @@ def _count_gold_annotations( def _create_example(self, project_name: str, project_id: str, cui: str, ann: PredictedAnnotation | GoldAnnotation, confidence: float = 1.0) -> dict: - """Create a standardized example payload for true positive, false positive, and + """Create a standardized example payload for true positive, false positive, and false negative examples.""" return { 'cui': cui, @@ -357,10 +340,10 @@ def _create_example(self, project_name: str, project_id: str, cui: str, 'source_value': ann['text'], 'text': ann['context'], } - - def _record_tp(self, - state: RawStats, - gold: GoldAnnotation, + + def _record_tp(self, + state: RawStats, + gold: GoldAnnotation, pred: PredictedAnnotation, project_id: str, project_name: str) -> None: @@ -368,6 +351,7 @@ def _record_tp(self, cui = pred['cui'] state.tp += 1 state.cui_tp[cui] = state.cui_tp.get(cui, 0) + 1 + if cui not in state.examples['tp']: state.examples['tp'][cui] = [] state.examples['tp'][cui].append(self._create_example( @@ -378,27 +362,45 @@ def _record_tp(self, confidence=pred['confidence'] )) - def _record_fn(self, - state: RawStats, + def _record_fn(self, + state: RawStats, gold: GoldAnnotation, project_id: str, - project_name: str) -> None: + project_name: str) -> dict: """Record a false negative.""" cui = gold['cui'] state.fn += 1 state.cui_fn[cui] = state.cui_fn.get(cui, 0) + 1 - + if cui not in state.examples['fn']: state.examples['fn'][cui] = [] - state.examples['fn'][cui].append(self._create_example( + example = self._create_example( project_name=project_name, project_id=project_id, cui=cui, ann=gold - )) - - def _record_fp(self, - state: RawStats, + ) + state.examples['fn'][cui].append(example) + return example + + def _add_failure_mode( + self, + mut_doc: MutableDocument, + example: dict, + gold: GoldAnnotation, + pred_anns: list[PredictedAnnotation], + ) -> None: + if self.failure_mode_finder is None: + return # NOTE: already checked before call, really + suitable_ids = self._find_all_matching_prediction( + gold, pred_anns, check_cui=False) + suitable = [pred_anns[idx] for idx in suitable_ids] + example['failure_mode'] = self.failure_mode_finder.get_failure_mode( + mut_doc, example, suitable, pred_anns, + ) + + def _record_fp(self, + state: RawStats, pred: PredictedAnnotation, project_id: str, project_name: str) -> None: @@ -406,7 +408,7 @@ def _record_fp(self, cui = pred['cui'] state.fp += 1 state.cui_fp[cui] = state.cui_fp.get(cui, 0) + 1 - + if cui not in state.examples['fp']: state.examples['fp'][cui] = [] state.examples['fp'][cui].append(self._create_example( @@ -416,11 +418,11 @@ def _record_fp(self, ann=pred, confidence=pred['confidence'] )) - - def _record_no_tokens(self, - state: RawStats, - pred: PredictedAnnotation, - project_id: str, + + def _record_no_tokens(self, + state: RawStats, + pred: PredictedAnnotation, + project_id: str, project_name: str) -> None: """Record a prediction with no tokens (ID -1000).""" # When there's an entity with no way for the tokenizer to parse it @@ -443,7 +445,7 @@ def _record_no_tokens(self, state.no_tokens += 1 state.cui_no_tokens[cui] = state.cui_no_tokens.get(cui, 0) + 1 self._record_fn(state, gold, project_id, project_name) - + def _find_matching_prediction( self, gold: GoldAnnotation, @@ -458,31 +460,56 @@ def _find_matching_prediction( - Predicted CUI is in gold's acceptable CUIs - Not already matched """ - for idx, pred in enumerate(predictions): + logger.debug( + " Trying to find a match for gold... @ [%d:%d] w %s (%s). " + "Already matched %d/%d", + gold['start'], gold['end'], gold['cuis'], gold['cui'], + len(matched_preds), len(predictions) + ) + all_matches = self._find_all_matching_prediction( + gold, predictions) + for idx in all_matches: if idx in matched_preds: continue + return idx + + return None + def _find_all_matching_prediction( + self, + gold: GoldAnnotation, + predictions: list[PredictedAnnotation], + check_cui: bool = True, + ) -> list[int]: + matches: list[int] = [] + for idx, pred in enumerate(predictions): + logger.debug( + " Compare to %04d : %04d w %s (%s)", + pred['start'], pred['end'], pred['cui'], + pred['cui'] in gold['cuis'] + ) # Exact span match if pred['start'] == gold['start']: # Check if predicted CUI is acceptable - if pred['cui'] in gold['cuis']: - return idx + if not check_cui or pred['cui'] in gold['cuis']: + matches.append(idx) - return None - - def _score_annotations(self, - gold_anns: list[GoldAnnotation], + return matches + + def _score_annotations(self, + mut_doc: MutableDocument, + gold_anns: list[GoldAnnotation], pred_anns: list[PredictedAnnotation], project_index: int, project_id: str, project_name: str, - mode: MetricMode, + mode: MetricMode, filter_fp_by_cui: bool = True) -> None: aggregate_stats = self.stats.get_aggregate_stats() project_stats = self.stats.get_project_stats(project_index) all_projects_state = aggregate_stats.get_mode(mode) project_state = project_stats.get_mode(mode) - + project_and_aggregate = (all_projects_state, project_state) for state in project_and_aggregate: @@ -499,28 +526,41 @@ def _score_annotations(self, if mode == MetricMode.LINKING: for pred in pred_anns: if pred['no_tokens'] == 1: - self._record_no_tokens(state.stats, pred, project_id, + self._record_no_tokens(state.stats, pred, project_id, project_name) # NOTE: All predictions where ID are -1000 are false positives. # this should only really happen on the linker testing, as it's a perfect # NER step which is trying to create tokenless entities. # Phase 1: Match gold annotations to predictions (find TPs and FNs) + logger.debug("%d gold anns", len(gold_anns)) + num_found = 0 for gold in gold_anns: + logger.debug( + " Scoring for CUI: %s @[%d: %d]: '%s'", + gold['cui'], gold['start'], gold['end'], gold['text'] + ) if not gold['cuis']: # No valid CUIs for this gold annotation, skip it + logger.debug(" NO CUI - ignore") continue match_idx = self._find_matching_prediction( gold, pred_anns, matched_preds) if match_idx is not None: + num_found += 1 + logger.debug(" GOT match - true positive") # True Positive matched_preds.add(match_idx) pred = pred_anns[match_idx] self._record_tp(state.stats, gold, pred, project_id, project_name) else: + logger.debug(" NO match - false negative") # False Negative - self._record_fn(state.stats, gold, project_id, project_name) + example = self._record_fn( + state.stats, gold, project_id, project_name) + if self.failure_mode_finder: + self._add_failure_mode(mut_doc, example, gold, pred_anns) # Phase 2: Remaining predictions are False Positives for idx, pred in enumerate(pred_anns): @@ -530,11 +570,11 @@ def _score_annotations(self, continue self._record_fp(state.stats, pred, project_id, project_name) - def _to_ner_views(self, - gold_anns: list[GoldAnnotation], + def _to_ner_views(self, + gold_anns: list[GoldAnnotation], pred_anns: list[PredictedAnnotation] ) -> tuple[list[GoldAnnotation], list[PredictedAnnotation]]: - ner_cui = 'DETECTED_ENTITY' + ner_cui = NER_DETECTED_ENTITY_TAG eval_pred_anns: list[PredictedAnnotation] = [ {**pred, "cui": ner_cui} for pred in pred_anns @@ -545,7 +585,7 @@ def _to_ner_views(self, for gold in gold_anns ] return eval_gold_anns, eval_pred_anns - + def _build_character_sets( self, anns: list[PredictedAnnotation] | list[GoldAnnotation], @@ -559,9 +599,8 @@ def _build_character_sets( char_idxs = set(range(start, end)) chars_by_cui[cui].update(char_idxs) - return dict(chars_by_cui) - + def _character_cohen_kappa( self, gold_chars: set[int], @@ -569,9 +608,9 @@ def _character_cohen_kappa( document_length: int, ) -> float: """ - The voices in my chatbot told me this is faster than the + The voices in my chatbot told me this is faster than the sklearn implementation, and it is also more memory efficient. - + Testing shows same metrics, and halving computation speed. """ @@ -608,7 +647,7 @@ def _character_cohen_kappa( return 1.0 return (po - pe) / denominator - + def _calculate_document_character_scores( self, gold_anns: list[GoldAnnotation], @@ -699,19 +738,19 @@ def _update_project_stats( state.stats.cohen_k_sum += doc_cohen_k state.stats.char_docs += 1 - def _score_character_annotations(self, - gold_anns: list[GoldAnnotation], + def _score_character_annotations(self, + gold_anns: list[GoldAnnotation], pred_anns: list[PredictedAnnotation], - project_index: int, - mode: MetricMode, + project_index: int, + mode: MetricMode, doc_length: int) -> None: """ Calculate: - Character Intersection over Union (IoU) for gold and predicted annotations. - - Gold label Character Intersection over Union (IoU) for gold and + - Gold label Character Intersection over Union (IoU) for gold and predicted annotations. - Cohen's Kappa for gold and predicted annotations. - + Cheat sheet of what we're generating: # iou = sum of document-level macro IoUs # -> divide by number of documents @@ -753,6 +792,7 @@ def _score_character_annotations(self, def process_document( self, + mut_doc: MutableDocument, doc: MedCATTrainerExportDocument, project_index: int, project_name: str, @@ -791,10 +831,17 @@ def process_document( linked_predictions = predictions full_pipe_pred_linked_anns = self._extract_predictions(linked_predictions, doc) + logger.debug( + "Processing %d gold annotations and %d predictions " + "(%d after filtering).", + len(full_pipe_gold_anns), len(predictions), + len(linked_predictions) + ) self._count_gold_annotations(full_pipe_gold_anns, project_index, mode) self._score_annotations( + mut_doc, full_pipe_gold_anns, full_pipe_pred_linked_anns, project_index, @@ -815,37 +862,34 @@ def process_document( ner_gold_anns, ner_pred_anns = self._to_ner_views( full_pipe_gold_anns, full_pipe_pred_ner_anns) self._count_gold_annotations(ner_gold_anns, project_index, - mode=MetricMode.NER) - self._score_annotations(ner_gold_anns, ner_pred_anns, + mode=MetricMode.NER) + self._score_annotations(mut_doc, ner_gold_anns, ner_pred_anns, project_index, project_name, project_id, mode=MetricMode.NER, filter_fp_by_cui=False) self._score_character_annotations( - ner_gold_anns, + ner_gold_anns, ner_pred_anns, - project_index, + project_index, mode=MetricMode.NER, doc_length=len(doc['text']) ) - + def process_project(self, project: MedCATTrainerExportProject, project_index: int, - entity_getter: Callable[ - [str], - list[MutableEntity] - | tuple[list[MutableEntity], list[MutableEntity]], - ], + doc_getter: Callable[[str], MutableDocument | None], mode: MetricMode, calculate_ner_performance: bool = False, use_project_filters: bool = False, extra_cui_filter: set[str] | None = None ) -> None: """Process all documents in a project. - + Args: project: The project data containing documents and annotations. project_index: Index of the project in the export. - entity_getter: Function to get predicted entities from text. + doc_getter: Function to get the document (with predicted entities) + from text. mode: Evaluation mode (full, ner, linking). calculate_ner_performance: Whether to calculate NER performance. use_project_filters: Whether to apply project-specific filters. @@ -857,8 +901,23 @@ def process_project(self, project: MedCATTrainerExportProject, use_project_filters): for doc in tqdm(project['documents'], desc='Documents'): - predictions = entity_getter(doc['text']) + mut_doc = doc_getter(doc['text']) + if mut_doc is None: + logger.warning( + "Got no document from model! Ignoring doc %s (%s)", + doc.get('id', 'Unkown'), doc.get('name', 'Unknown'), + ) + continue + predictions: ( + list[MutableEntity] | + tuple[list[MutableEntity], list[MutableEntity]] + ) + if calculate_ner_performance: + predictions = (mut_doc.ner_ents, mut_doc.linked_ents) + else: + predictions = mut_doc.linked_ents self.process_document( + mut_doc, doc, project_index, project['name'], @@ -868,25 +927,6 @@ def process_project(self, project: MedCATTrainerExportProject, calculate_ner_performance=calculate_ner_performance, ) - def _get_linked_ents( - self, - cat: CAT, - text: str, - calculate_ner_performance: bool = False - ) -> ( - tuple[list[MutableEntity], list[MutableEntity]] - | list[MutableEntity] - ): - """Required for mypy cleanliness""" - doc = cat(text) - if doc is None: - return [] - if calculate_ner_performance: - # return both NER_ents and linked_ents for NER performance evaluation - return (doc.ner_ents, doc.linked_ents) - else: - return doc.linked_ents - def process_export(self, cat: CAT, export: MedCATTrainerExport, mode: MetricMode, calculate_ner_performance: bool = False, @@ -894,7 +934,7 @@ def process_export(self, cat: CAT, export: MedCATTrainerExport, extra_cui_filter: set[str] | None = None, filter_before_disamb: bool = False) -> None: """Process all projects in the export. - + Args: cat: The MedCAT CAT instance for entity linking. export: The MedCAT trainer export data. @@ -908,17 +948,15 @@ def process_export(self, cat: CAT, export: MedCATTrainerExport, cat.config.components.linking.filter_before_disamb = True for i, proj in tqdm(enumerate(export['projects']), desc='Projects'): self.process_project( - proj, + proj, i, - lambda text: self._get_linked_ents(cat, - text, - calculate_ner_performance), + cat.__call__, mode=mode, calculate_ner_performance=calculate_ner_performance, use_project_filters=use_project_filters, extra_cui_filter=extra_cui_filter ) - + @staticmethod def _compute_prf(tp: int, fp: int, fn: int, no_tokens: int) -> dict: """Compute precision, recall, F1.""" @@ -927,10 +965,10 @@ def _compute_prf(tp: int, fp: int, fn: int, no_tokens: int) -> dict: f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0 no_tokens_ratio = no_tokens / (tp + fn) if (tp + fn) > 0 else 0.0 return { - 'precision': prec, - 'recall': rec, - 'f1': f1, - 'no_tokens': no_tokens, + 'precision': prec, + 'recall': rec, + 'f1': f1, + 'no_tokens': no_tokens, 'no_tokens_ratio': f'{no_tokens_ratio:.4f}' } @@ -940,11 +978,11 @@ def _get_cui_name(self, cui: str) -> str: if info: return info.get('preferred_name') or list(info['names'])[0] return cui - + def _safe_mean(self, values): return sum(values) / len(values) if values else 0.0 - def _prepare_metrics(self, + def _prepare_metrics(self, raw_stats: RawStats) -> tuple[OverallMetrics, dict[str, dict]]: """Prepare overall and per-CUI metrics from raw accumulated state.""" # project metrics @@ -1032,7 +1070,7 @@ def compute_metrics( if mode_stats is None: return overall, per_cui = self._prepare_metrics(mode_stats.stats) - + # Store computed metrics in the ModeStats object mode_stats.metrics = Metrics( overall=overall, @@ -1041,7 +1079,7 @@ def compute_metrics( for cui, metrics in per_cui.items() }, ) - + def compute_all_metrics(self, ner_performance: bool = True, linking_performance: bool = True) -> None: @@ -1052,20 +1090,20 @@ def compute_all_metrics(self, self.compute_metrics(stats, MetricMode.NER) if linking_performance: self.compute_metrics(stats, MetricMode.LINKING) - + if self.num_projects > 1: for i in range(self.num_projects): stats = self.stats.get_project_stats(i) - self.compute_metrics(stats, + self.compute_metrics(stats, MetricMode.FULL) if ner_performance: - self.compute_metrics(stats, + self.compute_metrics(stats, MetricMode.NER) if linking_performance: - self.compute_metrics(stats, + self.compute_metrics(stats, MetricMode.LINKING) - - # these 3 functions are just copied from previous, + + # these 3 functions are just copied from previous, # they get nice names for concepts def _empty(self, cui: str) -> CUIInfo: return get_new_cui_info( @@ -1077,7 +1115,7 @@ def _get_or_empty(self, cui: str) -> CUIInfo: def _get_pref_name(self, cui: str) -> str: info = self._get_or_empty(cui) return info['preferred_name'] or list(info['names'])[0] - + def print_stats(self, epoch: int, mode_stats: ModeStats, @@ -1182,7 +1220,8 @@ def legacy_stats(self, mode_stats: "ModeStats") -> tuple[ ) return to_return -def get_stats_calculator(cat: CAT, + +def get_stats_calculator(cat: CAT, data: MedCATTrainerExport, epoch: int = 0, use_project_filters: bool = False, @@ -1190,16 +1229,18 @@ def get_stats_calculator(cat: CAT, ner_performance: bool = False, linking_performance: bool = False, extra_cui_filter: Optional[set[str]] = None, - do_print: bool = True,) -> StatsCalculator: + do_print: bool = True, + include_failure_modes: bool = False, +) -> StatsCalculator: """Return the stats calculator. - This doesn't just return the per project stats for the full pipeline, - but all stats for each project and the aggregate, for all modes + This doesn't just return the per project stats for the full pipeline, + but all stats for each project and the aggregate, for all modes (full, ner, linking) if required. - This should be called without calling "get_stats", as that is just a + This should be called without calling "get_stats", as that is just a legacy wrapper. - + Args: cat: The MedCAT CAT instance for entity linking. data: The MedCAT trainer export data. @@ -1214,12 +1255,15 @@ def get_stats_calculator(cat: CAT, Returns: StatsCalculator: An instance of StatsCalculator with computed statistics. """ + fm_finder = ( + FailureModeFinder.from_cat(cat) if include_failure_modes else None) calculator = StatsCalculator( filters=cat.config.components.linking.filters, cui2info=cat.cdb.cui2info, num_projects=len(data['projects']), ner_performance=ner_performance, - linking_performance=linking_performance + linking_performance=linking_performance, + failure_mode_finder=fm_finder, ) # Always compute full pipeline metrics. # If ner is of interest then also compute NER metrics from the same pass. @@ -1243,6 +1287,7 @@ def get_stats_calculator(cat: CAT, ) calculator.compute_all_metrics(ner_performance, linking_performance) + if do_print: full_stats = calculator.stats.all_projects.get_mode(MetricMode.FULL) if full_stats is None: @@ -1250,22 +1295,25 @@ def get_stats_calculator(cat: CAT, calculator.print_stats(epoch, full_stats) return calculator -def get_stats(cat: CAT, + +def get_stats(cat: CAT, data: MedCATTrainerExport, epoch: int = 0, use_project_filters: bool = False, use_overlaps: bool = False, extra_cui_filter: Optional[set[str]] = None, - do_print: bool = True,) -> tuple[ + do_print: bool = True, + include_failure_modes: bool = False, + ) -> tuple[ dict[str, int], dict[str, int], dict[str, int], dict[str, float], dict[str, float], dict[str, float], dict[str, int], dict - ]: +]: """Return stats for the entire project and full pipeline. - This should be called without calling "get_stats", as that is just a + This should be called without calling "get_stats", as that is just a legacy wrapper. - + Args: cat: The MedCAT CAT instance for entity linking. data: The MedCAT trainer export data. @@ -1294,11 +1342,12 @@ def get_stats(cat: CAT, ner_performance=False, linking_performance=False, extra_cui_filter=extra_cui_filter, - do_print=False + do_print=False, + include_failure_modes=include_failure_modes, ) full_stats = calculator.stats.all_projects.full_pipeline if do_print: if full_stats is None: raise ValueError("No statistics available for the full pipeline mode.") calculator.print_stats(epoch, full_stats, legacy=True) - return calculator.legacy_stats(full_stats) \ No newline at end of file + return calculator.legacy_stats(full_stats) diff --git a/medcat/medcat/utils/cdb_utils.py b/medcat/medcat/utils/cdb_utils.py index 3eda30858..346b6c570 100644 --- a/medcat/medcat/utils/cdb_utils.py +++ b/medcat/medcat/utils/cdb_utils.py @@ -299,3 +299,11 @@ def find_parents(cui, cuis2nodes, child_node=None): logger.error(f'Cannot find path concept path for CUI: {cui}', exc_info=True) return {'node_path': {}, 'links': []} + + +def reverse_pt2ch(pt2ch: dict[str, list[str]]) -> dict[str, list[str]]: + ch2pt: defaultdict[str, list[str]] = defaultdict(list) + for pt, children in pt2ch.items(): + for ch in children: + ch2pt[ch].append(pt) + return dict(ch2pt) diff --git a/medcat/medcat/utils/training_utils.py b/medcat/medcat/utils/training_utils.py index 3cd46bea8..5891237d0 100644 --- a/medcat/medcat/utils/training_utils.py +++ b/medcat/medcat/utils/training_utils.py @@ -1,4 +1,4 @@ -from typing import Callable, Optional +from typing import Callable, Optional, cast from typing_extensions import Self from contextlib import contextmanager import logging @@ -24,9 +24,13 @@ class _CheatingComponent(AbstractEntityProvidingComponent): name = 'cheating_component' - def __init__(self, - comp_type: CoreComponentType, - predictor: Callable[[MutableDocument], list[MutableEntity]]): + def __init__( + self, + orig_comp: AbstractEntityProvidingComponent, + comp_type: CoreComponentType, + predictor: Callable[[MutableDocument], list[MutableEntity]] + ): + self._orig_comp = orig_comp self._comp_type = comp_type super().__init__( comp_type == CoreComponentType.linking, @@ -39,7 +43,14 @@ def get_type(self) -> CoreComponentType: def predict_entities(self, doc: MutableDocument, ents: list[MutableEntity] | None = None ) -> list[MutableEntity]: - return self._predictor(doc) + try: + return self._predictor(doc) + except NoSuchDocumentException as e: + logger.debug( + "Unable to find matching document in dataset, " + "falling back to original component" + ) + return self._orig_comp.predict_entities(doc, ents) @classmethod def create_new_component( @@ -65,10 +76,11 @@ def cheating_component( """ comps_list = cat.pipe._components # find original index - original_comp = cat.pipe.get_component(comp_type) + original_comp = cast( + AbstractEntityProvidingComponent, cat.pipe.get_component(comp_type)) replace_index = comps_list.index(original_comp) # create and replace - cheater = _CheatingComponent(comp_type, predictor) + cheater = _CheatingComponent(original_comp, comp_type, predictor) comps_list[replace_index] = cheater try: yield @@ -76,6 +88,9 @@ def cheating_component( # restore original component comps_list[replace_index] = original_comp +class NoSuchDocumentException(ValueError): + pass + def _identify_document( doc: MutableDocument, @@ -84,7 +99,7 @@ def _identify_document( for ann_doc in proj['documents']: if ann_doc['text'] == doc.base.text: return ann_doc - raise ValueError("Unable to identify correct document") + raise NoSuchDocumentException("Unable to identify correct document") def _parse_raw_name( diff --git a/medcat/tests/resources/mct_export_win_some_loose_some.json b/medcat/tests/resources/mct_export_win_some_loose_some.json new file mode 100644 index 000000000..00dab0317 --- /dev/null +++ b/medcat/tests/resources/mct_export_win_some_loose_some.json @@ -0,0 +1,202 @@ +{ + "projects": [ + { + "cuis": "", + "documents": [ + { + "annotations": [ + { + "cui": "C01", + "start": 38, + "end": 52, + "value": "kidney failure" + }, + { + "cui": "C01", + "start": 122, + "end": 145, + "value": "loss of kidney function" + }, + { + "cui": "C02", + "start": 192, + "end": 200, + "value": "diabetes" + }, + { + "cui": "C02", + "start": 279, + "end": 296, + "value": "diabetes mellitus" + }, + { + "cui": "C03", + "start": 390, + "end": 395, + "value": "fever" + }, + { + "cui": "C03", + "start": 454, + "end": 470, + "value": "high temperature" + } + ], + "id": "ID-0", + "last_modified": "2024-08-21", + "name": "Doc#0", + "text": "Patient had been diagnosed with acute kidney failure the week before. The current complaint was related to the same acute loss of kidney function as the diagnosis. The patient also has severe diabetes even though they have never consumed any sugar. The prior diagnosis of severe diabetes mellitus was confirmed by doctor. Due to the previous issues, patient had been suffering from a light fever all day. They took some paracetamol but still had a light high temperature afterwards." + }, + { + "annotations": [ + { + "cui": "C04", + "start": 20, + "end": 27, + "value": "seizure" + }, + { + "cui": "C04", + "start": 81, + "end": 88, + "value": "fittest" + } + ], + "id": "ID-1", + "last_modified": "2024-08-21", + "name": "Doc#1", + "text": "Patient had a acute seizure during visit with GP. This is the first time a minor fittest was observed for this patient. " + }, + { + "annotations": [ + { + "cui": "C05", + "start": 26, + "end": 33, + "value": "healthy" + }, + { + "cui": "C05", + "start": 84, + "end": 91, + "value": "fittest" + } + ], + "id": "ID-2", + "last_modified": "2024-08-21", + "name": "Doc#2", + "text": "The patient is considered healthy as per tests run. The patient would be considered fittest according to any standard known." + }, + { + "annotations": [ + { + "cui": "C04", + "start": 24, + "end": 31, + "value": "seizure" + }, + { + "cui": "C04", + "start": 65, + "end": 72, + "value": "fittest" + } + ], + "id": "ID-3", + "last_modified": "2024-08-21", + "name": "Doc#3", + "text": "The patient has a minor seizure every day. The presence of daily fittest is extremely problematic." + }, + { + "annotations": [ + { + "cui": "C05", + "start": 16, + "end": 23, + "value": "healthy" + }, + { + "cui": "C05", + "start": 111, + "end": 118, + "value": "fittest" + } + ], + "id": "ID-4", + "last_modified": "2024-08-21", + "name": "Doc#4", + "text": "The RHS male is healthy as considered by all available tests. There are no indications that the patient is not fittest." + }, + { + "annotations": [ + { + "cui": "C03", + "start": 12, + "end": 19, + "value": "pyrexia" + }, + { + "cui": "C02", + "start": 27, + "end": 29, + "value": "dm" + }, + { + "cui": "C01", + "start": 34, + "end": 47, + "value": "renal failure" + } + ], + "id": "ID-5", + "last_modified": "2026-10-06", + "name": "Doc#5-FAIL(no names)", + "text": "Patient has pyrexia due to dm and renal failure as well for some reason" + }, + { + "annotations": [ + { + "cui": "C06", + "start": 22, + "end": 49, + "value": "acute myocardial infarction" + }, + { + "cui": "C07", + "start": 63, + "end": 73, + "value": "ill health" + } + ], + "id": "ID-6", + "last_modified": "2026-10-06", + "name": "Doc#6-FAIL(no cui)", + "text": "Patient suffered from acute myocardial infarction due to their ill health" + }, + { + "annotations": [ + { + "cui": "C04", + "start": 14, + "end": 19, + "value": "fever" + }, + { + "cui": "C05", + "start": 43, + "end": 51, + "value": "diabetes" + } + ], + "id": "ID-7", + "last_modified": "2026-10-06", + "name": "Doc#7-FAIL(missed link - i.e the name exists, but doesn't link to this conept)", + "text": "Patient had a fever and was diagnosed with diabetes" + } + ], + "id": "Project#0", + "name": "Project-0", + "tuis": "" + } + ] +} diff --git a/medcat/tests/stats/test_failure_modes.py b/medcat/tests/stats/test_failure_modes.py new file mode 100644 index 000000000..947113533 --- /dev/null +++ b/medcat/tests/stats/test_failure_modes.py @@ -0,0 +1,479 @@ +"""Tests for false-negative failure mode classification.""" +from types import SimpleNamespace + +import unittest + +# NOTE: adjust this import to wherever the module lives +from medcat.components import base +from medcat.stats.failuremodes import ( + FailureMode as FM, FailureModeFinder, + _get_failure_mode_for_partial_span, _get_local_span, +) + +WINDOW = 5 +DOC = "The patient was admitted with a stroke after a long fever last week." + +# Concept hierarchy used throughout: +# +# root +# ├── A +# │ ├── B +# │ │ └── C +# │ └── D +# └── E +# +# X is in the CDB but not in the hierarchy at all. +PT2CH = {"root": ["A", "E"], "A": ["B", "D"], "B": ["C"]} +ALL_CUIS = ["root", "A", "B", "C", "D", "E", "X"] +NAMES = { + "stroke": {"per_cui_status": {"B": "P"}}, + "heart attack": {"per_cui_status": {"B": "P"}}, + "fever": {"per_cui_status": {"X": "P"}}, +} +_GOLD_NAME = "stroke" +GOLD_CONTEXT = DOC[ + min(DOC.find(_GOLD_NAME) - WINDOW, 0): ( + DOC.find(_GOLD_NAME) + len(_GOLD_NAME) + WINDOW) +] + + +# --- fakes / helpers --------------------------------------------------- + + +class FakeToken: + + def __init__(self, start_char_index: int, text: str) -> None: + self.char_index = start_char_index + self.base = self + self.text = text + self.to_skip = False + self.text_versions = [text,] + + @classmethod + def from_proposal( + cls, + proposal: list[tuple[int, int, set[str]]], + ) -> list['FakeToken']: + if not proposal: + return [cls(0, "")] + prop0 = proposal[0] + text = next(iter(prop0[-1])) + # don't need end index for now + out_l = [*prop0[:-2], text] + return [cls(*out_l)] + + +class FakeEntity: + + def __init__(self, start_char_index: int, end_char_index: int, names: set[str], cuis: list[str]) -> None: + self.start_char_index = start_char_index + self.end_char_index = end_char_index + self.names = names + self.base = self + self.link_candidates = cuis + + +class FakeDocument: + + def __init__(self, tokens: list[FakeToken], ner_ents: list[FakeEntity]) -> None: + self.tokens = tokens + self.ner_ents = ner_ents + self.base = self + self.text = " ".join(tkn.text for tkn in tokens) + + @classmethod + def from_proposal( + cls, + proposal: list[tuple[int, int, set[str]]], + cui: str, + full_text: str = GOLD_CONTEXT, + ) -> 'FakeDocument': + ner_ents = [FakeEntity(*prop, cuis=[cui]) for prop in proposal] + words = full_text.split() + per_word_start_ends = [] + cursor = 0 + for word in words: + per_word_start_ends.append((cursor, cursor + len(word), {word})) + cursor += len(word) + 1 + tkns = [ + FakeToken.from_proposal([pwse])[0] + for pwse in per_word_start_ends + ] + return cls(tkns, ner_ents) + + @classmethod + def from_example(cls, example: dict, full_text: str) -> 'FakeDocument': + ent_locs = [(example['start'], example['end'], {example['source_value']})] + return cls.from_proposal(ent_locs, example['cui'], full_text=full_text) + + def __iter__(self): + yield from self.tokens + + +def make_example( + doc: str = DOC, value: str = "stroke", cui: str = "B", +) -> dict: + """Build an example the way get_stats does: a window either side.""" + start = doc.index(value) + end = start + len(value) + return { + "cui": cui, "start": start, "end": end, "source_value": value, + "text": doc[max(0, start - WINDOW): end + WINDOW], + } + + +GOLD = make_example() +GOLD_START, GOLD_END = GOLD["start"], GOLD["end"] +GOLD_LOCAL = (WINDOW, WINDOW + len("stroke")) + + +def ner_proposing(*proposals): + """NER stand-in: always proposes the given (start, end, candidates).""" + return lambda text: list(proposals) + + +def make_finder( + *, cuis=ALL_CUIS, names=NAMES, pt2ch=PT2CH, + allow=(), disallow=(), +) -> FailureModeFinder: + return FailureModeFinder( + cui2info={cui: SimpleNamespace() for cui in cuis}, + name2info=names, + pt2ch=pt2ch, + linking_filters=SimpleNamespace( + cuis=set(allow), cuis_exclude=set(disallow)), + token_separator=" ", + ) + + +def pred(cui: str, start: int, end: int) -> dict: + return {"cui": cui, "start": start, "end": end} + + +def run(finder: FailureModeFinder, span=(), all_=(), example=GOLD, full_text=DOC) -> FM: + mut_doc = FakeDocument.from_example(example, full_text) + return finder.get_failure_mode( + mut_doc, example, list(span), + list(all_), window_size=WINDOW) + + +# --- span relations ---------------------------------------------------- + +class TestPartialSpan(unittest.TestCase): + + EXAMPLES = [ + ((5, 16), FM.SPAN_PRED_CONTAINS_GOLD), # starts earlier + ((10, 20), FM.SPAN_PRED_CONTAINS_GOLD), # ends later + ((5, 20), FM.SPAN_PRED_CONTAINS_GOLD), + ((12, 14), FM.SPAN_GOLD_CONTAINS_PRED), + ((10, 14), FM.SPAN_GOLD_CONTAINS_PRED), + ((12, 16), FM.SPAN_GOLD_CONTAINS_PRED), + ((12, 20), FM.SPAN_PARTIAL_OVERLAP), # shifted right + ((5, 12), FM.SPAN_PARTIAL_OVERLAP), # shifted left + ((16, 20), None), # adjacent: no overlap + ((4, 10), None), # adjacent: no overlap + ((30, 40), None), + ] + + def test_span_relation(self): + for pred_span, expected in self.EXAMPLES: + with self.subTest(f"{pred_span} Exp: {expected}"): + self._test_span_relation(pred_span, expected) + + def _test_span_relation(self, pred_span, expected): + got = _get_failure_mode_for_partial_span("A", 10, 16, "A", *pred_span) + self.assertIs(got, expected) + + def test_different_cui_is_ignored(self): + self.assertIsNone( + _get_failure_mode_for_partial_span( + "A", 10, 16, "B", 10, 20)) + + +class TestLocalSpan(unittest.TestCase): + + def test_mid_document(self): + ex = make_example() + self.assertEqual(_get_local_span( + ex["text"], ex["start"], ex["end"], ex["source_value"], WINDOW + ), GOLD_LOCAL) + + def test_at_document_start(self): + ex = make_example("Stroke noted on arrival, nothing else.", "Stroke") + self.assertEqual(_get_local_span( + ex["text"], ex["start"], ex["end"], ex["source_value"], WINDOW + ), (0, 6)) + + def test_at_document_end(self): + ex = make_example("Admitted with a stroke", "stroke") + self.assertEqual(_get_local_span( + ex["text"], ex["start"], ex["end"], ex["source_value"], WINDOW + ), (5, 11)) + + def test_value_missing_from_context_does_not_match(self): + context = "completely different text" + start, end = _get_local_span(context, 100, 106, "stroke", WINDOW) + self.assertNotEqual(context[start:end], "stroke") + + # @pytest.mark.xfail(strict=True, reason=( + # "repeated value near a document edge: picks the occurrence closest " + # "to the middle of the context, not the real one")) + def test_repeated_value_near_document_start(self): + ex = make_example("ab ab cd efgh", "ab") + self.assertGreater(_get_local_span( + ex["text"], ex["start"], ex["end"], ex["source_value"], WINDOW + ), (0, 2)) + + +# --- hierarchy relations ----------------------------------------------- + +class TestDisambiguationRelation(unittest.TestCase): + + EXAMPLES = [ + ("B", "A", FM.WRONG_CONCEPT_DIRECT_PARENT), + ("A", "B", FM.WRONG_CONCEPT_DIRECT_CHILD), + ("B", "D", FM.WRONG_CONCEPT_SIBLING), + ("C", "A", FM.WRONG_CONCEPT_ANCESTOR), + ("C", "root", FM.WRONG_CONCEPT_ANCESTOR), + ("A", "C", FM.WRONG_CONCEPT_DESCENDANT), + ("root", "C", FM.WRONG_CONCEPT_DESCENDANT), + # cousins: share an ancestor but not a direct parent + ("B", "E", FM.WRONG_CONCEPT_UNRELATED), + # not in the hierarchy at all + ("B", "X", FM.WRONG_CONCEPT_UNRELATED), + ] + + def test_relation(self): + for gold, predicted, expected in self.EXAMPLES: + with self.subTest(f"G: {gold} -> pred {predicted} = exp {expected}"): + self._test_relation(gold, predicted, expected) + + def _test_relation(self, gold, predicted, expected): + finder = make_finder() + self.assertIs(finder._get_disamb_failure_mode(gold, predicted), expected) + + def test_cyclic_hierarchy_terminates(self): + # P -> [G, Q], Q -> [P]: walking up from G cycles P -> Q -> P + finder = make_finder(pt2ch={"P": ["G", "Q"], "Q": ["P"]}) + self.assertIs( + finder._get_disamb_failure_mode("G", "Z"), + FM.WRONG_CONCEPT_UNRELATED + ) + self.assertIs(finder._get_disamb_failure_mode("Z", "G"), + FM.WRONG_CONCEPT_UNRELATED + ) + + +# --- individual steps -------------------------------------------------- + +class TestStep0GoldSanity(unittest.TestCase): + + def test_matching_value_passes(self): + self.assertIsNone(make_finder()._step_0_gold_sanity( + GOLD["text"], *GOLD_LOCAL, "stroke")) + + def test_mismatching_value_flagged(self): + self.assertIs( + make_finder()._step_0_gold_sanity( + GOLD["text"], *GOLD_LOCAL, "stoke" + ), + FM.SPAN_INCORRECT_FOR_VALUE + ) + + +class TestStep1NameLookup(unittest.TestCase): + + @classmethod + def setUpClass(cls) -> None: + cls.gold_tkns = FakeToken.from_proposal([[GOLD_START, GOLD_END, {'stroke'}]]) + cls.unknown_name_tkns = FakeToken.from_proposal([[0, 1, {'something'}]]) + fever_start = GOLD_CONTEXT.find("fever") + cls.fever_tkns = FakeToken.from_proposal( + [(fever_start, fever_start + len('fever'), {'fever'})]) + + def test_cui_not_in_cdb(self): + finder = make_finder(cuis=[c for c in ALL_CUIS if c != "B"]) + self.assertIs(finder._step_1_concept_and_name_lookup( + self.gold_tkns, "F"), FM.CUI_NOT_IN_CDB) + + def test_name_unknown(self): + self.assertIs(make_finder()._step_1_concept_and_name_lookup( + self.unknown_name_tkns, "B"), FM.NAME_UNKNOWN) + + def test_name_not_linked_to_cui(self): + self.assertIs(make_finder()._step_1_concept_and_name_lookup( + self.fever_tkns, "B"), FM.NAME_NOT_LINKED_TO_CUI) + + def test_known_name_linked_to_cui(self): + self.assertIs(make_finder()._step_1_concept_and_name_lookup( + self.gold_tkns, "B"), None) + + def test_any_name_variant_linking_to_gold_is_enough(self): + # "Stroke" has the variants "Stroke" and "stroke"; only one of them + # links to the gold concept + names = { + "Stroke": {"per_cui_status": {"B": "P"}}, + "stroke": {"per_cui_status": {"X": "P"}}, + } + tkns = FakeToken.from_proposal( + [(0, 5, {"Stroke"}), ] + ) + self.assertIs(make_finder(names=names)._step_1_concept_and_name_lookup( + tkns, "B"), None) + + +class TestStep2Ner(unittest.TestCase): + + PROPOSALS_A: list[list[tuple[int, int, set[str]]]] = [ + [], # nothing proposed + [(*GOLD_LOCAL, {"X"})], # gold not among the candidates + [(0, 3, {"B"})], # elsewhere in the context + [(11, 15, {"B"})], # adjacent only + ] + + def test_ner_miss(self): + for prop in self.PROPOSALS_A: + with self.subTest(f"Proposal: {prop}"): + self._test_ner_miss(prop) + + def _test_ner_miss(self, proposal: list[tuple[int, int, set[str]]]): + finder = make_finder() + mut_doc = FakeDocument.from_proposal(proposal, "B") + tkns = FakeToken.from_proposal([]) + self.assertIs(finder._step_2_ner( + mut_doc, tkns), FM.NER_NO_SPAN + ) + + PROPOSALS_B = [ + (5, 11, {"B"}), # exact + (3, 8, {"B"}), # shifted: left to the span step + (0, 20, {"B", "X"}), # longer + (6, 8, {"B"}), # shorter + ] + + def test_any_overlapping_proposal_with_gold_is_not_a_miss(self): + for prop in self.PROPOSALS_B: + with self.subTest(f"Proposal: {prop}"): + self._test_any_overlapping_proposal_with_gold_is_not_a_miss( + prop) + + def _test_any_overlapping_proposal_with_gold_is_not_a_miss( + self, + proposal: list[tuple[int, int, set[str]]] + ): + finder = make_finder() + mut_doc = FakeDocument.from_proposal([proposal], 'B') + tkns = FakeToken.from_proposal([proposal]) + self.assertIsNone(finder._step_2_ner(mut_doc, tkns)) + + +class TestStep3Filters(unittest.TestCase): + + def test_no_filters(self): + self.assertIsNone(make_finder()._step_3_filters("B")) + + def test_not_in_allow_list(self): + self.assertIs( + make_finder(allow=["A"])._step_3_filters("B"), + FM.FILTERED_NOT_IN_ALLOW_LIST + ) + + def test_in_allow_list(self): + self.assertIsNone(make_finder(allow=["B"])._step_3_filters("B")) + + def test_in_disallow_list(self): + self.assertIs( + make_finder(disallow=["B"])._step_3_filters("B"), + FM.FILTERED_DISALLOW_LIST + ) + + def test_other_concept_in_disallow_list(self): + self.assertIsNone(make_finder(disallow=["A"])._step_3_filters("B")) + + +class TestStep4Disambiguation(unittest.TestCase): + + def test_nothing_predicted(self): + self.assertIsNone(make_finder()._step_4_disambiguation([], "B", "stroke")) + + def test_relation_of_prediction(self): + got = make_finder()._step_4_disambiguation( + [pred("A", GOLD_START, GOLD_END)], "B", "stroke") + self.assertIs(got, FM.WRONG_CONCEPT_DIRECT_PARENT) + + def test_multiple_predictions_pick_one_of_the_relations(self): + got = make_finder()._step_4_disambiguation( + [pred("A", GOLD_START, GOLD_END), + pred("root", GOLD_START, GOLD_END)], "B", "stroke") + self.assertIn(got, { + FM.WRONG_CONCEPT_DIRECT_PARENT, + FM.WRONG_CONCEPT_ANCESTOR + }) + + +class TestStep5PartialOverlap(unittest.TestCase): + + def test_no_predictions(self): + self.assertIsNone(make_finder()._step_5_partial_overlap(GOLD, [])) + + def test_other_concepts_are_ignored(self): + self.assertIsNone(make_finder()._step_5_partial_overlap( + GOLD, [pred("A", GOLD_START - 2, GOLD_END)])) + + def test_longer_prediction(self): + got = make_finder()._step_5_partial_overlap( + GOLD, [pred("B", GOLD_START - 2, GOLD_END)]) + self.assertIs(got, FM.SPAN_PRED_CONTAINS_GOLD) + + def test_most_common_mode_wins(self): + got = make_finder()._step_5_partial_overlap(GOLD, [ + pred("B", GOLD_START - 2, GOLD_END), + pred("B", GOLD_START - 3, GOLD_END), + pred("B", GOLD_START + 2, GOLD_END + 4), + ]) + self.assertIs(got, FM.SPAN_PRED_CONTAINS_GOLD) + + +# --- the whole cascade ------------------------------------------------- + +class TestGetFailureMode(unittest.TestCase): + + def test_falls_through_to_below_threshold(self): + self.assertIs(run(make_finder()), FM.BELOW_THRESHOLD) + + def test_gold_at_document_start(self): + doc = "stroke noted on arrival, nothing else." + ex = make_example(doc, "stroke") + finder = make_finder() + self.assertIs(run(finder, example=ex, full_text=doc), FM.BELOW_THRESHOLD) + + def test_gold_at_document_end(self): + doc = "Admitted with a stroke" + ex = make_example(doc, "stroke") + finder = make_finder() + self.assertIs(run(finder, example=ex, full_text=doc), FM.BELOW_THRESHOLD) + + def test_sanity_beats_lookup(self): + bad = {**GOLD, "source_value": "stoke"} + finder = make_finder(cuis=[c for c in ALL_CUIS if c != "B"]) + self.assertIs(run(finder, example=bad), FM.SPAN_INCORRECT_FOR_VALUE) + + def test_lookup_beats_ner(self): + finder = make_finder(names={}) + self.assertIs(run(finder), FM.NAME_UNKNOWN) + + def test_filters_beat_disambiguation(self): + finder = make_finder(disallow=["B"]) + span = [pred("A", GOLD_START, GOLD_END)] + self.assertIs(run(finder, span=span, all_=span), FM.FILTERED_DISALLOW_LIST) + + def test_disambiguation_beats_span(self): + span = [pred("A", GOLD_START, GOLD_END)] + overlapping = pred("B", GOLD_START - 2, GOLD_END) + got = run(make_finder(), span=span, all_=[*span, overlapping]) + self.assertIs(got, FM.WRONG_CONCEPT_DIRECT_PARENT) + + def test_span_beats_threshold(self): + overlapping = pred("B", GOLD_START - 2, GOLD_END) + got = run(make_finder(), all_=[overlapping]) + self.assertIs(got, FM.SPAN_PRED_CONTAINS_GOLD) diff --git a/medcat/tests/stats/test_stats.py b/medcat/tests/stats/test_stats.py index b309b0361..4f42005f1 100644 --- a/medcat/tests/stats/test_stats.py +++ b/medcat/tests/stats/test_stats.py @@ -1,9 +1,12 @@ from __future__ import annotations +from collections import Counter +import json import os import re from dataclasses import dataclass, field from types import SimpleNamespace +from medcat.stats.failuremodes import FailureMode from medcat.stats.stats import MetricMode from medcat.components.types import CoreComponentType from medcat.stats import stats @@ -432,4 +435,64 @@ def test_example_contents(self) -> None: self.assertEqual(sample[0]['start'], 16) self.assertEqual(sample[0]['end'], 22) self.assertEqual(sample[0]['acc'], 1.0) - self.assertEqual(sample[0]['source_value'], "asthma") \ No newline at end of file + self.assertEqual(sample[0]['source_value'], "asthma") + + +class StatsWithFailureModesTests(TrainedModelTests): + DATA_PATH = os.path.join( + RESOURCES_PATH, "mct_export_win_some_loose_some.json") + EXPECTED_FAILURE_MODES = { + FailureMode.NAME_UNKNOWN, FailureMode.NAME_NOT_LINKED_TO_CUI, + FailureMode.CUI_NOT_IN_CDB, + } + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.model.cdb.addl_info['pt2ch'] = { + cui: [] for cui in cls.model.cdb.cui2info + } + with open(cls.DATA_PATH) as f: + cls.data = json.load(f) + cls.stats = stats.get_stats( + cat=cls.model, + data=cls.data, + use_project_filters=False, + do_print=False, + include_failure_modes=True, + ) + cls.examples = cls.stats[-1] + cls.all_fms = Counter( + ex['failure_mode'] for examples in cls.examples['fn'].values() + for ex in examples + ) + + def test_fns_have_failure_modes(self): + fns = self.examples['fn'] + self.assertTrue(fns) + for cui, examples in fns.items(): + for ex in examples: + with self.subTest(f"{cui} w {ex}"): + self.assertIn('failure_mode', ex) + val = ex['failure_mode'] + self.assertIsInstance(val, FailureMode) + + def test_others_do_not_have_failure_modes(self): + for key, vals in self.examples.items(): + if key == 'fn': + continue + for cui, examples in vals.items(): + for ex in examples: + with self.subTest(f"{key}: {cui} w {ex}"): + self.assertNotIn('failure_mode', ex) + + def test_has_multiple_types_of_failure_modes(self): + self.assertTrue(self.all_fms) + self.assertGreater(len(self.all_fms), 1) + self.assertGreaterEqual(set(self.all_fms), self.EXPECTED_FAILURE_MODES) + + def test_has_failure_mode_per_false_negative(self): + self.assertEqual( + self.all_fms.total(), + sum(self.stats[1].values()) + )