Spaces:
Sleeping
Sleeping
File size: 8,637 Bytes
c2b7494 7c142c4 c2b7494 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
import pandas as pd
import yaml
@dataclass
class EventDefinitionForm:
"""Structured phenotype definition (EDF) providing semantic context for
LLM-assisted concept retrieval and code classification."""
name: str
preferred_name: str
definition: str = ""
inclusion_criteria: list[str] = field(default_factory=list)
narrow_definition: str = ""
exclusion_criteria: list[str] = field(default_factory=list)
narrow_decision_rules: list[str] = field(default_factory=list)
possible_decision_rules: list[str] = field(default_factory=list)
exclude_decision_rules: list[str] = field(default_factory=list)
synonyms: list[str] = field(default_factory=list)
anatomical_location: str = ""
population_restrictions: str = ""
temporal_characteristics: str = ""
etiology: str = ""
morphology: str = ""
diagnostic_certainty: str = ""
administrative_context: str = ""
candidate_anchor_codes: list[str] = field(default_factory=list)
ambiguities: list[str] = field(default_factory=list)
references: list[str] = field(default_factory=list)
def to_prompt_context(self, include_possible: bool = True) -> str:
"""Render the EDF as structured text suitable for retrieval or an LLM prompt.
Possible rules are never discarded. In binary mode they define cases
that are insufficient for Narrow and therefore map to Exclude.
"""
lines = [f"Preferred name: {self.preferred_name}"]
if self.definition:
lines.append(f"Definition: {self.definition}")
if self.narrow_definition:
lines.append(f"Narrow definition: {self.narrow_definition}")
if self.inclusion_criteria:
lines.append("Inclusion criteria:")
lines += [f" - {c}" for c in self.inclusion_criteria]
if self.exclusion_criteria:
lines.append("Exclusion criteria:")
lines += [f" - {c}" for c in self.exclusion_criteria]
if self.narrow_decision_rules:
lines.append("Criteria for Narrow classification:")
lines += [f" - {c}" for c in self.narrow_decision_rules]
if self.possible_decision_rules:
if include_possible:
lines.append("Criteria for Possible classification (when enabled):")
else:
lines.append(
"Criteria that are insufficient for Narrow classification "
"(classify as Exclude because Possible is disabled):"
)
lines += [f" - {c}" for c in self.possible_decision_rules]
if self.exclude_decision_rules:
lines.append("Criteria for Exclude classification:")
lines += [f" - {c}" for c in self.exclude_decision_rules]
if self.synonyms:
lines.append(f"Synonyms: {', '.join(self.synonyms)}")
if self.anatomical_location:
lines.append(f"Anatomical location: {self.anatomical_location}")
if self.population_restrictions:
lines.append(f"Population restrictions: {self.population_restrictions}")
if self.temporal_characteristics:
lines.append(f"Temporal characteristics: {self.temporal_characteristics}")
if self.etiology:
lines.append(f"Etiology: {self.etiology}")
if self.morphology:
lines.append(f"Morphology/pathology: {self.morphology}")
if self.diagnostic_certainty:
lines.append(f"Diagnostic certainty: {self.diagnostic_certainty}")
if self.administrative_context:
lines.append(f"Administrative/historical context: {self.administrative_context}")
if self.candidate_anchor_codes:
lines.append("Candidate anchor codes:")
lines += [f" - {c}" for c in self.candidate_anchor_codes]
if self.ambiguities:
lines.append("Ambiguities:")
lines += [f" - {c}" for c in self.ambiguities]
if self.references:
lines.append("References:")
lines += [f" - {c}" for c in self.references]
return "\n".join(lines)
def to_lexical_query(self) -> str:
"""Render the phenotype name and its synonyms for lexical matching."""
terms = [self.preferred_name.strip(), *(term.strip() for term in self.synonyms)]
return " ".join(dict.fromkeys(term for term in terms if term))
def to_embedding_query(self, *, include_labels: bool = True) -> str:
"""Render retrieval fields; retain labeled defaults for study compatibility."""
if not include_labels:
values = [self.preferred_name]
if self.narrow_definition:
values.append(self.narrow_definition)
if self.synonyms:
values.append(", ".join(self.synonyms))
values.extend(self.narrow_decision_rules)
values.extend(self.candidate_anchor_codes)
if self.morphology:
values.append(self.morphology)
if self.anatomical_location:
values.append(self.anatomical_location)
return "\n".join(str(value) for value in values)
lines = [f"Preferred name: {self.preferred_name}"]
if self.narrow_definition:
lines.append(f"Narrow definition: {self.narrow_definition}")
if self.synonyms:
lines.append(f"Synonyms: {', '.join(self.synonyms)}")
if self.narrow_decision_rules:
lines.append("Narrow decision rules:")
lines += [f" - {c}" for c in self.narrow_decision_rules]
if self.candidate_anchor_codes:
lines.append("Candidate anchor codes:")
lines += [f" - {c}" for c in self.candidate_anchor_codes]
if self.morphology:
lines.append(f"Morphology/pathology: {self.morphology}")
if self.anatomical_location:
lines.append(f"Anatomical location: {self.anatomical_location}")
return "\n".join(lines)
def load_edf(name: str, edf_dir: str | Path = "data/edfs/yamls") -> EventDefinitionForm:
"""Load an Event Definition Form by phenotype name from a YAML file."""
path = Path(edf_dir) / f"{name}.yaml"
if not path.exists():
raise FileNotFoundError(f"EDF not found for phenotype '{name}': {path}")
raw = yaml.safe_load(path.read_text()) or {}
return EventDefinitionForm(name=name, **raw)
def load_aesi_dataset(csv_path: str | Path) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Load a raw AESI (Adverse Event of Special Interest) review export and
split it into a candidate code corpus and expert-curated gold labels.
These CSV exports (e.g. ``C_MYOCARD_AESI_filtered.csv``) contain the full
set of candidate codes considered during expert phenotype curation, with
one row per (code, vocabulary) pair and a ``tags`` column holding the
reviewer's final classification (``narrow``/``possible``/``exclude``).
Parameters
----------
csv_path : str or Path
Path to the raw AESI CSV export.
Returns
-------
tuple[pd.DataFrame, pd.DataFrame]
``(codes, gold_labels)`` where ``codes`` has columns
``code``, ``description``, ``vocabulary`` and ``gold_labels`` has
columns ``code``, ``code_name``, ``vocabulary``, ``label``
(Narrow/Possible/Exclude).
"""
path = Path(csv_path)
if not path.exists():
raise FileNotFoundError(f"AESI dataset not found: {path}")
raw = pd.read_csv(path, dtype=str)
required = {"code", "code_name", "coding_system", "tags"}
missing = required - set(raw.columns)
if missing:
raise ValueError(f"AESI dataset missing required columns: {missing}")
df = raw.dropna(subset=["tags"]).copy()
df["label"] = df["tags"].str.strip().str.title()
valid_labels = {"Narrow", "Possible", "Exclude"}
invalid = set(df["label"]) - valid_labels
if invalid:
raise ValueError(f"AESI dataset contains invalid tag values: {invalid}")
codes = (
df[["code", "code_name", "coding_system"]]
.rename(columns={"code_name": "description", "coding_system": "vocabulary"})
.drop_duplicates(subset=["code", "vocabulary"], keep="first")
.reset_index(drop=True)
)
gold_labels = (
df[["code", "code_name", "coding_system", "label"]]
.rename(columns={"coding_system": "vocabulary"})
.drop_duplicates(subset=["code", "vocabulary"], keep="first")
.reset_index(drop=True)
)
return codes, gold_labels
|