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