Spaces:
Sleeping
Sleeping
File size: 4,346 Bytes
021e07f 82a4c62 021e07f 82a4c62 021e07f 82a4c62 021e07f 82a4c62 021e07f 82a4c62 021e07f 82a4c62 021e07f 82a4c62 021e07f | 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 | """
Text preprocessing pipeline for MAUDE narrative reports.
Cleans and normalizes raw MDR narrative text for NLP ingestion.
"""
import re
import string
import logging
from typing import List
import pandas as pd
logger = logging.getLogger(__name__)
# Common medical abbreviation expansions (extend as needed)
ABBREVIATION_MAP = {
r"\bpt\b": "patient",
r"\bpts\b": "patients",
r"\bmd\b": "physician",
r"\bdr\b": "doctor",
r"\bhosp\b": "hospital",
r"\badm\b": "admitted",
r"\bdx\b": "diagnosis",
r"\btx\b": "treatment",
r"\brx\b": "prescription",
r"\bs/p\b": "status post",
r"\bw/\b": "with",
r"\bh/o\b": "history of",
r"\bc/o\b": "complaint of",
r"\bn/v\b": "nausea vomiting",
r"\bSOB\b": "shortness of breath",
r"\bUNK\b": "unknown",
}
def expand_abbreviations(text: str) -> str:
"""Replace common medical abbreviations with full forms."""
for pattern, replacement in ABBREVIATION_MAP.items():
text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
return text
def remove_boilerplate(text: str) -> str:
"""
Remove common MAUDE boilerplate phrases that carry no signal.
These often appear at the start/end of narrative sections.
"""
boilerplate_patterns = [
r"it was reported that",
r"the reporter stated",
r"according to the report",
r"this is a report",
r"per the report",
r"the following information was received",
r"information has been received",
r"no further information (is|was) available",
r"follow.?up (is|will be) requested",
]
for pattern in boilerplate_patterns:
text = re.sub(pattern, " ", text, flags=re.IGNORECASE)
return text
def clean_text(text: str, lowercase: bool = True, preserve_digits: bool = False) -> str:
"""
Full cleaning pipeline for a single narrative string.
Steps:
1. Strip whitespace
2. Remove boilerplate
3. Expand abbreviations
4. Remove special characters (and optionally digits)
5. Normalize whitespace
6. Lowercase (optional)
Args:
text: Raw narrative text string.
lowercase: Whether to convert to lowercase.
preserve_digits: If True, keep digit tokens (required for BERT — device
model numbers and dosages are meaningful to the encoder).
If False (default, TF-IDF mode), digits are stripped.
Returns:
Cleaned text string.
"""
if not isinstance(text, str) or not text.strip():
return ""
text = text.strip()
text = remove_boilerplate(text)
text = expand_abbreviations(text)
# Remove non-alphanumeric characters (keep spaces; optionally keep digits)
if preserve_digits:
text = re.sub(r"[^a-zA-Z0-9\s]", " ", text)
else:
text = re.sub(r"[^a-zA-Z\s]", " ", text)
# Collapse multiple spaces
text = re.sub(r"\s+", " ", text).strip()
if lowercase:
text = text.lower()
return text
def clean_dataframe(
df: pd.DataFrame,
text_col: str = "narrative_text",
output_col: str = "clean_text",
preserve_digits: bool = False,
) -> pd.DataFrame:
"""
Apply clean_text to an entire DataFrame column.
Args:
df: Input DataFrame with raw narrative text.
text_col: Name of the column containing raw text.
output_col: Name of the new column to store cleaned text.
preserve_digits: Passed through to clean_text; set True for BERT.
Returns:
DataFrame with a new cleaned text column, with empty-text rows dropped.
"""
logger.info(f"Cleaning text in column '{text_col}'...")
df = df.copy()
df[output_col] = df[text_col].apply(
lambda t: clean_text(t, preserve_digits=preserve_digits)
)
# Drop rows where cleaning produced empty strings
before = len(df)
df = df[df[output_col].str.len() > 10].reset_index(drop=True)
after = len(df)
logger.info(f"Dropped {before - after} rows with insufficient text. {after} rows remaining.")
return df
def get_label_distribution(df: pd.DataFrame, label_col: str = "severity_label") -> pd.Series:
"""Return label counts for inspection."""
return df[label_col].value_counts()
|