alt_methods.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### should write a prompt that takes guidelines and converts to amenable format for each model here\n",
    "\n",
    "# TODO\n",
    "# - do I need to update the NER spans again?\n",
    "# - get gliner (with bio backbone), nuextract working\n",
    "# - scores per report as per standard eval\n",
    "# - scores per entity/20 reports (x=models, y=entity type)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### convert my QA dataset to format for NER app\n",
    "\n",
    "from src.utils.json import load_json, save_json\n",
    "\n",
    "input_json = load_json(\"src/renal_biopsy/data/output_report_first100.json\")\n",
    "ner_json = [f\"MICROSCOPY SECTION: {report_dict['microscopy_section']} \\n \\\n",
    "              CONCLUSION SECTION: {report_dict['conclusion_section']}\"\n",
    "            for report_dict in input_json]\n",
    "\n",
    "# save_json(ner_json, \"src/ner/example_output/data/ner_output_report_first100.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from src.utils.json import load_json\n",
    "input_json = load_json(f\"src/ner/example_output/data/ner_output_report_first100.json\")\n",
    "ner_gt_json = load_json(f\"src/ner/annotations_output/annotations_first20.json\")\n",
    "qa_gt_json = load_json(f\"src/renal_biopsy/data/output_report_first100.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import spacy\n",
    "from spacy.tokens import Span\n",
    "from spacy.language import Language\n",
    "import re\n",
    "\n",
    "@Language.component(\"custom_entity_ruler\")\n",
    "def custom_entity_ruler(doc):\n",
    "    patterns = {\n",
    "        \"cortex\": r\"(?i)cortex(?!\\s+(?:absent|missing|not\\s+(?:present|identified|seen)))\",\n",
    "        \"medulla\": r\"(?i)medulla(?!\\s+(?:absent|missing|not\\s+(?:present|identified|seen)))\",\n",
    "        \"cortex_absent\": r\"(?i)cortex\\s+(?:absent|missing|not\\s+(?:present|identified|seen))\",\n",
    "        \"medulla_absent\": r\"(?i)medulla\\s+(?:absent|missing|not\\s+(?:present|identified|seen))\",\n",
    "        \"n_total\": r\"(?i)(\\d+)\\s*glomerul[ius]\",\n",
    "        \"n_global\": r\"(?i)(\\d+)\\s*(?:globally|complete(?:ly)?)\\s*sclerosed\\s*glomerul[ius]\",\n",
    "        \"n_segmental\": r\"(?i)(\\d+)\\s*(?:segment(?:al(?:ly)?)|partial(?:ly)?)\\s*sclerosed\\s*glomerul[ius]\",\n",
    "        \"abnormal_glomeruli\": r\"(?i)(?:glomerul(?:ar|i))?\\s*(?:(?P<count>\\d+)\\s+)?(?:scarring|isch(?:a)?emic\\s+changes|thickening\\s+of\\s+bowman(?:'s)?\\s+capsule|thickening\\s+of\\s+(?:capillary\\s+)?basement\\s+membrane|mesangial\\s+(?:expansion|thickening)|changes?\\s+in\\s+(?:size|shape)|irregular\\s+(?:size|shape))\",\n",
    "        \"chronic_change_adj\": r\"(?i)(mild|moderate|severe|minimal|marked|extensive)\\s+(?:tubular\\s+atrophy|interstitial\\s+fibrosis|chronic\\s+changes?)\",\n",
    "        \"chronic_change_percentage\": r\"(?i)(\\d+(?:\\.\\d+)?)\\s*%\\s*(?:of\\s*)?(?:tubular\\s+atrophy|interstitial\\s+fibrosis|chronic\\s+changes?)\",\n",
    "        \"transplant\": r\"(?i)(transplant(?:ation)?|infiltrate[sd]?|infection)\",\n",
    "        \"diagnosis\": r\"(?i)(?:final\\s*)?diagnosis\\s*[:;]\\s*(.*?)(?:\\.|$)\"\n",
    "    }\n",
    "    \n",
    "    entities = []\n",
    "    for ent_type, pattern in patterns.items():\n",
    "        for match in re.finditer(pattern, doc.text):\n",
    "            start, end = match.span()\n",
    "            if not any(start < e.end_char and end > e.start_char for e in entities):\n",
    "                span = doc.char_span(start, end, label=ent_type)\n",
    "                if span is not None:\n",
    "                    entities.append(span)\n",
    "    \n",
    "    doc.ents = entities\n",
    "    return doc\n",
    "\n",
    "def process_reports(texts, n_prototype=40):\n",
    "    nlp = spacy.load(\"en_core_web_sm\")\n",
    "    nlp.add_pipe(\"custom_entity_ruler\")\n",
    "    \n",
    "    all_results = []\n",
    "    for i, text in enumerate(texts):\n",
    "        if i == n_prototype:\n",
    "            break\n",
    "        \n",
    "        doc = nlp(text)\n",
    "        \n",
    "        results = {\n",
    "            \"cortex_present\": True,\n",
    "            \"medulla_present\": False,\n",
    "            \"n_total\": 0,\n",
    "            \"n_segmental\": 0,\n",
    "            \"n_global\": 0,\n",
    "            \"abnormal_glomeruli\": False,\n",
    "            \"chronic_change\": None,\n",
    "            \"transplant\": False,\n",
    "            \"diagnosis\": None\n",
    "        }\n",
    "        \n",
    "        has_cortex_mention = False\n",
    "        has_medulla_mention = False\n",
    "        chronic_change_perc = None\n",
    "        chronic_change_adj = None\n",
    "        \n",
    "        for ent in doc.ents:\n",
    "            if ent.label_ == \"cortex\":\n",
    "                results[\"cortex_present\"] = True\n",
    "                has_cortex_mention = True\n",
    "            elif ent.label_ == \"cortex_absent\":\n",
    "                results[\"cortex_present\"] = False\n",
    "                has_cortex_mention = True\n",
    "            elif ent.label_ == \"medulla\":\n",
    "                results[\"medulla_present\"] = True\n",
    "                has_medulla_mention = True\n",
    "            elif ent.label_ == \"medulla_absent\":\n",
    "                results[\"medulla_present\"] = False\n",
    "                has_medulla_mention = True\n",
    "            elif ent.label_ == \"transplant\":\n",
    "                results[\"transplant\"] = True\n",
    "            elif ent.label_ == \"abnormal_glomeruli\":\n",
    "                results[\"abnormal_glomeruli\"] = True\n",
    "            elif ent.label_ == \"chronic_change_percentage\":\n",
    "                chronic_change_perc = float(re.search(r\"(\\d+(?:\\.\\d+)?)\", ent.text).group(1))\n",
    "            elif ent.label_ == \"chronic_change_adj\":\n",
    "                chronic_change_adj = re.search(r\"(mild|moderate|severe|minimal|marked|extensive)\", \n",
    "                                            ent.text, re.I).group(1).lower()\n",
    "            elif ent.label_ in [\"n_total\", \"n_segmental\", \"n_global\"]:\n",
    "                results[ent.label_] = int(re.search(r\"(\\d+)\", ent.text).group(1))\n",
    "            elif ent.label_ == \"diagnosis\":\n",
    "                results[\"diagnosis\"] = ent.text.split(\":\", 1)[-1].strip()\n",
    "        \n",
    "        results[\"chronic_change\"] = chronic_change_perc if chronic_change_perc is not None else chronic_change_adj\n",
    "        all_results.append(results)\n",
    "    \n",
    "    return all_results\n",
    "\n",
    "# Example usage\n",
    "texts = [\n",
    "    \"Cortex present with 20% chronic changes. Total of 31 glomeruli.\",\n",
    "    \"Severe chronic changes noted. No rejection.\"\n",
    "]\n",
    "# json_output = process_reports(texts)\n",
    "# print(json.dumps(json_output, indent=2))\n",
    "\n",
    "# Example usage\n",
    "sample_report = \"\"\"\n",
    "Kidney biopsy shows not cortex present with just medulla present.\n",
    "Total of 18 glomeruli identified, with 3 globally sclerosed glomeruli.\n",
    "Moderate tubular atrophy noted.\n",
    "Severe interstitial fibrosis with 40% chronic changes.\n",
    "Mild chronic changes.\n",
    "This appears to be a transplant kidney.\n",
    "Final diagnosis: Acute cellular rejection, Banff grade IA.\n",
    "\"\"\"\n",
    "# results = process_reports([sample_report])\n",
    "# print(results)\n",
    "\n",
    "all_results = process_reports(input_json, n_prototype=10)\n",
    "all_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from transformers import AutoTokenizer, AutoModelForQuestionAnswering, pipeline\n",
    "import torch\n",
    "import json\n",
    "\n",
    "class MedicalReportQA:\n",
    "    def __init__(self, model_name=\"dmis-lab/biobert-large-cased-v1.1-squad\"):\n",
    "        self.tokenizer = AutoTokenizer.from_pretrained(model_name)\n",
    "        self.model = AutoModelForQuestionAnswering.from_pretrained(model_name)\n",
    "        self.qa_pipeline = pipeline(\"question-answering\", model=self.model, tokenizer=self.tokenizer)\n",
    "        \n",
    "        self.questions = {\n",
    "            \"cortex_present\": \"Is cortex present in the sample?\",\n",
    "            \"medulla_present\": \"Is medulla present in the sample?\",\n",
    "            \"n_total\": \"How many total glomeruli are there?\",\n",
    "            \"n_segmental\": \"How many segmentally sclerosed glomeruli are there?\",\n",
    "            \"n_global\": \"How many globally sclerosed glomeruli are there?\",\n",
    "            \"abnormal_glomeruli\": \"Are there any abnormal glomeruli features like scarring, thickening, or irregular shape?\",\n",
    "            \"chronic_change\": \"What percentage or degree of chronic changes are present?\",\n",
    "            \"transplant\": \"Is this a transplant case or are there infiltrates/infection?\",\n",
    "            \"diagnosis\": \"What is the final diagnosis?\"\n",
    "        }\n",
    "\n",
    "    def process_answer(self, answer, entity_type):\n",
    "        if answer['score'] < 0.1:  # Confidence threshold\n",
    "            return self._get_default_value(entity_type)\n",
    "            \n",
    "        text = answer['answer'].lower().strip()\n",
    "        \n",
    "        if entity_type.startswith('n_'):\n",
    "            numbers = ''.join(c for c in text if c.isdigit())\n",
    "            return int(numbers) if numbers else 0\n",
    "            \n",
    "        elif entity_type == 'chronic_change':\n",
    "            if '%' in text:\n",
    "                return float(''.join(c for c in text if c.isdigit() or c == '.'))\n",
    "            elif any(adj in text for adj in ['mild', 'moderate', 'severe', 'minimal', 'marked', 'extensive']):\n",
    "                return text\n",
    "            return None\n",
    "            \n",
    "        elif entity_type in ['cortex_present', 'medulla_present', 'abnormal_glomeruli', 'transplant']:\n",
    "            return not any(neg in text for neg in ['no', 'not', 'absent', 'missing'])\n",
    "            \n",
    "        return text\n",
    "\n",
    "    def _get_default_value(self, entity_type):\n",
    "        defaults = {\n",
    "            'cortex_present': True,\n",
    "            'medulla_present': False,\n",
    "            'n_total': 0,\n",
    "            'n_segmental': 0,\n",
    "            'n_global': 0,\n",
    "            'abnormal_glomeruli': False,\n",
    "            'chronic_change': 0,\n",
    "            'transplant': False,\n",
    "            'diagnosis': None\n",
    "        }\n",
    "        return defaults[entity_type]\n",
    "\n",
    "    def process_report(self, text):\n",
    "        results = {}\n",
    "        for entity, question in self.questions.items():\n",
    "            qa_result = self.qa_pipeline(question=question, context=text)\n",
    "            results[entity] = self.process_answer(qa_result, entity)\n",
    "        return results\n",
    "\n",
    "    def process_reports(self, texts):\n",
    "        return [self.process_report(text) for text in texts]\n",
    "\n",
    "# Example usage\n",
    "def main(reports):\n",
    "    qa_model = MedicalReportQA()\n",
    "    results = qa_model.process_reports(reports)\n",
    "    # print(json.dumps(results, indent=2))\n",
    "    return results\n",
    "    \n",
    "\n",
    "sample_reports = [\n",
    "        \"\"\"\n",
    "        Kidney biopsy shows cortex present with medulla noted.\n",
    "        Total of 18 glomeruli identified, with 3 globally sclerosed glomeruli.\n",
    "        Approximately 20% chronic change observed.\n",
    "        This appears to be a transplant kidney.\n",
    "        Final diagnosis: Acute cellular rejection, Banff grade IA.\n",
    "        \"\"\",\n",
    "        \"\"\"\n",
    "        Cortex present. No medulla seen.\n",
    "        12 glomeruli, 2 showing segmental sclerosis.\n",
    "        Moderate chronic changes noted.\n",
    "        \"\"\"\n",
    "    ]\n",
    "\n",
    "results = main(input_json[0:2])\n",
    "results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# gliner\n",
    "from gliner import GLiNER\n",
    "model_name = \"urchade/gliner_mediumv2.1\" # \"urchade/gliner_large_bio-v0.1\" \n",
    "model = GLiNER.from_pretrained(model_name)\n",
    "model.eval()\n",
    "\n",
    "text = \"\"\"\n",
    "    PERCUTANEOUS RENAL BIOPSY; FEATURES SUGGESTIVE OF ACUTE URINARY TRACT INFECTION ON A BACKGROUND OF MODERATE CHRONIC DAMAGE\n",
    "\"\"\"\n",
    "\n",
    "labels = ['diagnosis']\n",
    "\n",
    "entities = model.predict_entities(text, labels, threshold=0.4)\n",
    "\n",
    "for entity in entities:\n",
    "    print(entity[\"text\"], \"=>\", entity[\"label\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuextract"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "gosh-llm-2 (ipython)",
   "language": "python",
   "name": "gosh-llm-2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}