Clinical text

openbtk.data.clinical_text must be imported explicitly — it registers its loaders, preprocessors and chunkers as a side effect. None of them need their optional dependency (medspaCy, spaCy, pandas) merely to be defined, only to actually run.

Schemas

openbtk.data.clinical_text.schemas.ClinicalTextRecord

Bases: BaseModel

One clinical note, as loaded -- before or after de-identification.

Example

record = ClinicalTextRecord( ... record_id="note-1", ... source="synthea", ... text="Chief Complaint: chest pain.", ... ) record.deid_status

openbtk.data.clinical_text.schemas.ClinicalTextChunk

Bases: BaseModel

One retrieval- or embedding-ready unit of a ClinicalTextRecord.

Example

chunk = ClinicalTextChunk( ... chunk_id="note-1:0", ... record_id="note-1", ... text="Chief Complaint: chest pain.", ... span=TextSpan(start=0, end=29, label="chunk", confidence=1.0), ... token_count=7, ... ) chunk.token_count 7

Loaders

openbtk.data.clinical_text.loaders.PlainTextLoader

PlainTextLoader(*, note_type: str | None = None, encoding: str = 'utf-8')

Bases: BaseLoader[str, ClinicalTextRecord]

Load every *.txt file in a directory as one record each.

Example

import contextlib import io import tempfile from pathlib import Path with tempfile.TemporaryDirectory() as d: ... _ = (Path(d) / "note1.txt").write_text("Chief Complaint: pain.") ... loader = PlainTextLoader() ... with contextlib.redirect_stdout(io.StringIO()): # logging noise ... records = list(loader.load(d)) records[0].record_id 'note1' records[0].source 'plain_text'

Source code in src/openbtk/data/clinical_text/loaders.py

def __init__(
    self, *, note_type: str | None = None, encoding: str = "utf-8"
) -> None:
    self._note_type = note_type
    self._encoding = encoding

load

load(source: str) -> Iterator[ClinicalTextRecord]

Yield one record per *.txt file in the directory source, in sorted filename order (deterministic, not filesystem-order-dependent).

Raises:

Type	Description
LoaderError	If source is not a directory, or a file cannot be read or decoded.

Source code in src/openbtk/data/clinical_text/loaders.py

def load(self, source: str) -> Iterator[ClinicalTextRecord]:
    """Yield one record per ``*.txt`` file in the directory ``source``,
    in sorted filename order (deterministic, not filesystem-order-dependent).

    Raises:
        LoaderError: If ``source`` is not a directory, or a file cannot
            be read or decoded.
    """
    directory = Path(source)
    if not directory.is_dir():
        raise LoaderError(
            f"Not a directory: {source}",
            context={"modality": "clinical_text", "stage": "load"},
        )
    log.info("loader.start", modality="clinical_text", source=str(directory))
    for path in sorted(directory.glob("*.txt")):
        try:
            text = path.read_text(encoding=self._encoding)
        except (OSError, UnicodeDecodeError) as e:
            raise LoaderError(
                f"Failed to read {path.name}.",
                context={
                    "modality": "clinical_text",
                    "stage": "load",
                    "filename": path.name,
                },
            ) from e
        yield ClinicalTextRecord(
            record_id=path.stem,
            source="plain_text",
            text=text,
            note_type=self._note_type,
        )

openbtk.data.clinical_text.loaders.JSONLLoader

Bases: BaseLoader[str, ClinicalTextRecord]

Load records from a JSON Lines file, one ClinicalTextRecord per line.

Each line must be a JSON object with at least record_id and text; any other ClinicalTextRecord field may be present. A line with no source key defaults to "jsonl".

Example

import contextlib import io import tempfile from pathlib import Path with tempfile.NamedTemporaryFile( ... mode="w", suffix=".jsonl", delete=False ... ) as f: ... _ = f.write('{"record_id": "n1", "text": "chest pain"}\n') ... path = f.name with contextlib.redirect_stdout(io.StringIO()): # logging noise ... records = list(JSONLLoader().load(path)) records[0].text 'chest pain' Path(path).unlink()

load

load(source: str) -> Iterator[ClinicalTextRecord]

LoaderError: If source cannot be opened, or a line is not valid JSON or does not satisfy ClinicalTextRecord.

Source code in src/openbtk/data/clinical_text/loaders.py

def load(self, source: str) -> Iterator[ClinicalTextRecord]:
    """Raises:
    LoaderError: If ``source`` cannot be opened, or a line is not
        valid JSON or does not satisfy ``ClinicalTextRecord``.
    """
    path = Path(source)
    log.info("loader.start", modality="clinical_text", source=str(path))
    try:
        handle = path.open(encoding="utf-8")
    except OSError as e:
        raise LoaderError(
            f"Could not open {source}.",
            context={"modality": "clinical_text", "stage": "load"},
        ) from e
    try:
        for lineno, line in enumerate(handle, start=1):
            stripped = line.strip()
            if not stripped:
                continue
            try:
                raw: dict[str, Any] = json.loads(stripped)
            except json.JSONDecodeError as e:
                raise LoaderError(
                    f"Invalid JSON on line {lineno} of {path.name}.",
                    context={
                        "modality": "clinical_text",
                        "stage": "load",
                        "filename": path.name,
                        "line": lineno,
                    },
                ) from e
            raw.setdefault("source", "jsonl")
            try:
                yield ClinicalTextRecord.model_validate(raw)
            except Exception as e:
                raise LoaderError(
                    f"Line {lineno} of {path.name} is not a valid "
                    "ClinicalTextRecord.",
                    context={
                        "modality": "clinical_text",
                        "stage": "load",
                        "filename": path.name,
                        "line": lineno,
                    },
                ) from e
    finally:
        handle.close()

openbtk.data.clinical_text.loaders.MIMICNotesLoader

MIMICNotesLoader(*, chunk_size: int = 1000)

Bases: BaseLoader[str, ClinicalTextRecord]

Load a MIMIC-III/IV NOTEEVENTS-shaped CSV, chunked via pandas so peak memory stays O(chunk_size), not O(corpus) (ADR-0004).

Expects (a subset of) MIMIC's real column names: ROW_ID, TEXT, and optionally SUBJECT_ID, HADM_ID, CATEGORY, CHARTDATE. MIMIC-IV is itself a de-identified, IRB-released research dataset -- SUBJECT_ID is already a study-assigned pseudonymous identifier, not real PHI, so it is passed through as patient_ref unchanged.

Source code in src/openbtk/data/clinical_text/loaders.py

def __init__(self, *, chunk_size: int = 1000) -> None:
    self._chunk_size = chunk_size

load

load(source: str) -> Iterator[ClinicalTextRecord]

LoaderError: If source cannot be read, or lacks the required ROW_ID/TEXT columns.

Source code in src/openbtk/data/clinical_text/loaders.py

def load(self, source: str) -> Iterator[ClinicalTextRecord]:
    """Raises:
    LoaderError: If ``source`` cannot be read, or lacks the required
        ``ROW_ID``/``TEXT`` columns.
    """
    pandas = require("pandas", extra="text")
    log.info("loader.start", modality="clinical_text", source=source)
    try:
        reader = pandas.read_csv(source, chunksize=self._chunk_size)
    except FileNotFoundError as e:
        raise LoaderError(
            f"Could not read {source}.",
            context={"modality": "clinical_text", "stage": "load"},
        ) from e
    # pandas' chunked TextFileReader holds an open file handle;
    # generator scope alone does not guarantee it closes (a caller who
    # never exhausts load() would otherwise leak it until GC, which
    # surfaces as a real "Exception ignored" warning -- caught by
    # actually running these tests, not assumed away).
    try:
        for chunk in reader:
            yield from self._records_from_chunk(chunk, source)
    finally:
        reader.close()

Preprocessing

openbtk.data.clinical_text.preprocessing.SectionSegmenter

SectionSegmenter(*, backend: Literal['rule', 'medspacy'] = 'rule')

Bases: BasePreprocessor[ClinicalTextRecord]

Populate record.sections from record.text. See module docstring for the two backends.

Example

record = ClinicalTextRecord( ... record_id="n1", ... source="synthea", ... text="Chief Complaint:\nchest pain\nPlan:\nadmit", ... ) result = SectionSegmenter().process(record) sorted(result.sections) ['chief_complaint', 'plan']

Source code in src/openbtk/data/clinical_text/preprocessing.py

def __init__(self, *, backend: Literal["rule", "medspacy"] = "rule") -> None:
    self._backend = backend

requires_model_download class-attribute

requires_model_download: bool = False

Only true for the (non-default) "medspacy" backend -- see _get_medspacy_pipeline. No single class-level flag can express "sometimes"; callers who need to know should check self._backend.

process

process(record: ClinicalTextRecord) -> ClinicalTextRecord

ProcessingError: If the "medspacy" backend's model is not downloaded. The "rule" backend never raises.

Source code in src/openbtk/data/clinical_text/preprocessing.py

def process(self, record: ClinicalTextRecord) -> ClinicalTextRecord:
    """Raises:
    ProcessingError: If the "medspacy" backend's model is not
        downloaded. The "rule" backend never raises.
    """
    if self._backend == "rule":
        sections = _rule_based_sections(record.text)
    else:
        sections = _medspacy_sections(record.text)
    return record.model_copy(update={"sections": sections})

openbtk.data.clinical_text.preprocessing.DeidPreprocessor

DeidPreprocessor(*, mode: DeidMode = DeidMode.REDACT, recall_bias: RecallBias = 'high', recognizers: Sequence[str] = ('rule',))

Bases: BasePreprocessor[ClinicalTextRecord]

Wrap openbtk.deid.DeidEngine as a pipeline preprocessing step: de-identifies record.text, updating deid_status to match.

Registered under the "general" scope, not "clinical_text": the underlying engine has no modality dependency at all (openbtk.deid's own docstring: "cross-modal by design"). Only this ADAPTER currently targets ClinicalTextRecord -- the one modality record shape that exists pre-v0.5 (CLAUDE.md rule 13, the scope gate); a future ehr adapter would be a second, separate registration when that modality actually exists, not a generalisation forced onto this one ahead of a real second caller.

The full DeidReport (identifiers and counts only, never PHI values -- see openbtk.deid.schemas's own module docstring) is preserved under record.metadata["deid_report"] rather than discarded: docs/04_API_DESIGN.md section 5 says a report is "safe to ... attach to a run manifest", and this preprocessing step is the only place that report is ever produced -- a future pipeline stage collecting it into a RunManifest (task 3.7) reads it from there.

Example

import contextlib import io record = ClinicalTextRecord( ... record_id="n1", ... source="synthea", ... text="Patient SSN: 123-45-6789.", ... ) with contextlib.redirect_stdout(io.StringIO()): # logging noise ... result = DeidPreprocessor(mode=DeidMode.REDACT).process(record) result.text 'Patient SSN: [REDACTED].' result.deid_status

Source code in src/openbtk/data/clinical_text/preprocessing.py

def __init__(
    self,
    *,
    mode: DeidMode = DeidMode.REDACT,
    recall_bias: RecallBias = "high",
    recognizers: Sequence[str] = ("rule",),
) -> None:
    self._mode = mode
    self._engine = DeidEngine(
        mode=mode, recall_bias=recall_bias, recognizers=recognizers
    )

process

process(record: ClinicalTextRecord) -> ClinicalTextRecord

De-identify record.text.

Parameters:

Name	Type	Description	Default
record	ClinicalTextRecord	A loaded (optionally section-segmented) record.	required

Returns:

Type	Description
ClinicalTextRecord	A new record with text transformed, deid_status set,
ClinicalTextRecord	and the audit report attached under
ClinicalTextRecord	metadata["deid_report"].

Note

DeidEngine.deidentify requires a patient_id for SURROGATE/DATE_SHIFT consistency. When record.patient_ref is absent, record.record_id is used instead -- a real, disclosed narrowing to per-record (not per-patient) consistency in that case, not a silent one.

Source code in src/openbtk/data/clinical_text/preprocessing.py

def process(self, record: ClinicalTextRecord) -> ClinicalTextRecord:
    """De-identify ``record.text``.

    Args:
        record: A loaded (optionally section-segmented) record.

    Returns:
        A new record with ``text`` transformed, ``deid_status`` set,
        and the audit report attached under
        ``metadata["deid_report"]``.

    Note:
        ``DeidEngine.deidentify`` requires a ``patient_id`` for
        SURROGATE/DATE_SHIFT consistency. When ``record.patient_ref``
        is absent, ``record.record_id`` is used instead -- a real,
        disclosed narrowing to per-record (not per-patient)
        consistency in that case, not a silent one.
    """
    patient_id = (
        record.patient_ref if record.patient_ref is not None else record.record_id
    )
    result = self._engine.deidentify(
        record.text, patient_id=patient_id, document_id=record.record_id
    )
    status = (
        DeidStatus.SURROGATE
        if self._mode == DeidMode.SURROGATE
        else DeidStatus.DEIDENTIFIED
    )
    return record.model_copy(
        update={
            "text": result.text,
            "deid_status": status,
            "metadata": {
                **record.metadata,
                "deid_report": result.report.model_dump(mode="json"),
            },
        }
    )

Chunking

The one place Openbtk writes genuinely new logic, not a wrapped library — generic chunkers destroy clinical meaning by splitting mid-section.

openbtk.data.clinical_text.chunking.SectionAwareChunker

SectionAwareChunker(*, max_tokens: int = 512, overlap_tokens: int = 0, count_tokens: Callable[[str], int] = count_tokens_approximate)

Bases: BaseChunker[ClinicalTextRecord, ClinicalTextChunk]

Section-respecting chunks -- see module docstring for the full contract. Degrades to FixedTokenChunker when record.sections is empty or None.

Example

record = ClinicalTextRecord( ... record_id="n1", ... source="synthea", ... text="Chief Complaint:\nchest pain.\nPlan:\nadmit.", ... sections={ ... "chief_complaint": TextSpan( ... start=17, end=29, label="s", confidence=1.0 ... ), ... "plan": TextSpan(start=35, end=42, label="s", confidence=1.0), ... }, ... ) chunks = list(SectionAwareChunker(max_tokens=50).chunk(record)) [c.section for c in chunks] == ["chief_complaint", "plan"] True

Source code in src/openbtk/data/clinical_text/chunking.py

def __init__(
    self,
    *,
    max_tokens: int = 512,
    overlap_tokens: int = 0,
    count_tokens: Callable[[str], int] = count_tokens_approximate,
) -> None:
    self._max_tokens = max_tokens
    self._overlap_tokens = overlap_tokens
    self._count_tokens = count_tokens

openbtk.data.clinical_text.chunking.FixedTokenChunker

FixedTokenChunker(*, max_tokens: int = 512, overlap_tokens: int = 0, count_tokens: Callable[[str], int] = count_tokens_approximate)

Bases: BaseChunker[ClinicalTextRecord, ClinicalTextChunk]

Fixed-size, token-bounded chunks -- no sentence or section awareness. SectionAwareChunker composes this directly for its own "no sections" degenerate case (spec point 5), so behaviour matches exactly rather than being reimplemented and risking drift.

Example

from openbtk.data.clinical_text.tokenization import ( ... count_tokens_approximate, ... ) record = ClinicalTextRecord( ... record_id="n1", source="synthea", text="one two three four" ... ) chunker = FixedTokenChunker(max_tokens=2) [c.text for c in chunker.chunk(record)] == ["one two ", "three four"] True

Source code in src/openbtk/data/clinical_text/chunking.py

def __init__(
    self,
    *,
    max_tokens: int = 512,
    overlap_tokens: int = 0,
    count_tokens: Callable[[str], int] = count_tokens_approximate,
) -> None:
    self._max_tokens = max_tokens
    self._overlap_tokens = overlap_tokens
    self._count_tokens = count_tokens

Tokenization

openbtk.data.clinical_text.tokenization.count_tokens_approximate

count_tokens_approximate(text: str) -> int

Whitespace-based approximation: number of non-whitespace runs.

A documented undercount relative to a real subword tokenizer (roughly 30% on clinical text, per docs/05_DATA_MODALITY_SPEC.md section 1.3) -- exists as the zero-dependency fallback, never a silent claim of exactness.

Source code in src/openbtk/data/clinical_text/tokenization.py

def count_tokens_approximate(text: str) -> int:
    """Whitespace-based approximation: number of non-whitespace runs.

    A documented undercount relative to a real subword tokenizer (roughly
    30% on clinical text, per docs/05_DATA_MODALITY_SPEC.md section 1.3)
    -- exists as the zero-dependency fallback, never a silent claim of
    exactness.
    """
    return len(_WHITESPACE_RE.findall(text))

openbtk.data.clinical_text.tokenization.count_tokens_exact

count_tokens_exact(text: str, *, model_name: str = DEFAULT_EXACT_MODEL, revision: str | None = None) -> int

Exact subword count using model_name's real tokenizer.

Parameters:

Name	Type	Description	Default
text	str	The text to count. Never logged.	required
model_name	str	A HuggingFace model id. Its tokenizer is downloaded and cached on first use for that id.	DEFAULT_EXACT_MODEL
revision	str | None	The Hub commit to load. The default model is always pinned; for any other model, omitting this follows the repo's moving default branch and logs a tokenizer.unpinned warning.	None

Raises:

Type	Description
MissingDependencyError	If the text extra is not installed.

Source code in src/openbtk/data/clinical_text/tokenization.py

def count_tokens_exact(
    text: str,
    *,
    model_name: str = DEFAULT_EXACT_MODEL,
    revision: str | None = None,
) -> int:
    """Exact subword count using ``model_name``'s real tokenizer.

    Args:
        text: The text to count. Never logged.
        model_name: A HuggingFace model id. Its tokenizer is downloaded
            and cached on first use for that id.
        revision: The Hub commit to load. The default model is always pinned;
            for any other model, omitting this follows the repo's moving
            default branch and logs a ``tokenizer.unpinned`` warning.

    Raises:
        MissingDependencyError: If the ``text`` extra is not installed.
    """
    if not text:
        return 0
    tokenizer = _get_tokenizer(model_name, revision)
    return len(tokenizer.encode(text, add_special_tokens=False))